StreamManager { public void clearStreamsFromPreviousRun(Config prevConfig) { try { ApplicationConfig appConfig = new ApplicationConfig(prevConfig); LOGGER.info("run.id from previous run is {}", appConfig.getRunId()); StreamConfig streamConfig = new StreamConfig(prevConfig); Set<StreamSpec> intStreams = streamConfig.getStreamIds().stream() .filter(streamConfig::getIsIntermediateStream) .map(id -> new StreamSpec(id, streamConfig.getPhysicalName(id), streamConfig.getSystem(id))) .collect(Collectors.toSet()); intStreams.forEach(stream -> { LOGGER.info("Clear intermediate stream {} in system {}", stream.getPhysicalName(), stream.getSystemName()); systemAdmins.getSystemAdmin(stream.getSystemName()).clearStream(stream); }); TaskConfig taskConfig = new TaskConfig(prevConfig); taskConfig.getCheckpointManager(new MetricsRegistryMap()).ifPresent(CheckpointManager::clearCheckpoints); StorageConfig storageConfig = new StorageConfig(prevConfig); for (String store : storageConfig.getStoreNames()) { String changelog = storageConfig.getChangelogStream(store).orElse(null); if (changelog != null) { LOGGER.info("Clear store {} changelog {}", store, changelog); SystemStream systemStream = StreamUtil.getSystemStreamFromNames(changelog); StreamSpec spec = StreamSpec.createChangeLogStreamSpec(systemStream.getStream(), systemStream.getSystem(), 1); systemAdmins.getSystemAdmin(spec.getSystemName()).clearStream(spec); } } } catch (Exception e) { LOGGER.warn("Fail to clear internal streams from previous run. Please clean up manually.", e); } } StreamManager(Config config); @VisibleForTesting  StreamManager(SystemAdmins systemAdmins); void createStreams(List<StreamSpec> streams); void start(); void stop(); void clearStreamsFromPreviousRun(Config prevConfig); static String createUniqueNameForBatch(String stream, Config config);  }
@Test public void testClearStreamsFromPreviousRun() { SystemAdmin admin1 = mock(SystemAdmin.class); SystemAdmin admin2 = mock(SystemAdmin.class); SystemAdmins systemAdmins = mock(SystemAdmins.class); when(systemAdmins.getSystemAdmin(SYSTEM1)).thenReturn(admin1); when(systemAdmins.getSystemAdmin(SYSTEM2)).thenReturn(admin2); String runId = "123"; Map<String, String> config = new HashMap<>(); config.put(ApplicationConfig.APP_RUN_ID, "123"); config.put(ApplicationConfig.APP_MODE, ApplicationConfig.ApplicationMode.BATCH.name()); config.put("streams.stream-1.samza.system", SYSTEM1); config.put("streams.stream-1.samza.physical.name", STREAM1 + "-" + runId); config.put("streams.stream-1.samza.intermediate", "true"); config.put("task.checkpoint.factory", MockCheckpointManagerFactory.class.getName()); config.put("stores.test-store.factory", "dummyfactory"); config.put("stores.test-store.changelog", SYSTEM2 + "." + STREAM2); StreamManager manager = new StreamManager(systemAdmins); manager.clearStreamsFromPreviousRun(new MapConfig(config)); ArgumentCaptor<StreamSpec> captor = ArgumentCaptor.forClass(StreamSpec.class); verify(admin1).clearStream(captor.capture()); assertEquals(captor.getValue().getPhysicalName(), STREAM1 + "-" + runId); captor = ArgumentCaptor.forClass(StreamSpec.class); verify(admin2).clearStream(captor.capture()); assertEquals(captor.getValue().getPhysicalName(), STREAM2 + "-" + runId); verify(checkpointManager, times(1)).clearCheckpoints(); }