protected void setupSystem() {
    config = getConfig();
    Log4jSystemConfig log4jSystemConfig = new Log4jSystemConfig(config);

    if (streamName == null) {
      streamName = getStreamName(log4jSystemConfig.getJobName(), log4jSystemConfig.getJobId());
    }

    // TODO we need the ACTUAL metrics registry, or the metrics won't get reported by the metric reporters!
    MetricsRegistry metricsRegistry = new MetricsRegistryMap();
    metrics = new StreamAppenderMetrics("stream-appender", metricsRegistry);

    String systemName = log4jSystemConfig.getSystemName();
    String systemFactoryName = log4jSystemConfig.getSystemFactory(systemName)
        .orElseThrow(() -> new SamzaException(
            "Could not figure out \"" + systemName + "\" system factory for log4j StreamAppender to use"));
    SystemFactory systemFactory = ReflectionUtil.getObj(systemFactoryName, SystemFactory.class);

    setSerde(log4jSystemConfig, systemName, streamName);

    if (config.getBoolean(CREATE_STREAM_ENABLED, false)) {
      int streamPartitionCount = getPartitionCount();
      System.out.println(
          "[StreamAppender] creating stream " + streamName + " with partition count " + streamPartitionCount);
      StreamSpec streamSpec = StreamSpec.createStreamAppenderStreamSpec(streamName, systemName, streamPartitionCount);

      // SystemAdmin only needed for stream creation here.
      SystemAdmin systemAdmin = systemFactory.getAdmin(systemName, config);
      systemAdmin.start();
      systemAdmin.createStream(streamSpec);
      systemAdmin.stop();
    }

    systemProducer = systemFactory.getProducer(systemName, config, metricsRegistry, this.getClass().getSimpleName());
    systemStream = new SystemStream(systemName, streamName);
    systemProducer.register(SOURCE);
    systemProducer.start();

    log.info(SOURCE + " has been registered in " + systemName + ". So all the logs will be sent to " + streamName
        + " in " + systemName + ". Logs are partitioned by " + key);

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