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 shouldLoadBadFunctionButNotLetItExit2() {
    // Given:
    final List<SqlArgument> argList = singletonList(SqlArgument.of(SqlTypes.STRING));
    // We do need to set up the ExtensionSecurityManager for our test.
    // This is controlled by a feature flag and in this test, we just directly enable it.
    SecurityManager manager = System.getSecurityManager();
    System.setSecurityManager(ExtensionSecurityManager.INSTANCE);

    final UdfFactory function = FUNC_REG.getUdfFactory(FunctionName.of("bad_test_udf"));
    assertThat(function, not(nullValue()));

    KsqlScalarFunction ksqlScalarFunction = function.getFunction(argList);
    final Kudf badFunction = ksqlScalarFunction.newInstance(ksqlConfig);

    // Given:
    final Exception e2 = assertThrows(
        KsqlFunctionException.class,
        () -> badFunction.evaluate("foo")
    );

    // Then:
    assertThat(e2.getMessage(), containsString(
        "Failed to invoke function public org.apache.kafka.connect.data.Struct "
            + "io.confluent.ksql.function.udf.BadTestUdf.returnList(java.lang.String)"));
    System.setSecurityManager(manager);
    assertEquals(System.getSecurityManager(), manager);
  }