private static Set<SystemStreamPartition> getMatchedInputStreamPartitions(Config config,
      StreamMetadataCache streamMetadataCache) {
    Set<SystemStreamPartition> allSystemStreamPartitions = getInputStreamPartitions(config, streamMetadataCache);
    JobConfig jobConfig = new JobConfig(config);
    Optional<String> sspMatcherClassName = jobConfig.getSSPMatcherClass();
    if (sspMatcherClassName.isPresent()) {
      String sspMatcherConfigJobFactoryRegex = jobConfig.getSSPMatcherConfigJobFactoryRegex();
      Optional<String> streamJobFactoryClass = jobConfig.getStreamJobFactoryClass();
      if (streamJobFactoryClass.isPresent() && Pattern.matches(sspMatcherConfigJobFactoryRegex,
          streamJobFactoryClass.get())) {
        LOG.info(String.format("before match: allSystemStreamPartitions.size = %s", allSystemStreamPartitions.size()));
        SystemStreamPartitionMatcher sspMatcher =
            ReflectionUtil.getObj(sspMatcherClassName.get(), SystemStreamPartitionMatcher.class);
        Set<SystemStreamPartition> matchedPartitions = sspMatcher.filter(allSystemStreamPartitions, config);
        // Usually a small set hence ok to log at info level
        LOG.info(String.format("after match: matchedPartitions = %s", matchedPartitions));
        return matchedPartitions;
      }
    }
    return allSystemStreamPartitions;
  }
Builds the input {@see SystemStreamPartition} based upon the {@param config} defined by the user.
@param config configuration to fetch the metadata of the input streams.
@param streamMetadataCache required to query the partition metadata of the input streams.
@return the input SystemStreamPartitions of the job.
samza-core/src/main/java/org/apache/samza/coordinator/JobModelCalculator.java