Class<SamzaApplication> appClass = (Class<SamzaApplication>) Class.forName(appConfig.getAppClass());
if (StreamApplication.class.isAssignableFrom(appClass) || TaskApplication.class.isAssignableFrom(appClass)) {
  return appClass.newInstance();
}

public class PageViewCounterExample implements StreamApplication {

  public static void main(String[] args) {
    CommandLine cmdLine = new CommandLine();
    Config config = cmdLine.loadConfig(cmdLine.parser().parse(args));
    ApplicationRunner runner = ApplicationRunners.getApplicationRunner(ApplicationClassUtils.fromConfig(config), config);
    runner.run();
    runner.waitForFinish();
  }

  @Override
  public void describe(StreamApplicationDescriptor appDesc) {
      // Detailed examples on how to create InputDescriptor and OutputDescriptor are in SEP-14
      InputDescriptor pveStreamDescriptor = getInputDescriptor("pageViewEventStream", new JsonSerdeV2<>(PageViewEvent.class));
      OutputDescriptor outputDescriptor = getOutputDescriptor("pageViewEventPerMemberStream",
              KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewCount.class));
      MessageStream<PageViewEvent> pageViewEvents = null;
      pageViewEvents = appDesc.getInputStream(pveStreamDescriptor);
      OutputStream<KV<String, PageViewCount>> pageViewEventPerMemberStream =
          appDesc.getOutputStream(outputDescriptor);

      SupplierFunction<Integer> initialValue = () -> 0;
      FoldLeftFunction<PageViewEvent, Integer> foldLeftFn = (m, c) -> c + 1;
      pageViewEvents
          .window(Windows.keyedTumblingWindow(m -> m.memberId, Duration.ofSeconds(10), initialValue, foldLeftFn,
              null, null)
              .setEarlyTrigger(Triggers.repeat(Triggers.count(5)))
              .setAccumulationMode(AccumulationMode.DISCARDING), "tumblingWindow")
          .map(windowPane -> KV.of(windowPane.getKey().getKey(), new PageViewCount(windowPane)))
          .sendTo(pageViewEventPerMemberStream);
  }
}

public class TaskApplicationExample implements TaskApplication {

  public static void main(String[] args) {
    CommandLine cmdLine = new CommandLine();
    Config config = cmdLine.loadConfig(cmdLine.parser().parse(args));
    ApplicationRunner runner = ApplicationRuntimes.getApplicationRunner(new TaskApplicationExample(), config);
    runner.run();
    runner.waitForFinish();
  }

  @Override
  public void describe(TaskAppDescriptor appDesc) {
    // Detailed examples on how to create InputDescriptor and OutputDescriptor are in SEP-14
    InputDescriptor pveStreamDescriptor = getInputDescriptor("pageViewEventStream", new JsonSerdeV2<>(PageViewEvent.class));
    OutputDescriptor outputDescriptor = getOutputDescriptor("pageViewEventPerMemberStream",
        KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageViewCount.class));
    appDesc.addInputStream(pveStreamDescriptor);
    appDesc.addOutputStream(outputDescriptor);
    TableDescriptor td = new RocksDbTableDescriptor("mytable");
    appDesc.addTable(td);
    // create the task factory based on configuration
    appDesc.setTaskFactory(TaskFactoryUtil.createTaskFactory(appDesc.getConfig()));
  }

}

public interface SamzaApplication<S extends ApplicationDescriptor> { 
  void describe(S appDesc); 
} 

public interface StreamApplication extends SamzaApplication<StreamApplicationDescriptor> { 
}

public interface TaskApplication extends SamzaApplication<TaskApplicationDescriptor> { 
}

public interface ProcessorLifecycleListenerFactory extends Serializable {
  /**
   * Create an instance of {@link ProcessorLifecycleListener} for the StreamProcessor
   *
   * @param pContext the context of the corresponding StreamProcessor
   * @param config the configuration of the corresponding StreamProcessor
   * @return the {@link ProcessorLifecycleListener} callback object for the StreamProcessor
   */
  ProcessorLifecycleListener createInstance(ProcessorContext pContext, Config config);
}


/**
 * The context for a StreamProcessor. This is a stub class, just include the method to identify the current StreamProcessor.
 *
 */
public interface ProcessorContext extends Serializable {
  String getProcessorId();
}

public interface ProcessorLifecycleListener {
  /**
   * User defined initialization before a StreamProcessor is started
   */
  default void beforeStart() {}

  /**
   * User defined callback after a StreamProcessor is started
   *
   */
  default void afterStart() {}

  /**
   * User defined callback after a StreamProcessor is stopped successfully
   */
  default void afterStop() {}

