Spring Kafka don't respect max.poll.records with strange behavior

user_13814928 0
Viewed 5299

Well, I'm trying the following scenario:

  1. In application.properties set max.poll.records to 50.
  2. In application.properties set enable-auto-commit=false and ack-mode to manual.
  3. In my method added @KafkaListener, but don't commit any message, just read, log but don't make an ACK.

Actually, in my Kafka topic, I have 500 messages to be consumed, so I'm expecting the following behavior:

  1. Spring Kafka poll() 50 messages (offset 0 to 50).
  2. As I said, I didn't commit anything, just log the 50 messages.
  3. In the next Spring Kafka poll() invocation, get the same 50 messages (offset 0 to 50), as step 1. Spring Kafka, in my understanding, should continue in this loop (step 1-3) reading always the same messages.

But what happens is the following:

  1. Spring Kafka poll() 50 messages (offset 0 to 50).
  2. As I said, I didn't commit anything, just log the 50 messages.
  3. In the next Spring Kafka poll() invocation, get the NEXT 50 messages, different from step 1 (offset 50 to 100).

Spring Kafka reads the 500 messages, in blocks of 50 messages, but don't commit anything. If I shut down the application and start again, the 500 messages are received again.

So, my doubts:

  1. If I configured the max.poll.recors to 50, how spring Kafka get the next 50 records if I didn't commit anything? I understand the poll() method should return the same records.
  2. Does Spring Kafka have some cache? If yes, this can be a problem if I get 1million records in cache without commit.
spring-kafka apache-kafka spring
2 Answers

Your first question:

If I configured the max.poll.recors to 50, how spring Kafka get the next 50 records if I didn't commit anything? I understand the poll() method should return the same records.

First, to make sure that you did not commit anything, you must make sure that you understand the following 3 parameters, which i believe you understood.

  • ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, set it to false(which is also the recommended default). And if it is set to false, take note that auto.commit.interval.ms becomes irrelevant. Check out this documentation:

Because the listener container has it’s own mechanism for committing offsets, it prefers the Kafka ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG to be false. Starting with version 2.3, it unconditionally sets it to false unless specifically set in the consumer factory or the container’s consumer property overrides.

  • factory.getContainerProperties().setAckMode(AckMode.MANUAL); You take the responsibility to acknowledge. (Ignored when transactions are being used) and ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG can't be true.

  • factory.getContainerProperties().setSyncCommits(true/false); Set whether or not to call consumer.commitSync() or commitAsync() when the container is responsible for commits. Default true. This is responsible for sync-ing with Kafka, nothing else, if set to true, that call will block until Kafka responds.

Secondly, no the consumer poll() will not return the same records. For the current running consumer, it tracks its offset in memory with some internal index, we don't have to care about committing offsets. Please also see @GaryRussell s explanation here.

In short, he explained:

Once the records have been returned by the poll (and offsets not committed), they won't be returned again unless you restart the consumer or perform seek() operations on the consumer to reset the offset to the unprocessed ones.


Your second question:

Does Spring Kafka have some cache? If yes, this can be a problem if I get 1million records in cache without commit.

There is no "cache", it's all about offsets and commits, explanation as per above.



Now to achieve what you wanted to do, you can consider doing 2 things after fetching the first 50 records, i.e for the next poll():

  • Either, re-start the container programatically
  • Or call consumer.seek(partition, offset);


BONUS:
Whatever configuration you choose, you can always check out the results, by looking at the LAG column of this output:

kafka-consumer-groups.bat --bootstrap-server localhost:9091 --describe --group your_group_name
user_49826910

Consumer not committing the offset will have impact only in situations like:

  • Your consumer crashed after reading 200 messages, when you restart it, it will start again from 0.
  • Your consumer is no longer assigned a partition.

So in a perfect world, you don't need to commit at all and it will consume all the messages because consumer first asks for 1-50,then 51-100.

But if the consumer crashed, nobody knows what was the offset that consumer read. If the consumer had committed the offset, when it is restarted it can check the offset topic to see where the crashed consumer left and start from there.

max.poll.records defines how many records to fetch at one go but it does not define which records to fetch.

user_78756230
Related