SAMZA-2693: Make Samza log4j appenders agnostic of where they are running (apache#1532)

API/usage changes:
1. (functionality change) In worker containers, the logs for fetching the initial job model (in SamzaContainer.readJobModel) will no longer be sent to StreamAppender, because the config will not be initialized yet in the appender.
2. (test change) MockSystemAdmin.createdStreamName was changed to MockSystemAdmin.createdStreamSpec. The stream name can be accessed with MockSystemAdmin.createdStreamSpec.getPhysicalName().

Author: cameronlee314

build.gradle

@@ -449,6 +449,7 @@ project(":samza-log4j2_$scalaSuffix") {
     compile project(":samza-core_$scalaSuffix")
     compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion"
     testCompile "junit:junit:$junitVersion"
+    testCompile "org.mockito:mockito-core:$mockitoVersion"
   }
 }
 

samza-core/src/main/java/org/apache/samza/logging/LoggingContextHolder.java

@@ -0,0 +1,56 @@
+/*
+ * 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.logging;
+
+import java.util.concurrent.atomic.AtomicReference;
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.samza.config.Config;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Holds information to be used by loggers. For example, some custom Samza log4j/log4j2 logging appenders need system
+ * configs for initialization, so this allows the configs to be passed to those appenders.
+ */
+public class LoggingContextHolder {
+  private static final Logger LOG = LoggerFactory.getLogger(LoggingContextHolder.class);
+  public static final LoggingContextHolder INSTANCE = new LoggingContextHolder();
+
+  private final AtomicReference<Config> config = new AtomicReference<>();
+
+  @VisibleForTesting
+  LoggingContextHolder() {
+  }
+
+  /**
+   * Set the config to be used by Samza loggers.
+   * Only the config used in the first call to this method will be used. After the first call, this method will do
+   * nothing.
+   */
+  public void setConfig(Config config) {
+    if (!this.config.compareAndSet(null, config)) {
+      LOG.warn("Attempted to set config, but it was already set");
+    }
+  }
+
+  public Config getConfig() {
+    return this.config.get();
+  }
+}

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

@@ -44,6 +44,7 @@
 import org.apache.samza.coordinator.stream.messages.SetExecutionEnvContainerIdMapping;
 import org.apache.samza.diagnostics.DiagnosticsManager;
 import org.apache.samza.job.model.JobModel;
+import org.apache.samza.logging.LoggingContextHolder;
 import org.apache.samza.metadatastore.MetadataStore;
 import org.apache.samza.metrics.MetricsRegistryMap;
 import org.apache.samza.metrics.MetricsReporter;
@@ -81,14 +82,14 @@ public static void run(
       ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc,
       String jobName, String jobId, String containerId, Optional<String> execEnvContainerId,
       JobModel jobModel) {
+    Config config = jobModel.getConfig();
 
-    // populate MDC for logging
+    // logging setup: MDC, logging context
     MDC.put("containerName", "samza-container-" + containerId);
     MDC.put("jobName", jobName);
     MDC.put("jobId", jobId);
+    LoggingContextHolder.INSTANCE.setConfig(jobModel.getConfig());
 
-
-    Config config = jobModel.getConfig();
     DiagnosticsUtil.writeMetadataFile(jobName, jobId, containerId, execEnvContainerId, config);
     run(appDesc, jobName, jobId, containerId, execEnvContainerId, jobModel, config, buildExternalContext(config));
 

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

@@ -31,6 +31,7 @@
 import org.apache.samza.config.ShellCommandConfig;
 import org.apache.samza.container.SamzaContainer;
 import org.apache.samza.job.model.JobModel;
+import org.apache.samza.logging.LoggingContextHolder;
 import org.apache.samza.util.SamzaUncaughtExceptionHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,18 +51,22 @@ public static void main(String[] args) throws Exception {
         }));
 
     String containerId = System.getenv(ShellCommandConfig.ENV_CONTAINER_ID);
-    log.info(String.format("Got container ID: %s", containerId));
     System.out.println(String.format("Container ID: %s", containerId));
 
     String coordinatorUrl = System.getenv(ShellCommandConfig.ENV_COORDINATOR_URL);
-    log.info(String.format("Got coordinator URL: %s", coordinatorUrl));
     System.out.println(String.format("Coordinator URL: %s", coordinatorUrl));
 
     Optional<String> execEnvContainerId = Optional.ofNullable(System.getenv(ShellCommandConfig.ENV_EXECUTION_ENV_CONTAINER_ID));
 
     int delay = new Random().nextInt(SamzaContainer.DEFAULT_READ_JOBMODEL_DELAY_MS()) + 1;
     JobModel jobModel = SamzaContainer.readJobModel(coordinatorUrl, delay);
     Config config = jobModel.getConfig();
