@Override
  public void validateStream(StreamSpec streamSpec) throws StreamValidationException {
    LOG.info("About to validate stream = " + streamSpec);

    String streamName = streamSpec.getPhysicalName();
    SystemStreamMetadata systemStreamMetadata =
        getSystemStreamMetadata(Collections.singleton(streamName)).get(streamName);
    if (systemStreamMetadata == null) {
      throw new StreamValidationException(
          "Failed to obtain metadata for stream " + streamName + ". Validation failed.");
    }

    int actualPartitionCounter = systemStreamMetadata.getSystemStreamPartitionMetadata().size();
    int expectedPartitionCounter = streamSpec.getPartitionCount();
    LOG.info("actualCount=" + actualPartitionCounter + "; expectedCount=" + expectedPartitionCounter);
    if (actualPartitionCounter != expectedPartitionCounter) {
      throw new StreamValidationException(
          String.format("Mismatch of partitions for stream %s. Expected %d, got %d. Validation failed.", streamName,
              expectedPartitionCounter, actualPartitionCounter));
    }
  }
Converts a StreamSpec into a KafkaStreamSpec. Special handling for coordinator and changelog stream.
@param spec a StreamSpec object
@return KafkaStreamSpec object
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java