private void validateChangelogStreams() {
    LOG.info("Validating change log streams: " + storeChangelogs);

    for (SystemStream changelogSystemStream : storeChangelogs.values()) {
      SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(changelogSystemStream.getSystem());
      StreamSpec changelogSpec =
          StreamSpec.createChangeLogStreamSpec(changelogSystemStream.getStream(), changelogSystemStream.getSystem(),
              maxChangeLogStreamPartitions);

      systemAdmin.validateStream(changelogSpec);
    }
  }
Validates each changelog system-stream with its respective SystemAdmin.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java