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);
  }
@ 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
Should be called only by MapR Streams Producer . It creates a topic using KafkaProducer .