  /**
   * User defined callback after a StreamProcessor is stopped with failure
   *
   * @param t the error causing the stop of the StreamProcessor.
   */
  default void afterFailure(Throwable t) {}
}

public interface ApplicationDescriptor<S extends ApplicationDescriptor> { 
  /**
   * Get the {@link Config} of the application
   * @return config of the application
   */
  Config getConfig();

  /**
   * Sets the {@link ContextManager} for this application.
   * <p>
   * Setting the {@link ContextManager} is optional. The provided {@link ContextManager} can be used to build the shared
   * context between the operator functions within a task instance
   *
   * TODO: this should be replaced by the shared context factory when SAMZA-1714 is fixed.

   * @param contextManager the {@link ContextManager} to use for the application
   * @return type {@code S} of {@link ApplicationDescriptor} with {@code contextManager} set as its {@link ContextManager}
   */
  S withContextManager(ContextManager contextManager);

  /**
   * Sets the {@link ProcessorLifecycleListenerFactory} for this application.
   *
   * <p>Setting a {@link ProcessorLifecycleListenerFactory} is optional to a user application. It allows users to
   * plug in optional code to be invoked in different stages before/after the main processing logic is started/stopped in
   * the application.
   *
   * @param listenerFactory the user implemented {@link ProcessorLifecycleListenerFactory} that creates lifecycle listener
   *                        with callback methods before and after the start/stop of each StreamProcessor in the application
   * @return type {@code S} of {@link ApplicationDescriptor} with {@code listenerFactory} set as its {@link ProcessorLifecycleListenerFactory}
   */
  S withProcessorLifecycleListenerFactory(ProcessorLifecycleListenerFactory listenerFactory);

  /**
   * Sets a set of customized {@link MetricsReporterFactory}s in the application
   *
   * @param reporterFactories the map of customized {@link MetricsReporterFactory}s to be used
   * @return type {@code S} of {@link ApplicationDescriptor} with {@code reporterFactories}
   */
  S withMetricsReporterFactories(Map<String, MetricsReporterFactory> reporterFactories);
} 

public interface StreamApplicationDescriptor extends ApplicationDescriptor<StreamApplicationDescriptor> { 
  /**
   * Sets the default SystemDescriptor to use for intermediate streams. This is equivalent to setting
   * {@code job.default.system} and its properties in configuration.
   * <p>
   * If the default system descriptor is set, it must be set <b>before</b> creating any input/output/intermediate streams.
   * <p>
   * If an input/output stream is created with a stream-level Serde, they will be used, else the serde specified
   * for the {@code job.default.system} in configuration will be used.
   * <p>
   * Providing an incompatible message type for the intermediate streams that use the default serde will result in
   * {@link ClassCastException}s at runtime.
   *
   * @param defaultSystemDescriptor the default system descriptor to use
   * @return type {@code S} of {@link ApplicationDescriptor} with {@code defaultSystemDescriptor} set as its default system
   */
  StreamApplicationDescriptor withDefaultSystem(SystemDescriptor<?> defaultSystemDescriptor);

  /**
   * Gets the input {@link MessageStream} corresponding to the {@code inputDescriptor}.
   * <p>
   * A {@code MessageStream<KV<K, V>}, obtained by calling this method with a descriptor with a {@code KVSerde<K, V>},
   * can receive messages of type {@code KV<K, V>}. An input {@code MessageStream<M>}, obtained using a descriptor with
   * any other {@code Serde<M>}, can receive messages of type M - the key in the incoming message is ignored.
   * <p>
   * A {@code KVSerde<NoOpSerde, NoOpSerde>} or {@code NoOpSerde} may be used for the descriptor if the
   * {@code SystemConsumer} deserializes the incoming messages itself, and no further deserialization is required from  
   * the framework.
   * <p>
   * Multiple invocations of this method with the same {@code inputDescriptor} will throw an
   * {@link IllegalStateException}.
   *
   * @param inputDescriptor the descriptor for the stream
   * @param <M> the type of messages in the input {@link MessageStream}
   * @return the input {@link MessageStream}
   * @throws IllegalStateException when invoked multiple times with the same {@code inputDescriptor}
   */
  <M> MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor);

  /**
   * Gets the {@link OutputStream} corresponding to the {@code outputDescriptor}.
   * <p>
   * An {@code OutputStream<KV<K, V>>}, obtained by calling this method with a descriptor with a {@code KVSerde<K, V>},
   * can send messages of type {@code KV<K, V>}. An {@code OutputStream<M>}, obtained using a descriptor with any
   * other {@code Serde<M>}, can send messages of type M without a key.
   * <p>
   * A {@code KVSerde<NoOpSerde, NoOpSerde>} or {@code NoOpSerde} may be used for the descriptor if the
   * {@code SystemProducer} serializes the outgoing messages itself, and no prior serialization is required from
   * the framework.
   * <p>
   * When sending messages to an {@code OutputStream<KV<K, V>>}, messages are partitioned using their serialized key.
   * When sending messages to any other {@code OutputStream<M>}, messages are partitioned using a null partition key.
   * <p>
   * Multiple invocations of this method with the same {@code outputDescriptor} will throw an
   * {@link IllegalStateException}.
   *
   * @param outputDescriptor the descriptor for the stream
   * @param <M> the type of messages in the {@link OutputStream}
   * @return the {@link OutputStream}
   * @throws IllegalStateException when invoked multiple times with the same {@code outputDescriptor}
   */
  <M> OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor);

  /**
   * Gets the {@link Table} corresponding to the {@link TableDescriptor}.
   * <p>
   * Multiple invocations of this method with the same {@link TableDescriptor} will throw an
   * {@link IllegalStateException}.
   *
   * @param tableDescriptor the {@link TableDescriptor}
   * @param <K> the type of the key
   * @param <V> the type of the value
   * @return the {@link Table} corresponding to the {@code tableDescriptor}
   * @throws IllegalStateException when invoked multiple times with the same {@link TableDescriptor}
   */
  <K, V> Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor);
} 

