private KsqlScalarFunction createFunction(
      final Class theClass,
      final UdfDescription udfDescriptionAnnotation,
      final Udf udfAnnotation,
      final Method method,
      final String path,
      final String sensorName,
      final Class<? extends Kudf> udfClass
  ) {
    // sanity check
    FunctionLoaderUtils
        .instantiateFunctionInstance(method.getDeclaringClass(), udfDescriptionAnnotation.name());
    final FunctionInvoker invoker = FunctionLoaderUtils.createFunctionInvoker(method);
    final String functionName = udfDescriptionAnnotation.name();

    LOGGER.info("Adding function " + functionName + " for method " + method);

    final List<ParameterInfo> parameters = FunctionLoaderUtils
        .createParameters(method, functionName, typeParser);

    final ParamType javaReturnSchema = FunctionLoaderUtils
        .getReturnType(method, udfAnnotation.schema(), typeParser);

    final SchemaProvider schemaProviderFunction = FunctionLoaderUtils
        .handleUdfReturnSchema(
            theClass,
            javaReturnSchema,
            udfAnnotation.schema(),
            typeParser,
            udfAnnotation.schemaProvider(),
            udfDescriptionAnnotation.name(),
            method.isVarArgs()
        );

    return KsqlScalarFunction.create(
        schemaProviderFunction,
        javaReturnSchema,
        parameters,
        FunctionName.of(functionName.toUpperCase()),
        udfClass,
        getUdfFactory(method, udfDescriptionAnnotation, functionName, invoker, sensorName),
        udfAnnotation.description(),
        path,
        method.isVarArgs()
    );
  }
@Test
  @SuppressWarnings("rawtypes")
  public void shouldImplementTableAggregateFunctionWhenTableUdafClass() throws Exception {
    final UdafFactoryInvoker creator
        = createUdafLoader().createUdafFactoryInvoker(
        TestUdaf.class.getMethod("createSumLong"),
        FunctionName.of("test-udf"),
        "desc",
        new String[]{""},
        "",
        "");
    final KsqlAggregateFunction function = creator
        .createFunction(AggregateFunctionInitArguments.EMPTY_ARGS, Collections.emptyList());
    assertThat(function, instanceOf(TableAggregationFunction.class));
  }