class UdafAggregateFunctionFactory extends AggregateFunctionFactory {

  @Override
  public synchronized KsqlAggregateFunction<?, ?, ?> createAggregateFunction(
      final List<SqlType> argTypeList,
      final AggregateFunctionInitArguments initArgs
  ) {
    final List<SqlType> allParams = buildAllParams(argTypeList, initArgs);
    final UdafFactoryInvoker creator = udfIndex.getFunction(allParams);
    if (creator == null) {
      throw new KsqlException("There is no aggregate function with name='" + getName()
          + "' that has arguments of type="
          + allParams.stream()
          .map(SqlType::baseType)
          .map(Objects::toString)
          .collect(Collectors.joining(",")));
    }
    return creator.createFunction(initArgs);
  }

    UdafAggregateFunctionFactory(
      final UdfMetadata metadata,
      final List<UdafFactoryInvoker> factoryList
  );
  @VisibleForTesting  UdafAggregateFunctionFactory(
      final UdfMetadata metadata,
      final UdfIndex<UdafFactoryInvoker> index
  );

}