sdc-kafka_0_9_mapr_5_2/src/main/java/com/streamsets/pipeline/kafka/impl/MapR52StreamsValidationUtil09.java
@Override
  public void createTopicIfNotExists(String topic, Map<String, Object> kafkaClientConfigs, String metadataBrokerList)
      throws StageException {
    // check stream path and topic
    if (topic.startsWith("/") && topic.contains(":")) {
      String[] path = topic.split(":");
      if (path.length != 2) {
        // Stream topic has invalid format. Record will be sent to error
        throw new StageException(MapRStreamsErrors.MAPRSTREAMS_21, topic);
      }
      String streamPath = path[0];

      if (!streamCache.contains(streamPath)) {
        // This pipeline sees this stream path for the 1st time
        Configuration conf = new Configuration();
        kafkaClientConfigs.forEach((k, v) -> {
          conf.set(k, v.toString());
        });
        Admin streamAdmin = null;
        try {
          streamAdmin = Streams.newAdmin(conf);
          // Check if the stream path exists already
          streamAdmin.countTopics(streamPath);
          streamCache.add(streamPath);
        } catch (TableNotFoundException e) {
          LOG.debug("Stream not found. Creating a new stream: " + streamPath);
          try {
            streamAdmin.createStream(streamPath, Streams.newStreamDescriptor());
            streamCache.add(streamPath);
          } catch (IOException ioex) {
            throw new StageException(MapRStreamsErrors.MAPRSTREAMS_22, streamPath, e.getMessage(), e);
          }
        } catch (IOException | IllegalArgumentException e) {
          throw new StageException(MapRStreamsErrors.MAPRSTREAMS_23, e.getMessage(), e);
        } finally {
          if (streamAdmin != null) {
            streamAdmin.close();
          }
        }
      }
    }

    // Stream topic can be created through KafkaProducer if Stream Path exists already
    KafkaProducer<String, String> kafkaProducer = createProducerTopicMetadataClient(kafkaClientConfigs);
    kafkaProducer.partitionsFor(topic);
  }
@Override
  public void createTopicIfNotExists(String topic, Map<String, Object> kafkaClientConfigs, String metadataBrokerList)
      throws StageException {
    // check stream path and topic
    if (topic.startsWith("/") && topic.contains(":")) {
      String[] path = topic.split(":");
      if (path.length != 2) {
        // Stream topic has invalid format. Record will be sent to error
        throw new StageException(MapRStreamsErrors.MAPRSTREAMS_21, topic);
      }
      String streamPath = path[0];

      if (!streamCache.contains(streamPath)) {
        // This pipeline sees this stream path for the 1st time
        Configuration conf = new Configuration();
        kafkaClientConfigs.forEach((k, v) -> {
          conf.set(k, v.toString());
        });
        Admin streamAdmin = null;
        try {
          streamAdmin = Streams.newAdmin(conf);
          // Check if the stream path exists already
          streamAdmin.countTopics(streamPath);
          streamCache.add(streamPath);
        } catch (TableNotFoundException e) {
          LOG.debug("Stream not found. Creating a new stream: " + streamPath);
          try {
            streamAdmin.createStream(streamPath, Streams.newStreamDescriptor());
            streamCache.add(streamPath);
          } catch (IOException ioex) {
            throw new StageException(MapRStreamsErrors.MAPRSTREAMS_22, streamPath, e.getMessage(), e);
          }
        } catch (IOException | IllegalArgumentException e) {
          throw new StageException(MapRStreamsErrors.MAPRSTREAMS_23, e.getMessage(), e);
        } finally {
          if (streamAdmin != null) {
            streamAdmin.close();
          }
        }
      }
    }

    // Stream topic can be created through KafkaProducer if Stream Path exists already
    KafkaProducer<String, String> kafkaProducer = createProducerTopicMetadataClient(kafkaClientConfigs);
    kafkaProducer.partitionsFor(topic);
  }
Should be called only by MapR Streams Producer. It creates a topic using KafkaProducer.
@param topic
@param kafkaClientConfigs
@param metadataBrokerList
@throws StageException