SAMZA-2012: Add API for wiring an external context through to application processing code

This PR also refactors TestSamzaSqlRemoteTable to be in samza-test instead of samza-sql, since it seems to actually be an integration test. It is useful to move that test in this PR so that tests that may need an external context can be consolidated.

Author: Cameron Lee <calee@linkedin.com>

Reviewers: Prateek Maheshwari <pmaheshwari@apache.org>, Shanthoosh Venkatraman <svenkatr@linkedin.com>

Closes apache#829 from cameronlee314/external_context

Author: cameronlee314

samza-api/src/main/java/org/apache/samza/context/ApplicationContainerContextFactory.java

@@ -19,6 +19,7 @@
 package org.apache.samza.context;
 
 import java.io.Serializable;
+import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.application.SamzaApplication;
 import org.apache.samza.application.descriptors.ApplicationDescriptor;
 
@@ -35,14 +36,41 @@
  *
  * @param <T> concrete type of {@link ApplicationContainerContext} created by this factory
  */
+@InterfaceStability.Evolving
 public interface ApplicationContainerContextFactory<T extends ApplicationContainerContext> extends Serializable {
-
   /**
    * Creates an instance of the application-defined {@link ApplicationContainerContext}.
+   * <p>
+   * Applications should implement this to provide a context for container initialization.
    *
+   * @param externalContext external context provided for the application; null if it was not provided
    * @param jobContext framework-provided job context
    * @param containerContext framework-provided container context
    * @return a new instance of the application-defined {@link ApplicationContainerContext}
    */
-  T create(JobContext jobContext, ContainerContext containerContext);
+  default T create(ExternalContext externalContext, JobContext jobContext, ContainerContext containerContext) {
+    return create(jobContext, containerContext);
+  }
+
+  /**
+   * New implementations should not implement this directly. Implement
+   * {@link #create(ExternalContext, JobContext, ContainerContext)} instead.
+   * <p>
+   * This is the same as {@link #create(ExternalContext, JobContext, ContainerContext)}, except it does not provide
+   * access to external context.
+   * <p>
+   * This is being left here for backwards compatibility.
+   *
+   * @param jobContext framework-provided job context
+   * @param containerContext framework-provided container context
+   * @return a new instance of the application-defined {@link ApplicationContainerContext}
+   *
+   * Deprecated: Applications should implement {@link #create(ExternalContext, JobContext, ContainerContext)} directly.
+   * This is being left here for backwards compatibility.
+   */
+  @Deprecated
+  default T create(JobContext jobContext, ContainerContext containerContext) {
+    // adding this here so that new apps do not need to implement this
+    throw new UnsupportedOperationException("Please implement a version of create for the factory implementation.");
+  }
 }

samza-api/src/main/java/org/apache/samza/context/ApplicationTaskContextFactory.java

@@ -19,6 +19,7 @@
 package org.apache.samza.context;
 
 import java.io.Serializable;
+import org.apache.samza.annotation.InterfaceStability;
 import org.apache.samza.application.SamzaApplication;
 import org.apache.samza.application.descriptors.ApplicationDescriptor;
 
@@ -35,17 +36,49 @@
  *
  * @param <T> concrete type of {@link ApplicationTaskContext} created by this factory
  */
+@InterfaceStability.Evolving
 public interface ApplicationTaskContextFactory<T extends ApplicationTaskContext> extends Serializable {
-
   /**
    * Creates an instance of the application-defined {@link ApplicationTaskContext}.
+   * <p>
+   * Applications should implement this to provide a context for task initialization.
+   *
+   * @param externalContext external context provided for the application; null if it was not provided
+   * @param jobContext framework-provided job context
+   * @param containerContext framework-provided container context
+   * @param taskContext framework-provided task context
+   * @param applicationContainerContext application-defined container context; null if it was not provided
+   * @return a new instance of the application-defined {@link ApplicationTaskContext}
+   */
+  default T create(ExternalContext externalContext, JobContext jobContext, ContainerContext containerContext,
+      TaskContext taskContext, ApplicationContainerContext applicationContainerContext) {
+    return create(jobContext, containerContext, taskContext, applicationContainerContext);
+  }
+
+  /**
+   * New implementations should not implement this directly. Implement
+   * {@link #create(ExternalContext, JobContext, ContainerContext, TaskContext, ApplicationContainerContext)} instead.
+   * <p>
+   * This is the same as
+   * {@link #create(ExternalContext, JobContext, ContainerContext, TaskContext, ApplicationContainerContext)}, except it
+   * does not provide access to external context.
+   * <p>
+   * This is being left here for backwards compatibility.
    *
    * @param jobContext framework-provided job context
    * @param containerContext framework-provided container context
    * @param taskContext framework-provided task context
-   * @param applicationContainerContext application-defined container context
+   * @param applicationContainerContext application-defined container context; null if it was not provided
    * @return a new instance of the application-defined {@link ApplicationTaskContext}
+   *
+   * Deprecated: Applications should implement
+   * {@link #create(ExternalContext, JobContext, ContainerContext, TaskContext, ApplicationContainerContext)} directly.
+   * This is being left here for backwards compatibility.
    */
-  T create(JobContext jobContext, ContainerContext containerContext, TaskContext taskContext,
-      ApplicationContainerContext applicationContainerContext);
+  @Deprecated
+  default T create(JobContext jobContext, ContainerContext containerContext, TaskContext taskContext,
+      ApplicationContainerContext applicationContainerContext) {
+    // adding this here so that new apps do not need to implement this
+    throw new UnsupportedOperationException("Please implement a version of create for the factory implementation.");
+  }
 }

