Spring-Integration XML to Java

Question

How can I convert this code into Java config?

<int-kafka:outbound-channel-adapter
        id="mainOutboundChannelAdapter"
        kafka-producer-context-ref="kafkaProducerContext"
        channel="mainOutboundTopicChanel">
</int-kafka:outbound-channel-adapter>

Answer 1

Yes, you can. Please, find the latest Spring Integration Java DSL:

Your case may looks like:

@Bean
public IntegrationFlow sendToKafkaFlow(String serverAddress) {
    return f -> f.<String>split(p -> FastList.newWithNValues(100, () -> p), null)
            .handle(kafkaMessageHandler(serverAddress));
}

private KafkaProducerMessageHandlerSpec kafkaMessageHandler(String serverAddress) {
    return Kafka.outboundChannelAdapter(props -> props.put("queue.buffering.max.ms", "15000"))
            .messageKey(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER))
            .addProducer(TEST_TOPIC, serverAddress, this::producer);
}

private void producer(KafkaProducerMessageHandlerSpec.ProducerMetadataSpec metadata) {
    metadata.async(true)
            .batchNumMessages(10)
            .valueClassType(String.class)
            .<String>valueEncoder(String::getBytes)
            .keyEncoder(new IntEncoder(null));
}

UPDATE Without Lambdas, but still Spring Integration:

@Bean
@ServiceActivator(inputChannel = "mainOutboundTopicChanel")
public MessageHandler kafkaProducer() {
    return new KafkaProducerMessageHandler<String, String>(kafkaProducerContext());
}

@Bean
public KafkaProducerContext<String, String> kafkaProducerContext() {
    KafkaProducerContext<String, String> kafkaProducerContext = new KafkaProducerContext<String, String>();
    ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC);
    producerMetadata.setValueClassType(String.class);
    producerMetadata.setKeyClassType(String.class);
    Encoder<String> encoder = new StringEncoder<String>();
    producerMetadata.setValueEncoder(encoder);
    producerMetadata.setKeyEncoder(encoder);
    producerMetadata.setAsync(true);
    Properties props = new Properties();
    props.put("queue.buffering.max.ms", "15000");
    ProducerFactoryBean<String, String> producer =
            new ProducerFactoryBean<String, String>(producerMetadata, kafkaRule.getBrokersAsString(), props);
    ProducerConfiguration<String, String> config =
            new ProducerConfiguration<String, String>(producerMetadata, producer.getObject());
        kafkaProducerContext.setProducerConfigurations(Collections.singletonMap(TOPIC, config));
    return kafkaProducerContext;
}

And don't forget to add @EnableIntegration alongside with your @Configuration.

For the future: any XML tag in Spring is parsed by some NamespaceHandler, e.g. in this case it is KafkaNamespaceHandler. Reading its source code we can find these lines:

registerBeanDefinitionParser("outbound-channel-adapter", new KafkaOutboundChannelAdapterParser());
        registerBeanDefinitionParser("producer-context", new KafkaProducerContextParser());

When we go to the KafkaOutboundChannelAdapterParser and see that it populates a BeanDefinition:

final BeanDefinitionBuilder kafkaProducerMessageHandlerBuilder =
                                BeanDefinitionBuilder.genericBeanDefinition(KafkaProducerMessageHandler.class);

and so on by source code.

UPDATE 2

The Consumer part:

@Bean
@InboundChannelAdapter(value = "fromKafkaChannel",
    poller = @Poller(fixedRate = "10", maxMessagesPerPoll = "1"))
public MessageSource<Map<String, Map<Integer, List<Object>>>> kafkaMessageSource() {
    return new KafkaHighLevelConsumerMessageSource<String, String>();
}

@Bean
public KafkaConsumerContext<String, String> kafkaConsumerContext() {
    KafkaConsumerContext<String, String> kafkaConsumerContext = new KafkaConsumerContext<String, String>();
    .....
    kafkaConsumerContext.setConsumerConfigurations(map);
    return kafkaConsumerContext;
}