@Override
  public boolean createStream(StreamSpec streamSpec) {
    LOG.info("Creating Kafka topic: {} on system: {}", streamSpec.getPhysicalName(), streamSpec.getSystemName());
    final String replFactor = "replication.factor";

    KafkaStreamSpec kafkaStreamSpec = toKafkaSpec(streamSpec);
    String topicName = kafkaStreamSpec.getPhysicalName();

    // create topic.
    NewTopic newTopic = new NewTopic(topicName, kafkaStreamSpec.getPartitionCount(), (short) kafkaStreamSpec.getReplicationFactor());

    // specify the configs
    Map<String, String> streamConfig = new HashMap<>(kafkaStreamSpec.getConfig());
    // HACK - replication.factor is invalid config for AdminClient.createTopics
    if (streamConfig.containsKey(replFactor)) {
      String repl = streamConfig.get(replFactor);
      LOG.warn("Configuration {}={} for topic={} is invalid. Using kSpec repl factor {}",
          replFactor, repl, kafkaStreamSpec.getPhysicalName(), kafkaStreamSpec.getReplicationFactor());
      streamConfig.remove(replFactor);
    }
    newTopic.configs(new MapConfig(streamConfig));
    CreateTopicsResult result = adminClient.createTopics(ImmutableSet.of(newTopic));
    try {
      result.all().get(KAFKA_ADMIN_OPS_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
      if (e instanceof TopicExistsException || e.getCause() instanceof TopicExistsException) {
        LOG.info("Topic {} already exists.", topicName);
        return false;
      }

      throw new SamzaException(String.format("Creation of topic %s failed.", topicName), e);
    }
    LOG.info("Successfully created topic {}", topicName);
    DescribeTopicsResult desc = adminClient.describeTopics(ImmutableSet.of(topicName));
    try {
      TopicDescription td = desc.all().get(KAFKA_ADMIN_OPS_TIMEOUT_MS, TimeUnit.MILLISECONDS).get(topicName);
      LOG.info("Topic {} created with {}", topicName, td);
      return true;
    } catch (Exception e) {
      LOG.error("'Describe after create' failed for topic " + topicName, e);
      return false;
    }
  }
Fetch SystemStreamMetadata for each topic with the consumer
@param topics set of topics to get metadata info for
@return map of topic to SystemStreamMetadata
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java