public StreamPartitionCountMonitor build(Config config, StreamPartitionCountMonitor.Callback callback) {
    Set<SystemStream> inputStreamsToMonitor = new TaskConfig(config).getAllInputStreams();
    if (inputStreamsToMonitor.isEmpty()) {
      throw new SamzaException("Input streams to a job can not be empty.");
    }
    return new StreamPartitionCountMonitor(inputStreamsToMonitor, this.streamMetadataCache, this.metrics,
        new JobConfig(config).getMonitorPartitionChangeFrequency(), callback);
  }
Build {@link StreamPartitionCountMonitor} for input streams.
samza-core/src/main/java/org/apache/samza/coordinator/StreamPartitionCountMonitorFactory.java