public static TaskFactory getTaskFactory(String taskClassName) {
    Preconditions.checkArgument(StringUtils.isNotBlank(taskClassName), "task.class cannot be empty");
    log.info("Got task class name: {}", taskClassName);

    boolean isAsyncTaskClass;
    try {
      isAsyncTaskClass = AsyncStreamTask.class.isAssignableFrom(Class.forName(taskClassName));
    } catch (Throwable t) {
      throw new ConfigException(String.format("Invalid configuration for AsyncStreamTask class: %s", taskClassName), t);
    }

    if (isAsyncTaskClass) {
      return (AsyncStreamTaskFactory) () -> {
        try {
          return (AsyncStreamTask) Class.forName(taskClassName).newInstance();
        } catch (Throwable t) {
          log.error("Error loading AsyncStreamTask class: {}. error: {}", taskClassName, t);
          throw new SamzaException(String.format("Error loading AsyncStreamTask class: %s", taskClassName), t);
        }
      };
    }

    return (StreamTaskFactory) () -> {
      try {
        return (StreamTask) Class.forName(taskClassName).newInstance();
      } catch (Throwable t) {
        log.error("Error loading StreamTask class: {}. error: {}", taskClassName, t);
        throw new SamzaException(String.format("Error loading StreamTask class: %s", taskClassName), t);
      }
    };
  }
Creates a {@link TaskFactory} based on the configuration.
<p>
This should only be used to create {@link TaskFactory} defined in task.class

@param taskClassName  the task class name for this job
@return  a {@link TaskFactory} object, either a instance of {@link StreamTaskFactory} or {@link AsyncStreamTaskFactory}
samza-core/src/main/java/org/apache/samza/task/TaskFactoryUtil.java