Retry upon processing failures processes message twice

[mensfeld]

• Version of Ruby: ruby 2.4.2p198 (2017-09-14 revision 59899) [x86_64-linux]
• Version of Kafka: 0.10.2.0
• Version of ruby-kafka: master

Fixed on master? no

Required Kafka cluster setup

• Single Kafka node
• Single topic named 'a') with single partition (our favorite setup :D)

Steps to reproduce

require 'ruby-kafka'
kafka = Kafka.new(seed_brokers: ['kafka://127.0.0.1:9092'])
# Rand is on purpose just to prove, that the same message is being fetched
kafka.deliver_message(rand.to_s, topic: 'a')
consumer = kafka.consumer(group_id: 'consumer-name')
consumer.subscribe('a')

begin
consumer.each_message(automatically_mark_as_processed: true) do |message|
  p message
  raise StandardError
end
rescue
end

consumer.each_message(automatically_mark_as_processed: true) do |message|
  p message
end

Expected outcome

Ruby-Kafka should reconsume message once, not twice.

#<Kafka::FetchedMessage:0x0000000230cc60 @value="0.8970946704093968", @key=nil, @topic="a", @partition=0, @offset=0, @create_time=2017-10-23 17:27:42 +0200>
#<Kafka::FetchedMessage:0x00000002307558 @value="0.8970946704093968", @key=nil, @topic="a", @partition=0, @offset=0, @create_time=2017-10-23 17:27:42 +0200>

The problem occurs for both each_message and each_batch

Actual outcome

Re-consumption happens twice instead of once

#<Kafka::FetchedMessage:0x0000000230cc60 @value="0.8970946704093968", @key=nil, @topic="a", @partition=0, @offset=0, @create_time=2017-10-23 17:27:42 +0200>
#<Kafka::FetchedMessage:0x00000002307558 @value="0.8970946704093968", @key=nil, @topic="a", @partition=0, @offset=0, @create_time=2017-10-23 17:27:42 +0200>
#<Kafka::FetchedMessage:0x000000022dea40 @value="0.8970946704093968", @key=nil, @topic="a", @partition=0, @offset=0, @create_time=2017-10-23 17:27:42 +0200>

[dasch]

@mensfeld is this still the case on master?

[mensfeld]

@dasch yes:

#<Kafka::FetchedMessage:0x000000012118c0 @message=#<Kafka::Protocol::Message:0x00000001211cf8 @key=nil, @value="0.8360789393926844", @codec_id=0, @offset=2, @create_time=2018-01-09 11:16:59 +0100, @bytesize=18>, @topic="a", @partition=0>
#<Kafka::FetchedMessage:0x000000011ea428 @message=#<Kafka::Protocol::Message:0x000000011ea860 @key=nil, @value="0.8360789393926844", @codec_id=0, @offset=2, @create_time=2018-01-09 11:16:59 +0100, @bytesize=18>, @topic="a", @partition=0>

[dasch]

Hmm.

[mensfeld]

@dasch - sorry had conflicts and didn't notice, that my fork didn't get synced. Fixed.

[dasch]

🎉