Consumer get stucked in compacted topic

[Elyahou]

When replaying a compacted topic, with 20 partitions, the consumer get stucked in 5 partitions at some offsets, and don't succeed to read any more messages. All the other partitions were successully replayed...

I succeeded to fix this by manually increasing the consumer group offset for the problematic partitions, after I saw that it was stucked in an offset that was compacted and there was no messages for it in kafka.

Example:

offset	status	remark
1	read	message present
2	read	message present
3	read	message present
4	stucked here	deleted by kafka compaction
5		deleted by kafka compaction
6		deleted by kafka compaction
7		message present

So it works only after I moved the offset of the consumer group to 7.

Some remarks:

• There is other offsets that was compacted by kafka and there was no issue there
• The kafka console consumer succeeded to consume these messages continuously without any issue
• I tried all kind of configuration changes on socket_timeout, connect_timeout and max_wait_time., but it did not help.

• Version of Ruby: 2.3.5
• Version of Kafka: 0.11.0.2
• Version of ruby-kafka: 0.5.4

Code of the consumer

require "kafka"

kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
consumer = kafka.consumer(group_id: "my-consumer")
consumer.subscribe("greetings")
trap("TERM") { consumer.stop }

consumer.each_message do |message|
  puts message.topic, message.partition
  puts message.offset, message.key, message.value
end

[dasch]

Good job reproducing the bug – it sounds like it's related to how fetches work when asking for messages at a starting offset that no longer exists. It used to be that the Kafka would just start at the next existing offset, but that may have changed. Would you be willing to dive into the Java client source and see how they handle this?

[Elyahou]

I am not sure it is the issue, because in the case I reset the consumer to offset 2 in the precedent example, the consumer will receive only message 2 and 3 in its batch, and will not receive further messages... So it is not only an issue when the fetch operation is on an empty offset

[dasch]

It could be a problem with the next offset computation – that's rather complicated.

[dasch]

Would you be willing to bundle open ruby-kafka and sprinkle debug statements into the code?

[Elyahou]

@dasch Yes of course I can

[Elyahou]

I continued my analysis of this issue, I think I maybe understood it...

Kafka new format is RecordBatch and if this batch is compacted entirely, it will still be returned to the consumer with lastOffsetDelta field, that will allow the consumer to increment its offset to get next messages...

When using old consumer like we do with ruby-kafka, this batch is converted to old format before being returned to the consumer. And so an empty MessageSet is returned. But there is no information about next offsets, and so the consumer get stucked...

The only information that could maybe be used is highwater_mark_offset that indicate there is still offsets that should be fetched... But so, the consumer will need to increment by 1 and fetch until it get some messages... That do not sound good...

[dasch]

sigh

The Kafka protocol is getting increasingly complex... Surely the brokers provide backwards compatibility guarantees for old Java clients?

[0x2c7]

@Elyahou Hi, did the fix in #625 work, or you still face this issue?

[Elyahou]

Thanks @nguyenquangminh0711 I tested against 0.7.3 version of ruby-kafka, and the bug was fixed :)

[Elyahou]

Thanks @nguyenquangminh0711 I tested against 0.7.3 version of ruby-kafka, and the bug was fixed :)

[Elyahou]

Thanks @nguyenquangminh0711 I tested against 0.7.3 version, and it seems this is now fixed :)