Does Spring integeration kafka support dynamic topic creation

Question

I am a newbie to spring integeration kafka and I understand kafka-oubound-channel adapter. But is there a way i can create topics progamattically without having to setup in context xml?

ie: based on my message to transformer i would like to post message to kafka topic created for this message type.

Update:

Below is what I ended up doing. Will welcome any better solution.

    <int:channel id="inputForSolrPublish"></int:channel>

<int:service-activator input-channel="inputForSolrPublish"
    ref="solrMasterListRouter" >

-->

 private void postMessageToMasterSpecifcTopics(final List<String> topicNames,
                                              final String brokerList,
                                              final Message<?> message) throws Exception {

    for (String topicName : topicNames) {
        createProducerContext(topicName,
                              brokerList).send(topicName,
                                               message.getHeaders()
                                                      .get(KafkaHeaders.MESSAGE_KEY),
                                               message);

    }

}

private KafkaProducerContext<String, String> createProducerContext(final String topicName,
                                                                   final String brokerList) throws Exception {
    KafkaProducerContext<String, String> kafkaProducerContext = new KafkaProducerContext<String, String>();
    AvroReflectDatumBackedKafkaEncoder<String> kafkaReflectionEncoder = new AvroReflectDatumBackedKafkaEncoder<>(String.class);
    AvroSpecificDatumBackedKafkaEncoder<String> kafkaSpecificEncoder = new AvroSpecificDatumBackedKafkaEncoder<>(String.class);
    // Encoder<String> encoder = new
    // org.springframework.integration.kafka.serializer.common.StringEncoder<String>();

    ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(topicName);
    producerMetadata.setValueClassType(String.class);
    producerMetadata.setKeyClassType(String.class);
    producerMetadata.setValueEncoder(kafkaSpecificEncoder);
    producerMetadata.setKeyEncoder(kafkaReflectionEncoder);
    producerMetadata.setAsync(true);

    Properties props = buildProducerConfigProperties();
    ProducerFactoryBean<String, String> producer = new ProducerFactoryBean<String, String>(producerMetadata,
                                                                                           brokerList,
                                                                                           props);
    ProducerConfiguration<String, String> config = new ProducerConfiguration<String, String>(producerMetadata,
                                                                                             producer.getObject());
    kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(topicName,
                                                                            config));
    return kafkaProducerContext;
}

private Properties buildProducerConfigProperties() {
    Properties props = new Properties();
    props.put("topic.metadata.refresh.interval.ms",
              "3600000");
    props.put("message.send.max.retries",
              "5");
    props.put("tsend.buffer.bytes",
              "5242880");
    return props;

}

Hi @Harshjgs, Please tell me the usage of KafkaHeaders.MESSAGE_KEY field. I am not able to find its purpose. — rahul

Artem Bilan Artem Bilan · Accepted Answer · 2015-03-13T08:46:15

Yes, you can do that at runtime. See TopicUtils.ensureTopicCreated.

You can add it like a <service-activator> as one more subscriber (the first one) to the <publish-subscribe-channel> for sending messages. Something like this:

<publish-subscribe-channel id="sendMessageToKafkaChannel"/>

<service-activator input-channel="sendMessageToKafkaChannel" output-channel="nullChannel" order="1"
   ref="creatTopicService" method="creatTopic"/>

<int-kafka:outbound-channel-adapter channel="sendMessageToKafkaChannel" order="2"/>

Accept in that creatTopic the whole message and extract all required parameter from the message or during injection phase, e.g. inject ZookeeperConnect to extract getZkConnect() for the first zkAddress ensureTopicCreated argument.

But you should understand that you can't have an <int-kafka:message-driven-channel-adapter> without existing topic on the Kafka. So, I'm not sure how you are going to get deal with messages in those dynamically-created topic in the future. Although <int-kafka:inbound-channel-adapter> may work for that case...

Does Spring integeration kafka support dynamic topic creation

1 Answers