samza-api/src/main/java/org/apache/samza/context/Context.java

@@ -19,6 +19,8 @@
 package org.apache.samza.context;
 
 import org.apache.samza.application.descriptors.ApplicationDescriptor;
+import org.apache.samza.runtime.ApplicationRunner;
+
 
 /**
  * A holder for all framework and application defined contexts at runtime.
@@ -76,4 +78,13 @@ public interface Context {
    * @throws IllegalStateException if no {@link ApplicationTaskContextFactory} was provided for the application
    */
   ApplicationTaskContext getApplicationTaskContext();
+
+  /**
+   * Gets the {@link ExternalContext} that was created outside of the application.
+   * <p>
+   * Use {@link ApplicationRunner#run(ExternalContext)} to provide this context.
+   *
+   * @return the external context provided for the application
+   */
+  ExternalContext getExternalContext();
 }

samza-api/src/main/java/org/apache/samza/context/ExternalContext.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.context;
+
+/**
+ * An {@link ExternalContext} can be used to pass components created and managed outside of Samza into a Samza
+ * application. This will be made accessible through the {@link Context}.
+ * <p>
+ * This is passed to {@link org.apache.samza.runtime.ApplicationRunner#run(ExternalContext)} and propagated down to the
+ * {@link Context} object provided to tasks.
+ * <p>
+ * {@link ExternalContext} can be used to inject objects that need to be created by other frameworks, such as Spring.
+ * <p>
+ * This is currently just a marker interface for the object passed into Samza.
+ */
+public interface ExternalContext {
+}

samza-api/src/main/java/org/apache/samza/runtime/ApplicationRunner.java

@@ -20,6 +20,7 @@
 
 import java.time.Duration;
 import org.apache.samza.annotation.InterfaceStability;
+import org.apache.samza.context.ExternalContext;
 import org.apache.samza.job.ApplicationStatus;
 
 
