Kafka DSL - Manual Commit with Message

[muruganmuthu1980]

We are trying to implement manual commit with Spring DSL Kafka with the below code. We couldn't find any reference to it.

We can add consumer porperty as "Auto.commit" as "False", but we wanted to commit the message once we process the message successfully. Can some one help on this?

@Bean
        IntegrationFlow consumer()  {
    KafkaHighLevelConsumerMessageSourceSpec messageSourceSpec = Kafka
                    .inboundChannelAdapter(new ZookeeperConnect(this.kafkaConfig.getZookeeperAddress()))
                    .consumerProperties(
                            props -> props.put("zookeeper.session.timeout.ms", "500").put("zookeeper.sync.time.ms", "250").
                            put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer").
                            put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer").
                            put("auto.offset.reset", "smallest").
                            put("auto.commit.interval.ms", "100")
                            )
                    .addConsumer(this.kafkaConfig.getConsumerGroup(),
                            metadata -> metadata.consumerTimeout(100)
                                    .topicStreamMap(m -> m.put(this.kafkaConfig.getTopicRead(), 1)).maxMessages(1));

            Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer = e -> e.poller(p -> p.fixedDelay(100));
            return IntegrationFlows.from(messageSourceSpec, endpointConfigurer)
                    .<Map<String, List<byte[]>>> handle((payload, headers) -> {
                        payload.entrySet().forEach(e -> processMessage((ConcurrentHashMap<Integer, List<byte[]>>) e.getValue()));
                        return null;
                    }).get();
        }

Version:

<dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-java-dsl</artifactId>
            <version>1.1.0.RELEASE</version>
</dependency>

[artembilan]

Hello, @muruganmuthu1980 !

Thank you for heads up and sorry for delay.
We are busy with release process. 😄

Seems for me there is an SO question on the matter. Yours http://stackoverflow.com/questions/37119409/kafka-dsl-manual-commit-with-message ?

So, looking to this closely I'd say that you have to commit offsets manually:

ConsumerConnector
...
/**
   *  Commit the offsets of all broker partitions connected by this connector.
   */
  public void commitOffsets();
  public void commitOffsets(boolean retryOnFailure);

That is a Kafka Consumer bridge on which the KafkaHighLevelConsumerMessageSource is based.
You can get access to that via KafkaConsumerContext bean. Unfortunately it is registered with some generated name like, org.springframework.integration.kafka.support.KafkaConsumerContext#0 , but if you have only one such a Kafka Consumer (KafkaHighLevelConsumerMessageSourceSpec messageSourceSpec), you can just get that bean by type (or just @Autowired it).

After that you can go like:

kafkaConsumerContext.getConsumerConfiguration(this.kafkaConfig.getConsumerGroup())
                                    .getConsumerConnector()
                                    .commitOffsets();

Let me know that it make sense for you and I'll copy that as an answer to your SO question.

Note: Kafka High Level Consumer has been deprecated in Spring Integration for a while already.
So, consider to switch to KafkaMessageDrivenChannelAdapter.

Plus, play attention that this project is about just Kafka in version >= 0.9 and it fully doesn't relate to Spring Integration, nor to Spring Integration Java DSL.

So, please, be careful to choose the project in the future for proper place to raise an issue.

Although this is fully the SO candidate, I agree that sometime we need a jab to pay attention to the problem.

Thank you and hope I am clear and that will be helpfull

[muruganmuthu1980]

Thanks for your Help. Sorry for the wrong project was chosen. I will try this, thanks really so much for your help.

[muruganmuthu1980]

This solution works perfectly, thanks for your help

[artembilan]

Glad to hear!

Now it's time to react properly on your SO question for my answer 😄

Out of topic. Right now Spring Kafka project is based on Apache Kafka 0.9. Spring Integration Kafka 2.0 is based on this project. Both versions are planed to be released somewhere in the end of May.

There is no KafkaHighLevelConsumerMessageSource, as well as any High Level Consumer in Kafka.
So, consider to switch to a fresh more clear API.

Thank you for your feedback anyway!