+
+    // this call is also in ContainerLaunchUtil, but adding this here allows more logs to get handled by Samza loggers
+    LoggingContextHolder.INSTANCE.setConfig(config);
+    log.info(String.format("Got container ID: %s", containerId));
+    log.info(String.format("Got coordinator URL: %s", coordinatorUrl));
+
     JobConfig jobConfig = new JobConfig(config);
     String jobName = jobConfig.getName()
         .orElseThrow(() -> new SamzaException(String.format("Config %s is missing", JobConfig.JOB_NAME)));

samza-core/src/main/scala/org/apache/samza/coordinator/JobModelManager.scala

@@ -26,6 +26,7 @@ import org.apache.samza.coordinator.metadatastore.NamespaceAwareCoordinatorStrea
 import org.apache.samza.coordinator.server.{HttpServer, JobServlet, LocalityServlet}
 import org.apache.samza.coordinator.stream.messages.{SetContainerHostMapping, SetTaskContainerMapping, SetTaskModeMapping, SetTaskPartitionMapping}
 import org.apache.samza.job.model.JobModel
+import org.apache.samza.logging.LoggingContextHolder
 import org.apache.samza.metadatastore.MetadataStore
 import org.apache.samza.metrics.{MetricsRegistry, MetricsRegistryMap}
 import org.apache.samza.serializers.model.SamzaObjectMapper