@@ -32,12 +33,19 @@
  */
 @InterfaceStability.Evolving
 public interface ApplicationRunner {
+  /**
+   * This is like {@link #run(ExternalContext)}, except it provides a null {@link ExternalContext}.
+   */
+  default void run() {
+    run(null);
+  }
 
   /**
    * Deploy and run the Samza jobs to execute {@link org.apache.samza.application.SamzaApplication}.
    * It is non-blocking so it doesn't wait for the application running.
+   * @param externalContext nullable {@link ExternalContext} to pass through to the application
    */
-  void run();
+  void run(ExternalContext externalContext);
 
   /**
    * Kill the Samza jobs represented by {@link org.apache.samza.application.SamzaApplication}

samza-api/src/test/java/org/apache/samza/runtime/TestApplicationRunners.java

@@ -25,6 +25,7 @@
 import org.apache.samza.application.StreamApplication;
 import org.apache.samza.config.Config;
 import org.apache.samza.config.MapConfig;
+import org.apache.samza.context.ExternalContext;
 import org.apache.samza.job.ApplicationStatus;
 import org.junit.Test;
 
@@ -60,7 +61,7 @@ public MockApplicationRunner(SamzaApplication userApp, Config config) {
     }
 
     @Override
-    public void run() {
+    public void run(ExternalContext externalContext) {
 
     }
 

samza-core/src/main/java/org/apache/samza/context/ContextImpl.java

@@ -30,24 +30,26 @@ public class ContextImpl implements Context {
   private final TaskContext taskContext;
   private final Optional<ApplicationContainerContext> applicationContainerContextOptional;
   private final Optional<ApplicationTaskContext> applicationTaskContextOptional;
+  private final Optional<ExternalContext> externalContextOptional;
 
   /**
    * @param jobContext non-null job context
    * @param containerContext non-null framework container context
    * @param taskContext non-null framework task context
    * @param applicationContainerContextOptional optional application-defined container context
    * @param applicationTaskContextOptional optional application-defined task context
+   * @param externalContextOptional optional external context
    */
-  public ContextImpl(JobContext jobContext,
-      ContainerContext containerContext,
-      TaskContext taskContext,
+  public ContextImpl(JobContext jobContext, ContainerContext containerContext, TaskContext taskContext,
       Optional<ApplicationContainerContext> applicationContainerContextOptional,
-      Optional<ApplicationTaskContext> applicationTaskContextOptional) {
+      Optional<ApplicationTaskContext> applicationTaskContextOptional,
+      Optional<ExternalContext> externalContextOptional) {
     this.jobContext = Preconditions.checkNotNull(jobContext, "Job context can not be null");
     this.containerContext = Preconditions.checkNotNull(containerContext, "Container context can not be null");
     this.taskContext = Preconditions.checkNotNull(taskContext, "Task context can not be null");
     this.applicationContainerContextOptional = applicationContainerContextOptional;
     this.applicationTaskContextOptional = applicationTaskContextOptional;
+    this.externalContextOptional = externalContextOptional;
   }
 
   @Override
@@ -67,20 +69,24 @@ public TaskContext getTaskContext() {
 
   @Override
   public ApplicationContainerContext getApplicationContainerContext() {
-    if (!this.applicationContainerContextOptional.isPresent()) {
-      throw new IllegalStateException("No application-defined container context exists");
-    }
+    Preconditions.checkState(this.applicationContainerContextOptional.isPresent(),
+        "No application-defined container context exists");
     return this.applicationContainerContextOptional.get();
   }
 
   @Override
   public ApplicationTaskContext getApplicationTaskContext() {
-    if (!this.applicationTaskContextOptional.isPresent()) {
-      throw new IllegalStateException("No application-defined task context exists");
-    }
+    Preconditions.checkState(this.applicationTaskContextOptional.isPresent(),
+        "No application-defined task context exists");
     return this.applicationTaskContextOptional.get();
   }
 
+  @Override
+  public ExternalContext getExternalContext() {
+    Preconditions.checkState(this.externalContextOptional.isPresent(), "No external context exists");
+    return this.externalContextOptional.get();
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {
@@ -93,12 +99,13 @@ public boolean equals(Object o) {
     return Objects.equals(jobContext, context.jobContext) && Objects.equals(containerContext, context.containerContext)
         && Objects.equals(taskContext, context.taskContext) && Objects.equals(applicationContainerContextOptional,
         context.applicationContainerContextOptional) && Objects.equals(applicationTaskContextOptional,
-        context.applicationTaskContextOptional);
+        context.applicationTaskContextOptional) && Objects.equals(externalContextOptional,
+        context.externalContextOptional);
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(jobContext, containerContext, taskContext, applicationContainerContextOptional,
-        applicationTaskContextOptional);
+        applicationTaskContextOptional, externalContextOptional);
   }
 }

samza-core/src/main/java/org/apache/samza/processor/StreamProcessor.java

@@ -40,6 +40,7 @@
 import org.apache.samza.context.ApplicationContainerContextFactory;
 import org.apache.samza.context.ApplicationTaskContext;
 import org.apache.samza.context.ApplicationTaskContextFactory;
+import org.apache.samza.context.ExternalContext;
 import org.apache.samza.context.JobContextImpl;
 import org.apache.samza.coordinator.JobCoordinator;
 import org.apache.samza.coordinator.JobCoordinatorFactory;
@@ -115,6 +116,7 @@ public class StreamProcessor {
    * context object.
    */
   private final Optional<ApplicationTaskContextFactory<ApplicationTaskContext>> applicationDefinedTaskContextFactoryOptional;
+  private final Optional<ExternalContext> externalContextOptional;
   private final Map<String, MetricsReporter> customMetricsReporter;
   private final Config config;
   private final long taskShutdownMs;
@@ -169,7 +171,7 @@ public State getState() {
    * @param taskFactory task factory to instantiate the Task
    * @param processorListener listener to the StreamProcessor life cycle
    *
-   * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional,
+   * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
    * StreamProcessorLifecycleListenerFactory, JobCoordinator)} instead.
    */
   @Deprecated
@@ -179,7 +181,7 @@ public StreamProcessor(Config config, Map<String, MetricsReporter> customMetrics
   }
 
   /**
-   * Same as {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional,
+   * Same as {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
    * StreamProcessorLifecycleListenerFactory, JobCoordinator)}, with the following differences:
    * <ol>
    *   <li>Passes null for application-defined context factories</li>
@@ -193,14 +195,14 @@ public StreamProcessor(Config config, Map<String, MetricsReporter> customMetrics
    * @param processorListener listener to the StreamProcessor life cycle
    * @param jobCoordinator the instance of {@link JobCoordinator}
    *
-   * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional,
+   * Deprecated: Use {@link #StreamProcessor(Config, Map, TaskFactory, Optional, Optional, Optional,
    * StreamProcessorLifecycleListenerFactory, JobCoordinator)} instead.
    */
   @Deprecated
   public StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
       ProcessorLifecycleListener processorListener, JobCoordinator jobCoordinator) {
-    this(config, customMetricsReporters, taskFactory, Optional.empty(), Optional.empty(), sp -> processorListener,
-        jobCoordinator);
+    this(config, customMetricsReporters, taskFactory, Optional.empty(), Optional.empty(), Optional.empty(),
+        sp -> processorListener, jobCoordinator);
   }
 
   /**
@@ -211,19 +213,22 @@ public StreamProcessor(Config config, Map<String, MetricsReporter> customMetrics
    * @param taskFactory task factory to instantiate the Task
    * @param applicationDefinedContainerContextFactoryOptional optional factory for application-defined container context
    * @param applicationDefinedTaskContextFactoryOptional optional factory for application-defined task context
+   * @param externalContextOptional optional {@link ExternalContext} to pass through to the application
    * @param listenerFactory factory for creating a listener to the StreamProcessor life cycle
    * @param jobCoordinator the instance of {@link JobCoordinator}
    */
   public StreamProcessor(Config config, Map<String, MetricsReporter> customMetricsReporters, TaskFactory taskFactory,
       Optional<ApplicationContainerContextFactory<ApplicationContainerContext>> applicationDefinedContainerContextFactoryOptional,
       Optional<ApplicationTaskContextFactory<ApplicationTaskContext>> applicationDefinedTaskContextFactoryOptional,
-      StreamProcessorLifecycleListenerFactory listenerFactory, JobCoordinator jobCoordinator) {
+      Optional<ExternalContext> externalContextOptional, StreamProcessorLifecycleListenerFactory listenerFactory,
+      JobCoordinator jobCoordinator) {
     Preconditions.checkNotNull(listenerFactory, "StreamProcessorListenerFactory cannot be null.");
     this.config = config;
     this.customMetricsReporter = customMetricsReporters;
     this.taskFactory = taskFactory;
     this.applicationDefinedContainerContextFactoryOptional = applicationDefinedContainerContextFactoryOptional;
     this.applicationDefinedTaskContextFactoryOptional = applicationDefinedTaskContextFactoryOptional;
+    this.externalContextOptional = externalContextOptional;
     this.taskShutdownMs = new TaskConfigJava(config).getShutdownMs();
     this.jobCoordinator = (jobCoordinator != null) ? jobCoordinator : createJobCoordinator();
     this.jobCoordinatorListener = createJobCoordinatorListener();
@@ -318,7 +323,7 @@ SamzaContainer createSamzaContainer(String processorId, JobModel jobModel) {
         this.taskFactory, JobContextImpl.fromConfigWithDefaults(this.config),
         Option.apply(this.applicationDefinedContainerContextFactoryOptional.orElse(null)),
         Option.apply(this.applicationDefinedTaskContextFactoryOptional.orElse(null)),
-        null);
+        Option.apply(this.externalContextOptional.orElse(null)), null);
   }
 
   private JobCoordinator createJobCoordinator() {

samza-core/src/main/java/org/apache/samza/runtime/ApplicationRunnerMain.java

@@ -26,7 +26,6 @@
 import org.apache.samza.util.CommandLine;
 import org.apache.samza.util.Util;
 
-
 /**
  * This class contains the main() method used by run-app.sh.
  * It creates the {@link ApplicationRunner} based on the config, and then run the application.
@@ -59,7 +58,7 @@ public static void main(String[] args) throws Exception {
 
     switch (op) {
       case RUN:
-        appRunner.run();
+        appRunner.run(null);
         break;
       case KILL:
         appRunner.kill();