diff --git a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
index 9d3e477..660638c 100644
--- a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
+++ b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
@@ -30,8 +30,6 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.core.Filter;
 import org.apache.logging.log4j.core.Layout;
 import org.apache.logging.log4j.core.LogEvent;
@@ -88,14 +86,16 @@
   private SystemStream systemStream = null;
   private SystemProducer systemProducer = null;
   private String key = null;
+  private byte[] keyBytes; // Serialize the key once, since we will use it for every event.
   private String containerName = null;
   private int partitionCount = 0;
   private boolean isApplicationMaster;
   private Serde<LogEvent> serde = null;
-  private Logger log = LogManager.getLogger(StreamAppender.class);
+
   private Thread transferThread;
   private Config config = null;
   private String streamName = null;
+  private final boolean usingAsyncLogger;
 
   /**
    * used to detect if this thread is called recursively
@@ -107,9 +107,11 @@
   protected StreamAppenderMetrics metrics;
   protected long queueTimeoutS = DEFAULT_QUEUE_TIMEOUT_S;
 
-  protected StreamAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions, String streamName) {
+  protected StreamAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions,
+      boolean usingAsyncLogger, String streamName) {
     super(name, filter, layout, ignoreExceptions);
     this.streamName = streamName;
+    this.usingAsyncLogger = usingAsyncLogger;
   }
 
   @Override
@@ -123,6 +125,13 @@
           ". This is used as the key for the log appender, so can't proceed.");
     }
     key = containerName; // use the container name as the key for the logs
+    try {
+      // Serialize the key once, since we will use it for every event.
+      keyBytes = key.getBytes("UTF-8");
+    } catch (UnsupportedEncodingException e) {
+      throw new SamzaException(
+          String.format("Container name: %s could not be encoded to bytes. %s cannot proceed.", key, getName()), e);
+    }
 
     // StreamAppender has to wait until the JobCoordinator is up when the log is in the AM
     if (isApplicationMaster) {
@@ -134,7 +143,7 @@
   }
 
   /**
-   * Getter for the StreamName parameter. See also {@link #createAppender(String, Filter, Layout, boolean, String)} for when this is called.
+   * Getter for the StreamName parameter. See also {@link #createAppender(String, Filter, Layout, boolean, boolean, String)} for when this is called.
    * Example: {@literal <param name="StreamName" value="ExampleStreamName"/>}
    * @return The configured stream name.
    */
@@ -153,7 +162,7 @@
   }
 
   /**
-   * Getter for the number of partitions to create on a new StreamAppender stream. See also {@link #createAppender(String, Filter, Layout, boolean, String)} for when this is called.
+   * Getter for the number of partitions to create on a new StreamAppender stream. See also {@link #createAppender(String, Filter, Layout, boolean, boolean, String)} for when this is called.
    * Example: {@literal <param name="PartitionCount" value="4"/>}
    * @return The configured partition count of the StreamAppender stream. If not set, returns {@link JobConfig#getContainerCount()}.
    */
@@ -165,7 +174,7 @@
   }
 
   /**
-   * Setter for the number of partitions to create on a new StreamAppender stream. See also {@link #createAppender(String, Filter, Layout, boolean, String)} for when this is called.
+   * Setter for the number of partitions to create on a new StreamAppender stream. See also {@link #createAppender(String, Filter, Layout, boolean, boolean, String)} for when this is called.
    * Example: {@literal <param name="PartitionCount" value="4"/>}
    * @param partitionCount Configurable partition count.
    */
@@ -180,8 +189,9 @@
       @PluginElement("Filter") final Filter filter,
       @PluginElement("Layout") Layout layout,
       @PluginAttribute(value = "ignoreExceptions", defaultBoolean = true) final boolean ignoreExceptions,
+      @PluginAttribute(value = "usingAsyncLogger", defaultBoolean = false) final boolean usingAsyncLogger,
       @PluginAttribute("streamName") String streamName) {
-    return new StreamAppender(name, filter, layout, ignoreExceptions, streamName);
+    return new StreamAppender(name, filter, layout, ignoreExceptions, usingAsyncLogger, streamName);
   }
 
   @Override
