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) {
        ExtensionSecurityManager.INSTANCE.pushInUdf();
        try {
          ((Configurable) actualUdf)
              .configure(ksqlConfig.getKsqlFunctionsConfigProps(functionName));
        } finally {
          ExtensionSecurityManager.INSTANCE.popOutUdf();
        }
      }
      final PluggableUdf theUdf = new PluggableUdf(invoker, actualUdf);
      return metrics.<Kudf>map(m -> new UdfMetricProducer(
          m.getSensor(sensorName),
          theUdf,
          Time.SYSTEM
      )).orElse(theUdf);
    };
  }
@Test
  public void shouldUseConfigForExtDir() {
    final InternalFunctionRegistry functionRegistry = new InternalFunctionRegistry();
    // The tostring function is in the udf-example.jar that is found in src/test/resources
    final ImmutableMap<Object, Object> configMap
        = ImmutableMap.builder().put(KsqlConfig.KSQL_EXT_DIR, "src/test/resources")
        .put(KsqlConfig.KSQL_UDF_SECURITY_MANAGER_ENABLED, false)
        .build();
    final KsqlConfig config
        = new KsqlConfig(configMap);
    UserFunctionLoader.newInstance(
        config,
        functionRegistry,
        "",
        new Metrics()
    ).load();
    // will throw if it doesn't exist
    functionRegistry.getUdfFactory(FunctionName.of("tostring"));
  }