public static KafkaStreamSpec fromSpec(StreamSpec originalSpec) {
    if (originalSpec instanceof KafkaStreamSpec) {
      return (KafkaStreamSpec) originalSpec;
    }

    int replicationFactor = Integer.parseInt(originalSpec.getOrDefault(KafkaConfig.TOPIC_REPLICATION_FACTOR(),
        KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR()));

    return new KafkaStreamSpec(originalSpec.getId(),
                               originalSpec.getPhysicalName(),
                               originalSpec.getSystemName(),
                               originalSpec.getPartitionCount(),
                               replicationFactor,
                               mapToProperties(filterUnsupportedProperties(originalSpec.getConfig())));
  }
Converts any StreamSpec to a KafkaStreamSpec.
If the original spec already is a KafkaStreamSpec, it is simply returned.

@param originalSpec  The StreamSpec instance to convert to KafkaStreamSpec.
@return              A KafkaStreamSpec instance.
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaStreamSpec.java