Config generateConfig() {
    Map<String, String> streamConfig = new HashMap<>();
    StreamSpec spec = getStreamSpec();
    String streamId = spec.getId();
    streamConfig.put(String.format(StreamConfig.SYSTEM_FOR_STREAM_ID, streamId), spec.getSystemName());
    streamConfig.put(String.format(StreamConfig.PHYSICAL_NAME_FOR_STREAM_ID, streamId), spec.getPhysicalName());
    if (isIntermediate()) {
      streamConfig.put(String.format(StreamConfig.IS_INTERMEDIATE_FOR_STREAM_ID, streamId), "true");
      streamConfig.put(String.format(StreamConfig.DELETE_COMMITTED_MESSAGES_FOR_STREAM_ID, streamId), "true");

      // Setting offset.default to oldest only if the job is running in batch mode
      if (ApplicationConfig.ApplicationMode.BATCH.equals(new ApplicationConfig(config).getAppMode())) {
        streamConfig.put(String.format(StreamConfig.CONSUMER_OFFSET_DEFAULT_FOR_STREAM_ID, streamId), "oldest");
      }

      streamConfig.put(String.format(StreamConfig.PRIORITY_FOR_STREAM_ID, streamId), String.valueOf(Integer.MAX_VALUE));
    }
    spec.getConfig().forEach((property, value) -> {
      streamConfig.put(String.format(StreamConfig.STREAM_ID_PREFIX, streamId) + property, value);
    });

    return new MapConfig(streamConfig);
  }
A StreamEdge connects the source {@link JobNode}s to the target {@link JobNode}s with a stream.
If it's a sink StreamEdge, the target JobNode is empty.
If it's a source StreamEdge, the source JobNode is empty.
samza-core/src/main/java/org/apache/samza/execution/StreamEdge.java