@VisibleForTesting
  static Map<String, Properties> getIntermediateStreamProperties(Config config) {
    Map<String, Properties> intermedidateStreamProperties = Collections.emptyMap();
    ApplicationConfig appConfig = new ApplicationConfig(config);

    if (appConfig.getAppMode() == ApplicationConfig.ApplicationMode.BATCH) {
      StreamConfig streamConfig = new StreamConfig(config);
      intermedidateStreamProperties = streamConfig.getStreamIds()
          .stream()
          .filter(streamConfig::getIsIntermediateStream)
          .collect(Collectors.toMap(Function.identity(), streamId -> {
            Properties properties = new Properties();
            properties.putAll(streamConfig.getStreamProperties(streamId));
            properties.putIfAbsent(TopicConfig.RETENTION_MS_CONFIG, String.valueOf(KafkaConfig.DEFAULT_RETENTION_MS_FOR_BATCH()));
            return properties;
          }));
    }

    return intermedidateStreamProperties;
  }
Fetch stream properties for all intermediate streams.

@param config kafka system config
@return a {@link Map} from {@code streamId} to stream {@link Properties}
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java