@@ -195,35 +205,11 @@
             setupSystem();
             systemInitialized = true;
           } else {
-            log.trace("Waiting for the JobCoordinator to be instantiated...");
+            System.out.println("Waiting for the JobCoordinator to be instantiated...");
           }
         } else {
-          // Serialize the event before adding to the queue to leverage the caller thread
-          // and ensure that the transferThread can keep up.
-          if (!logQueue.offer(encodeLogEventToBytes(event), queueTimeoutS, TimeUnit.SECONDS)) {
-            // Do NOT retry adding system to the queue. Dropping the event allows us to alleviate the unlikely
-            // possibility of a deadlock, which can arise due to a circular dependency between the SystemProducer
-            // which is used for StreamAppender and the log, which uses StreamAppender. Any locks held in the callstack
-            // of those two code paths can cause a deadlock. Dropping the event allows us to proceed.
-
-            // Scenario:
-            // T1: holds L1 and is waiting for L2
-            // T2: holds L2 and is waiting to produce to BQ1 which is drained by T3 (SystemProducer) which is waiting for L1
-
-            // This has happened due to locks in Kafka and log4j (see SAMZA-1537), which are both out of our control,
-            // so dropping events in the StreamAppender is our best recourse.
-
-            // Drain the queue instead of dropping one message just to reduce the frequency of warn logs above.
-            int messagesDropped = logQueue.drainTo(new ArrayList<>()) + 1; // +1 because of the current log event
-            log.warn(String.format("Exceeded timeout %ss while trying to log to %s. Dropping %d log messages.",
-                queueTimeoutS,
-                systemStream.toString(),
-                messagesDropped));
-
-            // Emit a metric which can be monitored to ensure it doesn't happen often.
-            metrics.logMessagesDropped.inc(messagesDropped);
-          }
-          metrics.bufferFillPct.set(Math.round(100f * logQueue.size() / DEFAULT_QUEUE_SIZE));
+          // handle event based on if async or sync logger is being used
+          handleEvent(event);
         }
       } catch (Exception e) {
         if (metrics != null) { // setupSystem() may not have been invoked yet so metrics can be null here.
@@ -239,6 +225,46 @@
     }
   }
 
+  /**
+   * If async-Logger is enabled, the log-event is sent directly to the systemProducer. Else, the event is serialized
+   * and added to a bounded blocking queue, before returning to the "synchronous" caller.
+   * @param event the log event to append
+   * @throws InterruptedException
+   */
+  private void handleEvent(LogEvent event) throws InterruptedException {
+    if (usingAsyncLogger) {
+      sendEventToSystemProducer(encodeLogEventToBytes(event));
+      return;
+    }
+
+    // Serialize the event before adding to the queue to leverage the caller thread
+    // and ensure that the transferThread can keep up.
+    if (!logQueue.offer(encodeLogEventToBytes(event), queueTimeoutS, TimeUnit.SECONDS)) {
+      // Do NOT retry adding system to the queue. Dropping the event allows us to alleviate the unlikely
+      // possibility of a deadlock, which can arise due to a circular dependency between the SystemProducer
+      // which is used for StreamAppender and the log, which uses StreamAppender. Any locks held in the callstack
+      // of those two code paths can cause a deadlock. Dropping the event allows us to proceed.
+
+      // Scenario:
+      // T1: holds L1 and is waiting for L2
+      // T2: holds L2 and is waiting to produce to BQ1 which is drained by T3 (SystemProducer) which is waiting for L1
+
+      // This has happened due to locks in Kafka and log4j (see SAMZA-1537), which are both out of our control,
+      // so dropping events in the StreamAppender is our best recourse.
+
+      // Drain the queue instead of dropping one message just to reduce the frequency of warn logs above.
+      int messagesDropped = logQueue.drainTo(new ArrayList<>()) + 1; // +1 because of the current log event
+      System.err.println(String.format("Exceeded timeout %ss while trying to log to %s. Dropping %d log messages.",
+          queueTimeoutS,
+          systemStream.toString(),
+          messagesDropped));
+
+      // Emit a metric which can be monitored to ensure it doesn't happen often.
+      metrics.logMessagesDropped.inc(messagesDropped);
+    }
+    metrics.bufferFillPct.set(Math.round(100f * logQueue.size() / DEFAULT_QUEUE_SIZE));
+  }
+
   protected byte[] encodeLogEventToBytes(LogEvent event) {
     return serde.toBytes(subLog(event));
   }
