private int getPartitionCount(SystemStream systemStream) {
    return PARTITION_COUNT_CACHE.computeIfAbsent(systemStream, ss -> {
      SystemStreamMetadata metadata = metadataCache.getSystemStreamMetadata(ss, true);
      if (metadata == null) {
        throw new SamzaException("Unable to find metadata for stream " + systemStream);
      }
      return metadata.getSystemStreamPartitionMetadata().size();
    });
  }
This is a helper class to send control messages to an intermediate stream
samza-core/src/main/java/org/apache/samza/operators/impl/ControlMessageSender.java