public static void createChangelogStreams(Config config, int maxChangeLogStreamPartitions) {
    // Get changelog store config
    StorageConfig storageConfig = new StorageConfig(config);
    ImmutableMap.Builder<String, SystemStream> storeNameSystemStreamMapBuilder = new ImmutableMap.Builder<>();
    storageConfig.getStoreNames().forEach(storeName -> {
      Optional<String> changelogStream = storageConfig.getChangelogStream(storeName);
      if (changelogStream.isPresent() && StringUtils.isNotBlank(changelogStream.get())) {
        storeNameSystemStreamMapBuilder.put(storeName, StreamUtil.getSystemStreamFromNames(changelogStream.get()));
      }
    });
    Map<String, SystemStream> storeNameSystemStreamMapping = storeNameSystemStreamMapBuilder.build();

    // Get SystemAdmin for changelog store's system and attempt to create the stream
    SystemConfig systemConfig = new SystemConfig(config);
    storeNameSystemStreamMapping.forEach((storeName, systemStream) -> {
      // Load system admin for this system.
      SystemAdmin systemAdmin = systemConfig
          .getSystemFactories()
          .get(systemStream.getSystem())
          .getAdmin(systemStream.getSystem(), config, ChangelogStreamManager.class.getSimpleName());

      if (systemAdmin == null) {
        throw new SamzaException(String.format(
            "Error creating changelog. Changelog on store %s uses system %s, which is missing from the configuration.",
            storeName, systemStream.getSystem()));
      }

      StreamSpec changelogSpec =
          StreamSpec.createChangeLogStreamSpec(systemStream.getStream(), systemStream.getSystem(),
              maxChangeLogStreamPartitions);

      systemAdmin.start();

      if (systemAdmin.createStream(changelogSpec)) {
        LOG.info(String.format("created changelog stream %s.", systemStream.getStream()));
      } else {
        LOG.info(String.format("changelog stream %s already exists.", systemStream.getStream()));
      }
      systemAdmin.validateStream(changelogSpec);

      if (storageConfig.getAccessLogEnabled(storeName)) {
        String accesslogStream = storageConfig.getAccessLogStream(systemStream.getStream());
        StreamSpec accesslogSpec =
            new StreamSpec(accesslogStream, accesslogStream, systemStream.getSystem(), maxChangeLogStreamPartitions);
        systemAdmin.createStream(accesslogSpec);
        systemAdmin.validateStream(accesslogSpec);
      }

      systemAdmin.stop();
    });
  }
Creates and validates the changelog streams of a samza job.

@param config the configuration with changelog info.
@param maxChangeLogStreamPartitions the maximum number of changelog stream partitions to create.
samza-core/src/main/java/org/apache/samza/storage/ChangelogStreamManager.java