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();

            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) {
            log.error("Error sending StreamAppender event to SystemProducer", t);
          }
        }
      };

      transferThread = new Thread(transferFromQueueToSystem);
      transferThread.setDaemon(true);
      transferThread.setName("Samza StreamAppender Producer " + transferThread.getName());
      transferThread.start();

    } catch (UnsupportedEncodingException e) {
      throw new SamzaException(String.format(
          "Container name: %s could not be encoded to bytes. StreamAppender cannot proceed.", key),
          e);
    }
  }
This should only be called after verifying that the {@link LoggingContextHolder} has the config.
samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java