public Optional<String> getChangelogStream(String storeName) {
    String systemStream = StringUtils.trimToNull(get(String.format(CHANGELOG_STREAM, storeName), null));

    String systemStreamRes;
    if (systemStream != null && !systemStream.contains(".")) {
      Optional<String> changelogSystem = getChangelogSystem();
      // contains only stream name
      if (changelogSystem.isPresent()) {
        systemStreamRes = changelogSystem.get() + "." + systemStream;
      } else {
        throw new SamzaException("changelog system is not defined:" + systemStream);
      }
    } else {
      systemStreamRes = systemStream;
    }

    if (systemStreamRes != null) {
      systemStreamRes = StreamManager.createUniqueNameForBatch(systemStreamRes, this);
    }
    return Optional.ofNullable(systemStreamRes);
  }
If the config specifies 'stores.&lt;storename&gt;.changelog' as '&lt;system&gt;.&lt;stream&gt;' combination - it will take
precedence.
If this config only specifies &lt;astream&gt; and there is a value in job.changelog.system=&lt;asystem&gt; - these values will
be combined into &lt;asystem&gt;.&lt;astream&gt;
samza-core/src/main/java/org/apache/samza/config/StorageConfig.java