class UdfLoader {

  private Function<KsqlConfig, Kudf> getUdfFactory(
      final Method method,
      final UdfDescription udfDescriptionAnnotation,
      final String functionName,
      final FunctionInvoker invoker,
      final String sensorName
  ) {
    return ksqlConfig -> {
      final Object actualUdf = FunctionLoaderUtils.instantiateFunctionInstance(
          method.getDeclaringClass(), udfDescriptionAnnotation.name());
      if (actualUdf instanceof Configurable) {
        ((Configurable) actualUdf)
            .configure(ksqlConfig.getKsqlFunctionsConfigProps(functionName));
      }
      final PluggableUdf theUdf = new PluggableUdf(invoker, actualUdf);
      return metrics.<Kudf>map(m -> new UdfMetricProducer(
          m.getSensor(sensorName),
          theUdf,
          Time.SYSTEM
      )).orElse(theUdf);
    };
  }

  public  UdfLoader(
      final MutableFunctionRegistry functionRegistry,
      final Optional<Metrics> metrics,
      final SqlTypeParser typeParser,
      final boolean throwExceptionOnLoadFailure
  );

}