@ 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 .