@@ -40,15 +41,8 @@ import java.util.concurrent.atomic.AtomicReference
  * given a Config object.
  */
 object JobModelManager extends Logging {
-
   val SOURCE = "JobModelManager"
 
-  /**
-   * a volatile value to store the current instantiated <code>JobModelManager</code>
-   */
-  @volatile var currentJobModelManager: JobModelManager = _
-  val serializedJobModelRef = new AtomicReference[Array[Byte]]
-
   /**
    * Currently used only in the ApplicationMaster for yarn deployment model.
    * Does the following:
@@ -78,15 +72,14 @@ object JobModelManager extends Logging {
       val jobModel = jobModelHelper.newJobModel(config, changelogPartitionMapping)
       val jobModelToServe = new JobModel(jobModel.getConfig, jobModel.getContainers)
       val serializedJobModelToServe = SamzaObjectMapper.getObjectMapper().writeValueAsBytes(jobModelToServe)
-      serializedJobModelRef.set(serializedJobModelToServe)
 
       val clusterManagerConfig = new ClusterManagerConfig(config)
       val server = new HttpServer(port = clusterManagerConfig.getCoordinatorUrlPort)
-      server.addServlet("/", new JobServlet(serializedJobModelRef))
+      server.addServlet("/", new JobServlet(new AtomicReference[Array[Byte]](serializedJobModelToServe)))
       server.addServlet("/locality", new LocalityServlet(localityManager))
 
-      currentJobModelManager = new JobModelManager(jobModelToServe, server)
-      currentJobModelManager
+      LoggingContextHolder.INSTANCE.setConfig(jobModel.getConfig)
+      new JobModelManager(jobModelToServe, server)
     } finally {
       systemAdmins.stop()
       // Not closing coordinatorStreamStore, since {@code ClusterBasedJobCoordinator} uses it to read container locality through {@code JobModel}.

samza-core/src/test/java/org/apache/samza/logging/TestLoggingContextHolder.java

@@ -0,0 +1,60 @@
+/*
+ * 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.logging;
+
+import org.apache.samza.config.Config;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+
+
+public class TestLoggingContextHolder {
+  @Mock
+  private Config config;
+
+  private LoggingContextHolder loggingContextHolder;
+
+  @Before
+  public void setup() {
+    MockitoAnnotations.initMocks(this);
+    this.loggingContextHolder = new LoggingContextHolder();
+  }
+
+  @Test
+  public void testGet() {
+    assertNull(this.loggingContextHolder.getConfig());
+
+    this.loggingContextHolder.setConfig(this.config);
+    assertEquals(this.config, this.loggingContextHolder.getConfig());
+  }
+
+  @Test
+  public void testSetMultiple() {
+    this.loggingContextHolder.setConfig(this.config);
+    Config config0 = mock(Config.class);
+    this.loggingContextHolder.setConfig(config0);
+    // should still have first config
+    assertEquals(this.config, this.loggingContextHolder.getConfig());
+  }
+}

samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala

@@ -268,11 +268,10 @@ class TestSamzaContainer extends AssertionsForJUnit with MockitoSugar {
       "0" -> new ContainerModel("0", tasks),
       "1" -> new ContainerModel("1", tasks))
     val jobModel = new JobModel(config, containers)
-    def jobModelGenerator(): Array[Byte] = SamzaObjectMapper.getObjectMapper.writeValueAsBytes(jobModel)
     val server = new HttpServer
     val coordinator = new JobModelManager(jobModel, server)
-    JobModelManager.serializedJobModelRef.set(jobModelGenerator())
-    coordinator.server.addServlet("/*", new JobServlet(JobModelManager.serializedJobModelRef))
+    val serializedJobModel = SamzaObjectMapper.getObjectMapper.writeValueAsBytes(jobModel)
+    coordinator.server.addServlet("/*", new JobServlet(new AtomicReference[Array[Byte]](serializedJobModel)))
     try {
       coordinator.start
       assertEquals(jobModel, SamzaContainer.readJobModel(server.getUrl.toString))
@@ -293,11 +292,10 @@ class TestSamzaContainer extends AssertionsForJUnit with MockitoSugar {
       "0" -> new ContainerModel("0", tasks),
       "1" -> new ContainerModel("1", tasks))
     val jobModel = new JobModel(config, containers)
-    def jobModelGenerator(): Array[Byte] = SamzaObjectMapper.getObjectMapper.writeValueAsBytes(jobModel)
     val server = new HttpServer
     val coordinator = new JobModelManager(jobModel, server)
-    JobModelManager.serializedJobModelRef.set(jobModelGenerator())
-    val mockJobServlet = new MockJobServlet(2, JobModelManager.serializedJobModelRef)
+    val serializedJobModel = SamzaObjectMapper.getObjectMapper.writeValueAsBytes(jobModel)
+    val mockJobServlet = new MockJobServlet(2, new AtomicReference[Array[Byte]](serializedJobModel))
     coordinator.server.addServlet("/*", mockJobServlet)
     try {
       coordinator.start

samza-core/src/test/scala/org/apache/samza/coordinator/TestJobModelManager.scala

@@ -101,28 +101,26 @@ class TestJobModelManager extends FlatSpec with PrivateMethodTester {
     // We want the mocksystemconsumer to use the same instance across runs
     MockCoordinatorStreamSystemFactory.enableMockConsumerCache()
 
-    val coordinator = getTestJobModelManager(new MapConfig((config ++ otherConfigs).asJava))
+    val jobModelManager = getTestJobModelManager(new MapConfig((config ++ otherConfigs).asJava))
     val expectedJobModel = new JobModel(new MapConfig(config.asJava), containers.asJava)
 
-    // Verify that the atomicReference is initialized
-    assertNotNull(JobModelManager.serializedJobModelRef.get())
     val expectedContainerModels = new util.TreeMap[String, ContainerModel](expectedJobModel.getContainers)
-    val jobModel = SamzaObjectMapper.getObjectMapper.readValue(JobModelManager.serializedJobModelRef.get(), classOf[JobModel])
+    val jobModel = jobModelManager.jobModel
     val actualContainerModels = new util.TreeMap[String, ContainerModel](jobModel.getContainers)
     assertEquals(expectedContainerModels, actualContainerModels)
 
-    coordinator.start
-    val expectedConfig: Config = coordinator.jobModel.getConfig
+    jobModelManager.start
+    val expectedConfig: Config = jobModelManager.jobModel.getConfig
     val actualConfig: Config = new MapConfig(config.asJava)
     assertTrue(expectedConfig.entrySet().containsAll(actualConfig.entrySet()))
-    assertEquals(expectedJobModel.getContainers, coordinator.jobModel.getContainers)
+    assertEquals(expectedJobModel.getContainers, jobModelManager.jobModel.getContainers)
 
-    val response = HttpUtil.read(coordinator.server.getUrl)
+    val response = HttpUtil.read(jobModelManager.server.getUrl)
     // Verify that the JobServlet is serving the correct jobModel
     val jobModelFromCoordinatorUrl = SamzaObjectMapper.getObjectMapper.readValue(response, classOf[JobModel])
     assertEquals(expectedJobModel.getContainers, jobModelFromCoordinatorUrl.getContainers)
 
-    coordinator.stop
+    jobModelManager.stop
   }
 
   @Test