IntegrationFlow for Kafka Message error while configureListenerContainer

Question

I am trying to use IntegrationFlow for kafka to pass message received from Kafka to channel.

Below is my working code:-

    @Bean
    public MessageChannel fromKafka() {
        return new DirectChannel();
    }

    @Bean
    public IntegrationFlow topic1ListenerFromKafkaFlow1() throws Exception {
/*        return IntegrationFlows
                .from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
                                KafkaMessageDrivenChannelAdapter.ListenerMode.record, kafkaTopic)
                        .configureListenerContainer( c ->  c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)
                                        .id("topic1ListenerContainer"))
                        .recoveryCallback(new ErrorMessageSendingRecoverer(messageFromKafka(),
                                new RawRecordHeaderErrorMessageStrategy()))
                        .retryTemplate(new RetryTemplate())
                        .filterInRetry(true))
                .filter(Message.class, m ->
                                m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
                        f -> f.throwExceptionOnRejection(true))
                .<String, String>transform(String::toUpperCase)
                .channel(c -> c.queue("listeningFromKafkaResults1"))
                .get();*/



        return IntegrationFlows
                .from(Kafka.messageDrivenChannelAdapter(listener(), KafkaMessageDrivenChannelAdapter.ListenerMode.record))
                .channel("fromKafka")
                .get();
    }





    @Bean("listenerkafka")
    public KafkaMessageListenerContainer<String, String> listener() throws Exception {
        ContainerProperties properties = new ContainerProperties(kafkaTopic1);
        properties.setGroupId("kafka-test");
        return new KafkaMessageListenerContainer<>(consumerFactory, properties);
    }

    @ServiceActivator(inputChannel="fromKafka", outputChannel = "somechannel")
    public Message<CreatRequest> fromKafka(Message<?> msg) throws JsonProcessingException {
        CreatRequest  creatRequest =  objectMapper.readValue(msg.getPayload().toString(), CreatRequest.class);
        Message<CreatRequest> message= MessageBuilder.withPayload(creatRequest).build();
        logger.info("Inside fromKafka " + message);
        return  message;
    }

Issue which I am facing is commented code doesn't work inside topic1ListenerFromKafkaFlow1. Here I am not able to find c.ackMode(AbstractMessageListenerContainer.AckMode.MANUAL)

As it is showing compile time error ackmode not recognised. Can you please correct me where i am going wrong.

Also I need to pass this flow in another thread and not in main thread.

Answer 1

Use the Kafka message-driven channel adapter instead:

https://docs.spring.io/spring-integration/docs/current/reference/html/kafka.html#kafka-inbound

However, with two adapters on the same channel the requests will be round-robin distributed between them. If you want both to receive the message, you need a PublishSubscribeChannel.