@@ -275,12 +301,12 @@
 
   @Override
   public void stop() {
-    log.info(String.format("Shutting down the %s...", getName()));
+    System.out.println(String.format("Shutting down the %s...", getName()));
     transferThread.interrupt();
     try {
       transferThread.join();
     } catch (InterruptedException e) {
-      log.error("Interrupted while waiting for transfer thread to finish.", e);
+      System.err.println("Interrupted while waiting for transfer thread to finish." + e);
       Thread.currentThread().interrupt();
     }
 
@@ -382,50 +408,41 @@
     systemProducer.register(SOURCE);
     systemProducer.start();
 
-    log.info(SOURCE + " has been registered in " + systemName + ". So all the logs will be sent to " + streamName
+    System.out.println(SOURCE + " has been registered in " + systemName + ". So all the logs will be sent to " + streamName
         + " in " + systemName + ". Logs are partitioned by " + key);
 
     startTransferThread();
   }
 
   private void startTransferThread() {
-
-    try {
-      // Serialize the key once, since we will use it for every event.
-      final byte[] keyBytes = key.getBytes("UTF-8");
-
-      Runnable transferFromQueueToSystem = () -> {
-        while (!Thread.currentThread().isInterrupted()) {
-          try {
-            byte[] serializedLogEvent = logQueue.take();
-
-            metrics.logMessagesBytesSent.inc(serializedLogEvent.length);
-            metrics.logMessagesCountSent.inc();
-
-            OutgoingMessageEnvelope outgoingMessageEnvelope =
-                new OutgoingMessageEnvelope(systemStream, keyBytes, serializedLogEvent);
-            systemProducer.send(SOURCE, outgoingMessageEnvelope);
-
-          } catch (InterruptedException e) {
-            // Preserve the interrupted status for the loop condition.
-            Thread.currentThread().interrupt();
-          } catch (Throwable t) {
-            metrics.logMessagesErrors.inc();
-            log.error("Error sending " + getName() + " event to SystemProducer", t);
-          }
+    Runnable transferFromQueueToSystem = () -> {
+      while (!Thread.currentThread().isInterrupted()) {
+        try {
+          sendEventToSystemProducer(logQueue.take());
+        } catch (InterruptedException e) {
+          // Preserve the interrupted status for the loop condition.
+          Thread.currentThread().interrupt();
+        } catch (Throwable t) {
+          metrics.logMessagesErrors.inc();
+          System.err.println("Error sending " + getName() + " event to SystemProducer " + t);
         }
-      };
+      }
+    };
 
-      transferThread = new Thread(transferFromQueueToSystem);
-      transferThread.setDaemon(true);
-      transferThread.setName("Samza " + getName() + " Producer " + transferThread.getName());
-      transferThread.start();
+    transferThread = new Thread(transferFromQueueToSystem);
+    transferThread.setDaemon(true);
+    transferThread.setName("Samza " + getName() + " Producer " + transferThread.getName());
+    transferThread.start();
+  }
 
-    } catch (UnsupportedEncodingException e) {
-      throw new SamzaException(String.format(
-          "Container name: %s could not be encoded to bytes. %s cannot proceed.", key, getName()),
-          e);
-    }
+  /**
+   * Helper method to send a serialized log-event to the systemProducer, and increment respective methods.
+   * @param serializedLogEvent
+   */
+  private void sendEventToSystemProducer(byte[] serializedLogEvent) {
+    metrics.logMessagesBytesSent.inc(serializedLogEvent.length);
+    metrics.logMessagesCountSent.inc();
+    systemProducer.send(SOURCE, new OutgoingMessageEnvelope(systemStream, keyBytes, serializedLogEvent));
   }
 
   protected String getStreamName(String jobName, String jobId) {
diff --git a/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/MockSystemProducerAppender.java b/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/MockSystemProducerAppender.java
index 07df10c..ce6f081 100644
--- a/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/MockSystemProducerAppender.java
+++ b/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/MockSystemProducerAppender.java
@@ -41,8 +41,8 @@
 class MockSystemProducerAppender extends StreamAppender {
   private static Config config;
 
-  protected MockSystemProducerAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions, Config config, String streamName) {
-    super(name, filter, layout, ignoreExceptions, streamName);
+  protected MockSystemProducerAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions, final boolean usingAsyncLogger, Config config, String streamName) {
+    super(name, filter, layout, ignoreExceptions, usingAsyncLogger, streamName);
   }
 
   @PluginFactory
@@ -51,6 +51,7 @@
       @PluginElement("Filter") final Filter filter,
       @PluginElement("Layout") Layout<? extends Serializable> layout,
       @PluginAttribute(value = "ignoreExceptions", defaultBoolean = true) final boolean ignoreExceptions,
+      @PluginAttribute(value = "usingAsyncLogger", defaultBoolean = false) final boolean usingAsyncLogger,
       @PluginElement("Config") final Config testConfig,
       @PluginAttribute("streamName") String streamName) {
     if (testConfig == null) {
@@ -58,7 +59,7 @@
     } else {
       config = testConfig;
     }
-    return new MockSystemProducerAppender(name, filter, layout, ignoreExceptions, config, streamName);
+    return new MockSystemProducerAppender(name, filter, layout, ignoreExceptions, usingAsyncLogger, config, streamName);
   }
 
   @Override
diff --git a/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/TestStreamAppender.java b/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/TestStreamAppender.java
index 1680a84..2cfd59b 100644
--- a/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/TestStreamAppender.java
+++ b/samza-log4j2/src/test/java/org/apache/samza/logging/log4j2/TestStreamAppender.java
@@ -19,21 +19,28 @@
 
 package org.apache.samza.logging.log4j2;
 
-import static org.junit.Assert.*;
-
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import org.apache.logging.log4j.Level;
-import org.apache.logging.log4j.core.Logger;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.core.Appender;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.ConfigurationFactory;
+import org.apache.logging.log4j.core.config.ConfigurationSource;
+import org.apache.logging.log4j.core.config.Order;
+import org.apache.logging.log4j.core.config.builder.api.ConfigurationBuilder;
+import org.apache.logging.log4j.core.config.builder.api.RootLoggerComponentBuilder;
+import org.apache.logging.log4j.core.config.builder.impl.BuiltConfiguration;
+import org.apache.logging.log4j.core.config.plugins.Plugin;
 import org.apache.logging.log4j.core.layout.PatternLayout;
 import org.apache.samza.config.MapConfig;
 import org.apache.samza.logging.log4j2.serializers.LoggingEventJsonSerde;
@@ -43,6 +50,8 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import static org.junit.Assert.*;
+
 public class TestStreamAppender {
 
   static Logger log = (Logger) LogManager.getLogger(TestStreamAppender.class);
@@ -59,7 +68,8 @@
   public void testDefaultSerde() {
     System.setProperty("samza.container.name", "samza-container-1");
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
     assertNotNull(systemProducerAppender.getSerde());
     assertEquals(LoggingEventJsonSerde.class, systemProducerAppender.getSerde().getClass());
@@ -76,7 +86,7 @@
     map.put("systems.mock.streams.__samza_log4jTest_1_logs.samza.msg.serde", "log4j-string");
     map.put("task.log4j.system", "mock");
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, new MapConfig(map), null);
+    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, false, new MapConfig(map), null);
     systemProducerAppender.start();
     assertNotNull(systemProducerAppender.getSerde());
     assertEquals(LoggingEventStringSerde.class, systemProducerAppender.getSerde().getClass());
@@ -86,7 +96,7 @@
   public void testDefaultStreamName() {
     System.setProperty("samza.container.name", "samza-container-1");
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
     Assert.assertEquals("__samza_log4jTest_1_logs", systemProducerAppender.getStreamName());
@@ -96,7 +106,7 @@
   public void testCustomStreamName() {
     System.setProperty("samza.container.name", "samza-container-1");
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, "test-stream-name");
+    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, "test-stream-name");
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
     Assert.assertEquals("test-stream-name", systemProducerAppender.getStreamName());
@@ -107,7 +117,8 @@
     System.setProperty("samza.container.name", "samza-container-1");
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
 
     log.addAppender(systemProducerAppender);
@@ -118,11 +129,58 @@
   }
 
   @Test
+  public void testSystemProducerAppenderInContainerWithAsyncLogger() throws InterruptedException {
+    System.setProperty("samza.container.name", "samza-container-1");
+    // Enabling async logger on log4j2 programmatically
+    ConfigurationFactory.setConfigurationFactory(new AsyncLoggerConfigurationFactory());
+
+    PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, true, null, null);
+    systemProducerAppender.start();
+    log.addAppender(systemProducerAppender);
+    log.setLevel(Level.INFO);
+    List<String> messages = Lists.newArrayList("testing1", "testing2");
+    logAndVerifyMessages(messages);
+    systemProducerAppender.stop();
+  }
+
+
+  @Plugin(name = "AsyncLoggerConfigurationFactory", category = ConfigurationFactory.CATEGORY)
+  @Order(50)
+  public static class AsyncLoggerConfigurationFactory extends ConfigurationFactory {
+
+    private static Configuration createConfiguration(final String name, ConfigurationBuilder<BuiltConfiguration> builder) {
+      builder.setConfigurationName(name);
+      RootLoggerComponentBuilder rootLoggerBuilder = builder.newAsyncRootLogger(Level.INFO);
+      builder.add(rootLoggerBuilder);
+      return builder.build();
+    }
+
+    @Override
+    public Configuration getConfiguration(final LoggerContext loggerContext, final ConfigurationSource source) {
+      return getConfiguration(loggerContext, source.toString(), null);
+    }
+
+    @Override
+    public Configuration getConfiguration(final LoggerContext loggerContext, final String name, final URI configLocation) {
+      ConfigurationBuilder<BuiltConfiguration> builder = newConfigurationBuilder();
+      return createConfiguration(name, builder);
+    }
+
+    @Override
+    protected String[] getSupportedTypes() {
+      return new String[]{"*"};
+    }
+  }
+
+  @Test
   public void testSystemProducerAppenderInAM() throws InterruptedException {
     System.setProperty("samza.container.name", "samza-job-coordinator");
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
     log.setLevel(Level.INFO);
@@ -142,7 +200,8 @@
     System.setProperty("samza.container.name", "samza-container-1");
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
 
@@ -161,7 +220,8 @@
         "task.log4j.system", "mock"));
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, mapConfig, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, mapConfig, null);
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
 
