ChangelogStreamManager { public static void createChangelogStreams(Config config, int maxChangeLogStreamPartitions) { 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(); SystemConfig systemConfig = new SystemConfig(config); storeNameSystemStreamMapping.forEach((storeName, systemStream) -> { SystemAdmin systemAdmin = systemConfig.getSystemAdmin(systemStream.getSystem()); 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(); }); } ChangelogStreamManager(MetadataStore metadataStore); Map<TaskName, Integer> readPartitionMapping(); void writePartitionMapping(Map<TaskName, Integer> changelogEntries); void updatePartitionMapping(Map<TaskName, Integer> prevChangelogEntries,
      Map<TaskName, Integer> newChangelogEntries); static void createChangelogStreams(Config config, int maxChangeLogStreamPartitions);  }
@Test public void createChangelogStreams() { Map<String, String> map = new ImmutableMap.Builder<String, String>() .put("stores.store0.factory", "factory.class") .put("stores.store0.changelog", SYSTEM + "." + STREAM) .put("stores.store0.accesslog.enabled", "true") .put(String.format("systems.%s.samza.factory", SYSTEM), MockSystemAdminFactory.class.getName()) .build(); Config config = new MapConfig(map); ChangelogStreamManager.createChangelogStreams(config, MAX_CHANGELOG_STREAM_PARTITIONS); }