public interface TaskApplicationDescriptor extends ApplicationDescriptor<TaskApplicationDescriptor> { 
  /**
   * Sets the {@link TaskFactory} for the user application. The {@link TaskFactory#createInstance()} creates task instance
   * that implements the main processing logic of the user application.
   *
   * @param factory the {@link TaskFactory} including the low-level task processing logic. The only allowed task factory
   *                classes are {@link org.apache.samza.task.StreamTaskFactory} and {@link org.apache.samza.task.AsyncStreamTaskFactory}.
   */
  void setTaskFactory(TaskFactory factory);

  /**
   * Adds the input stream to the application.
   *
   * @param isd the {@link InputDescriptor}
   */
  void addInputStream(InputDescriptor isd);

  /**
   * Adds the output stream to the application.
   *
   * @param osd the {@link OutputDescriptor} of the output stream
   */
  void addOutputStream(OutputDescriptor osd);

  /**
   * Adds the {@link TableDescriptor} used in the application
   *
   * @param table {@link TableDescriptor}
   */
  void addTable(TableDescriptor table);
} 

public interface ApplicationRunner {
  /**
   * Start a runtime instance of the application
   */
  void run();

  /**
   * Stop a runtime instance of the application
   */
  void kill();

  /**
   * Get the {@link ApplicationStatus} of a runtime instance of the application
   * @return the runtime status of the application
   */
  ApplicationStatus status();

  /**
   * Wait the runtime instance of the application to complete.
   * This method will block until the application completes.
   */
  void waitForFinish();

  /**
   * Wait the runtime instance of the application to complete with a {@code timeout}
   *
   * @param timeout the time to block to wait for the application to complete
   * @return true if the application completes within timeout; false otherwise
   */
  boolean waitForFinish(Duration timeout);
}

public class ApplicationRunners {

  private ApplicationRunners() {

  }

  /**
   * Get the {@link ApplicationRunner} that runs the {@code userApp}
   *
   * @param userApp the user application object
   * @param config the configuration for this application
   * @return the {@link ApplicationRunner} object that will run the {@code userApp}
   */
  public static final ApplicationRunner getApplicationRunner(SamzaApplication userApp, Config config) {
    String appRunnerClassName = getAppRunnerClass(config);
    try {
      Class<?> runnerClass = Class.forName(appRunnerClassName);
      if (!ApplicationRunner.class.isAssignableFrom(runnerClass)) {
        throw new ConfigException(
            String.format("Class %s does not extend ApplicationRunner properly", appRunnerClassName));
      }
      Constructor<?> constructor = runnerClass.getConstructor(SamzaApplication.class, Config.class); // *sigh*
      return (ApplicationRunner) constructor.newInstance(userApp, config);
    } catch (ConfigException ce) {
      // this is thrown due to invalid app.runner.class configuration
      throw ce;
    } catch (Exception e) {
      // other types of exception during class loading and construction of new instance
      throw new ConfigException(String.format("Could not load ApplicationRunner class %s", appRunnerClassName), e);
    }
  }

  private static String getAppRunnerClass(Config config) {
    return config.getOrDefault(APP_RUNNER_CFG, DEFAULT_APP_RUNNER);
  }
}