@@ -171,7 +231,8 @@
   @Test
   public void testDefaultPartitionCount() {
     System.setProperty("samza.container.name", "samza-container-1");
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, null, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, null, false, false, null, null);
     Assert.assertEquals(1, systemProducerAppender.getPartitionCount()); // job.container.count defaults to 1
 
     Map<String, String> map = new HashMap<>();
@@ -180,10 +241,12 @@
     map.put("systems.mock.samza.factory", MockSystemFactory.class.getCanonicalName());
     map.put("task.log4j.system", "mock");
     map.put("job.container.count", "4");
-    systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, null, false, new MapConfig(map), null);
+    systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, null, false, false, new MapConfig(map), null);
     Assert.assertEquals(4, systemProducerAppender.getPartitionCount());
 
-    systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, null, false, null, null);
+    systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, null, false, false, null, null);
     systemProducerAppender.setPartitionCount(8);
     Assert.assertEquals(8, systemProducerAppender.getPartitionCount());
   }
@@ -193,7 +256,8 @@
     System.setProperty("samza.container.name", "samza-container-1");
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);
     log.setLevel(Level.INFO);
@@ -223,7 +287,8 @@
     System.setProperty("samza.container.name", "samza-container-1");
 
     PatternLayout layout = PatternLayout.newBuilder().withPattern("%m").build();
-    MockSystemProducerAppender systemProducerAppender = MockSystemProducerAppender.createAppender("testName", null, layout, false, null, null);
+    MockSystemProducerAppender systemProducerAppender =
+        MockSystemProducerAppender.createAppender("testName", null, layout, false, false, null, null);
     systemProducerAppender.queueTimeoutS = 1;
     systemProducerAppender.start();
     log.addAppender(systemProducerAppender);