protected void createKafkaTopics ( Set < String > topics ) { Map < String , Object > adminConfigs = new HashMap <> ( ) ; adminConfigs . put ( AdminClientConfig . BOOTSTRAP_SERVERS_CONFIG , "localhost:" + kafkaServerPort ) ; try ( AdminClient admin = AdminClient . create ( adminConfigs ) ) { List < NewTopic > newTopics = topics . stream ( ) . map ( t -> new NewTopic ( t , partitions , ( short ) 1 ) ) . collect ( Collectors . toList ( ) ) ; CreateTopicsResult createTopics = admin . createTopics ( newTopics ) ; try { createTopics . all ( ) . get ( ) ; } catch ( Exception e ) { log . warn ( "Failed to create Kafka topics" , e ) ; } } }
Create topics on embedded Kafka server .