StreamManager { Map<String, Integer> getStreamPartitionCounts(String systemName, Set<String> streamNames) { Map<String, Integer> streamToPartitionCount = new HashMap<>(); SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(systemName); if (systemAdmin == null) { throw new SamzaException(String.format("System %s does not exist.", systemName)); } Map<String, SystemStreamMetadata> streamToMetadata = systemAdmin.getSystemStreamMetadata(streamNames); streamToMetadata.forEach((stream, data) -> streamToPartitionCount.put(stream, data.getSystemStreamPartitionMetadata().size())); return streamToPartitionCount; } 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 testGetStreamPartitionCounts() { SystemAdmin admin1 = mock(SystemAdmin.class); SystemAdmins systemAdmins = mock(SystemAdmins.class); when(systemAdmins.getSystemAdmin(SYSTEM1)).thenReturn(admin1); Map<String, SystemStreamMetadata> map = new HashMap<>(); SystemStreamMetadata meta1 = mock(SystemStreamMetadata.class); Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitions = new HashMap<>(); partitions.put(new Partition(0), null); when(meta1.getSystemStreamPartitionMetadata()).thenReturn(partitions); map.put(STREAM1, meta1); SystemStreamMetadata meta2 = mock(SystemStreamMetadata.class); Map<Partition, SystemStreamMetadata.SystemStreamPartitionMetadata> partitions2 = new HashMap<>(); partitions2.put(new Partition(0), null); partitions2.put(new Partition(1), null); when(meta2.getSystemStreamPartitionMetadata()).thenReturn(partitions2); map.put(STREAM2, meta2); when(admin1.getSystemStreamMetadata(anyObject())).thenReturn(map); Set<String> streams = new HashSet<>(); streams.add(STREAM1); streams.add(STREAM2); StreamManager manager = new StreamManager(systemAdmins); Map<String, Integer> counts = manager.getStreamPartitionCounts(SYSTEM1, streams); assertTrue(counts.get(STREAM1).equals(1)); assertTrue(counts.get(STREAM2).equals(2)); }