StreamManager { public void createStreams(List<StreamSpec> streams) { Multimap<String, StreamSpec> streamsGroupedBySystem = HashMultimap.create(); streams.forEach(streamSpec -> streamsGroupedBySystem.put(streamSpec.getSystemName(), streamSpec)); for (Map.Entry<String, Collection<StreamSpec>> entry : streamsGroupedBySystem.asMap().entrySet()) { String systemName = entry.getKey(); SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(systemName); for (StreamSpec stream : entry.getValue()) { LOGGER.info("Creating stream {} with partitions {} on system {}", new Object[]{stream.getPhysicalName(), stream.getPartitionCount(), systemName}); systemAdmin.createStream(stream); } } } 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 testCreateStreams() { StreamSpec spec1 = new StreamSpec(STREAM1, STREAM1, SYSTEM1); StreamSpec spec2 = new StreamSpec(STREAM2, STREAM2, SYSTEM2); List<StreamSpec> specList = new ArrayList<>(); specList.add(spec1); specList.add(spec2); 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); StreamManager manager = new StreamManager(systemAdmins); manager.createStreams(specList); ArgumentCaptor<StreamSpec> captor = ArgumentCaptor.forClass(StreamSpec.class); verify(admin1).createStream(captor.capture()); assertEquals(STREAM1, captor.getValue().getPhysicalName()); captor = ArgumentCaptor.forClass(StreamSpec.class); verify(admin2).createStream(captor.capture()); assertEquals(STREAM2, captor.getValue().getPhysicalName()); }