public void loadUdtfFromClass(
      final Class<?> theClass,
      final String path
  ) {
    final UdtfDescription udtfDescriptionAnnotation = theClass.getAnnotation(UdtfDescription.class);
    if (udtfDescriptionAnnotation == null) {
      throw new KsqlException(String.format("Cannot load class %s. Classes containing UDTFs must"
          + "be annotated with @UdtfDescription.", theClass.getName()));
    }
    final String functionName = udtfDescriptionAnnotation.name();
    final String sensorName = "ksql-udtf-" + functionName;

    FunctionMetrics.initInvocationSensor(metrics, sensorName, "ksql-udtf", functionName + " udtf");

    final UdfMetadata metadata = new UdfMetadata(
        udtfDescriptionAnnotation.name(),
        udtfDescriptionAnnotation.description(),
        udtfDescriptionAnnotation.author(),
        udtfDescriptionAnnotation.version(),
        udtfDescriptionAnnotation.category(),
        path
    );

    final TableFunctionFactory factory = new TableFunctionFactory(metadata);

    for (final Method method : theClass.getMethods()) {
      if (method.getAnnotation(Udtf.class) != null) {
        final Udtf annotation = method.getAnnotation(Udtf.class);
        try {
          if (method.getReturnType() != List.class) {
            throw new KsqlException(String
                .format("UDTF functions must return a List. Class %s Method %s",
                    theClass.getName(), method.getName()
                ));
          }
          final Type ret = method.getGenericReturnType();
          if (!(ret instanceof ParameterizedType)) {
            throw new KsqlException(String
                .format(
                    "UDTF functions must return a parameterized List. Class %s Method %s",
                    theClass.getName(), method.getName()
                ));
          }
          final Type typeArg = ((ParameterizedType) ret).getActualTypeArguments()[0];
          final ParamType returnType = FunctionLoaderUtils
              .getReturnType(method, typeArg, annotation.schema(), typeParser);
          final List<ParameterInfo> parameters = FunctionLoaderUtils
              .createParameters(method, functionName, typeParser);
          final KsqlTableFunction tableFunction =
              createTableFunction(method, FunctionName.of(functionName), returnType,
                  parameters,
                  annotation.description(),
                  annotation
              );
          factory.addFunction(tableFunction);
        } catch (final KsqlException e) {
          if (throwExceptionOnLoadFailure) {
            throw e;
          } else {
            LOGGER.warn(
                "Failed to add UDTF to the MetaStore. name={} method={}",
                udtfDescriptionAnnotation.name(),
                method,
                e
            );
          }
        }
      }
    }

    functionRegistry.addTableFunctionFactory(factory);
  }
@Test
  public void shouldNotLoadUdtfWithRawListReturn() {
    // Given:
    final MutableFunctionRegistry functionRegistry = new InternalFunctionRegistry();
    final SqlTypeParser typeParser = create(EMPTY);
    final UdtfLoader udtfLoader = new UdtfLoader(
        functionRegistry, empty(), typeParser, true
    );

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> udtfLoader.loadUdtfFromClass(RawListReturn.class, INTERNAL_PATH)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "UDTF functions must return a parameterized List. Class io.confluent.ksql.function.UdtfLoaderTest$RawListReturn Method badReturn"));
  }