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 shouldAllowClassesWithSameFQCNInDifferentUDFJars() throws Exception {
    final File pluginDir = tempFolder.newFolder();
    Files.copy(Paths.get("src/test/resources/udf-example.jar"),
        new File(pluginDir, "udf-example.jar").toPath());
    Files.copy(Paths.get("src/test/resources/udf-isolated.jar"),
        new File(pluginDir, "udf-isolated.jar").toPath());

    final MutableFunctionRegistry functionRegistry = new InternalFunctionRegistry();
    final UserFunctionLoader udfLoader = new UserFunctionLoader(
        functionRegistry,
        pluginDir,
        PARENT_CLASS_LOADER,
        value -> false,
        Optional.empty(),
        true);

    udfLoader.load();

    final UdfFactory multiply = functionRegistry.getUdfFactory(FunctionName.of("multiply"));
    final UdfFactory multiply2 = functionRegistry.getUdfFactory(FunctionName.of("multiply2"));

    final Kudf multiplyUdf = multiply.getFunction(Arrays.asList(SqlArgument.of(
        SqlTypes.INTEGER),
        SqlArgument.of(SqlTypes.INTEGER))
    ).newInstance(ksqlConfig);

    final Kudf multiply2Udf = multiply2
        .getFunction(Arrays.asList(SqlArgument.of(SqlTypes.INTEGER), SqlArgument.of(SqlTypes.INTEGER)))
        .newInstance(ksqlConfig);

    assertThat(multiplyUdf.evaluate(2, 2), equalTo(4L));
    assertThat(multiply2Udf.evaluate(2, 2), equalTo(5L));
  }