{"action":"created","issue":{"url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/280","repository_url":"https://api.github.com/repos/zendesk/ruby-kafka","labels_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/280/labels{/name}","comments_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/280/comments","events_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/280/events","html_url":"https://github.com/zendesk/ruby-kafka/issues/280","id":175016821,"number":280,"title":"Committed offsets are shifted by one","user":{"login":"tulios","id":33231,"avatar_url":"https://avatars.githubusercontent.com/u/33231?v=3","gravatar_id":"","url":"https://api.github.com/users/tulios","html_url":"https://github.com/tulios","followers_url":"https://api.github.com/users/tulios/followers","following_url":"https://api.github.com/users/tulios/following{/other_user}","gists_url":"https://api.github.com/users/tulios/gists{/gist_id}","starred_url":"https://api.github.com/users/tulios/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tulios/subscriptions","organizations_url":"https://api.github.com/users/tulios/orgs","repos_url":"https://api.github.com/users/tulios/repos","events_url":"https://api.github.com/users/tulios/events{/privacy}","received_events_url":"https://api.github.com/users/tulios/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/zendesk/ruby-kafka/labels/bug","name":"bug","color":"fc2929"}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2016-09-05T08:08:24Z","updated_at":"2016-09-05T08:30:13Z","closed_at":null,"body":"Hi, after using the consumer for a while I've caught a difference from other libraries. In my company, we use [burrow](https://github.com/linkedin/Burrow) to monitor our kafka consumers (a lag checker from linkedin) and with ruby-kafka it reports one message lagging in each partition. I noticed it because we have some topics with seasonal load, topics that receive messages in one period of the day and not constantly, and this is really clear on those.\r\n\r\nI have started debugging burrow to see if the problem was there but after some checks I concluded that burrow was correct.\r\n\r\n* Version of Ruby:\r\n2.3.1\r\n\r\n* Version of Kafka:\r\n0.9.0.1\r\n\r\n* Version of ruby-kafka:\r\n0.3.15.beta3\r\n\r\n###### Steps to reproduce\r\n\r\nRegular ruby-kafka consumer group\r\n\r\n```ruby\r\nrequire 'kafka'\r\nkafka = Kafka.new(seed_brokers: ['localhost:9092'])\r\nconsumer = kafka.consumer(group_id: 'test1')\r\ntrap('QUIT') { consumer.stop }\r\n\r\nconsumer.subscribe('test')\r\nconsumer.each_message { |message| puts(message.inspect) }\r\n```\r\n\r\n###### Expected outcome\r\n\r\nComparing ruby-kafka and kafkacat consumers, for example, we have the following results:\r\n\r\nUsing a new topic with two partitions and publishing one message (the first message)\r\n\r\nburrow topic offsets from `/v2/kafka/local/topic/topic-name`\r\n```json\r\n{\r\n  \"error\": false,\r\n  \"message\": \"broker topic offsets returned\",\r\n  \"offsets\": [\r\n    0,\r\n    1\r\n  ]\r\n}\r\n```\r\n\r\nkafkacat consumer group offsets:\r\n(running with `kafkacat -C -b localhost:9092 -o beginning -G kafkacat topic-name`)\r\nburrow consumer group offsets from `/v2/kafka/local/consumer/kafkacat/topic/topic-name`\r\n\r\n```json\r\n{\r\n  \"error\": false,\r\n  \"message\": \"consumer group topic offsets returned\",\r\n  \"offsets\": [\r\n    -1,\r\n    1\r\n  ]\r\n}\r\n```\r\n\r\n###### Actual outcome\r\n\r\nruby-kafka consumer group offsets:\r\nburrow consumer group offsets from `/v2/kafka/local/consumer/ruby-kafka/topic/topic-name`\r\n\r\n```json\r\n{\r\n  \"error\": false,\r\n  \"message\": \"consumer group topic offsets returned\",\r\n  \"offsets\": [\r\n    -1,\r\n    0\r\n  ]\r\n}\r\n```\r\n\r\nThe kafka documentation says: \"The committed offset should always be the offset of the next message that your application will read. Thus, when calling commitSync(offsets) you should add one to the offset of the last message processed.\"\r\n\r\nhttps://kafka.apache.org/090/javadoc/index.html?org/apache/kafka/clients/consumer/KafkaConsumer.html in \"Manual Offset Control\"\r\n\r\nI managed to fix this behavior by adding one to `consumer#mark_message_as_processed`\r\n\r\n```ruby\r\ndef mark_message_as_processed(message)\r\n  # included +1\r\n  @offset_manager.mark_as_processed(message.topic, message.partition, message.offset + 1)\r\nend\r\n```\r\n\r\nand assuming that kafka will have the next offset, I removed the +1 from `offset_manager#next_offset_for`\r\n\r\n```ruby\r\ndef next_offset_for(topic, partition)\r\n  offset = @processed_offsets.fetch(topic, {}).fetch(partition) {\r\n    committed_offset_for(topic, partition)\r\n  }\r\n\r\n  if offset < 0\r\n    resolve_offset(topic, partition)\r\n  else\r\n    # removed +1\r\n    offset\r\n  end\r\nend\r\n```\r\n\r\nafter those changes burrow correctly notifies the lag and `ruby-kafka` consumers behave the same as the other consumers (java, erlang, etc).\r\n\r\nThe consumers are consuming all messages but since the offsets are shifted by one the monitoring tools are reporting lag or stopped consumers. I can create a PR with the changes but I decided to get some input first. I checked other libraries and they behave like that.\r\n\r\nThanks\r\n"},"comment":{"url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/comments/244688661","html_url":"https://github.com/zendesk/ruby-kafka/issues/280#issuecomment-244688661","issue_url":"https://api.github.com/repos/zendesk/ruby-kafka/issues/280","id":244688661,"user":{"login":"tulios","id":33231,"avatar_url":"https://avatars.githubusercontent.com/u/33231?v=3","gravatar_id":"","url":"https://api.github.com/users/tulios","html_url":"https://github.com/tulios","followers_url":"https://api.github.com/users/tulios/followers","following_url":"https://api.github.com/users/tulios/following{/other_user}","gists_url":"https://api.github.com/users/tulios/gists{/gist_id}","starred_url":"https://api.github.com/users/tulios/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tulios/subscriptions","organizations_url":"https://api.github.com/users/tulios/orgs","repos_url":"https://api.github.com/users/tulios/repos","events_url":"https://api.github.com/users/tulios/events{/privacy}","received_events_url":"https://api.github.com/users/tulios/received_events","type":"User","site_admin":false},"created_at":"2016-09-05T08:30:13Z","updated_at":"2016-09-05T08:30:13Z","body":"I'll move the +1 to the offset commit and prepare a PR today. Thanks"}}
{"actor":{"display_login":"tulios"}}