public KafkaStreamSpec toKafkaSpec(StreamSpec spec) {
    KafkaStreamSpec kafkaSpec;
    if (spec.isChangeLogStream()) {
      String topicName = spec.getPhysicalName();
      ChangelogInfo topicMeta = changelogTopicMetaInformation.get(topicName);
      if (topicMeta == null) {
        throw new StreamValidationException("Unable to find topic information for topic " + topicName);
      }

      kafkaSpec = new KafkaStreamSpec(spec.getId(), topicName, systemName, spec.getPartitionCount(),
          topicMeta.getReplicationFactor(), topicMeta.getKafkaProperties());
    } else if (spec.isCoordinatorStream()) {
      kafkaSpec =
          new KafkaStreamSpec(spec.getId(), spec.getPhysicalName(), systemName, 1, coordinatorStreamReplicationFactor,
              coordinatorStreamProperties);
    } else if (spec.isCheckpointStream()) {
      Properties checkpointTopicProperties = new Properties();
      checkpointTopicProperties.putAll(spec.getConfig());
      kafkaSpec = KafkaStreamSpec.fromSpec(StreamSpec.createCheckpointStreamSpec(spec.getPhysicalName(), spec.getSystemName()))
              .copyWithReplicationFactor(Integer.parseInt(new KafkaConfig(config).getCheckpointReplicationFactor().get()))
              .copyWithProperties(checkpointTopicProperties);
    } else if (intermediateStreamProperties.containsKey(spec.getId())) {
      kafkaSpec = KafkaStreamSpec.fromSpec(spec);
      Properties properties = kafkaSpec.getProperties();
      properties.putAll(intermediateStreamProperties.get(spec.getId()));
      kafkaSpec = kafkaSpec.copyWithProperties(properties);
    } else {
      kafkaSpec = KafkaStreamSpec.fromSpec(spec);

      // we check if there is a system-level rf config specified, else we use KafkaConfig.topic-default-rf
      int replicationFactorFromSystemConfig = Integer.valueOf(
          new KafkaConfig(config).getSystemDefaultReplicationFactor(spec.getSystemName(),
              KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR()));
      LOG.info("Using replication-factor: {} for StreamSpec: {}", replicationFactorFromSystemConfig, spec);
      return new KafkaStreamSpec(kafkaSpec.getId(), kafkaSpec.getPhysicalName(), kafkaSpec.getSystemName(),
          kafkaSpec.getPartitionCount(), replicationFactorFromSystemConfig, kafkaSpec.getProperties());
    }
    return kafkaSpec;
  }
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