Kafkajs: Consumer offset "stuck" on log compacted messages

[jlek]

Hello,

I'm looking for some help with an issue where my consumer is "stuck" at a certain offset. (Or bunch of offsets: one offset for each partition in the topic.)

To break it down:

• I am trying to consume a log-compacted topic with 12 partitions (hundreds of millions of messages).
• I deployed multiple instances of our consumer service.
• The consumers started out great, but after a while, they all came to a halt. The offsets have not changed for a few days. For each partition, they are still lagging many millions of messages behind.
• I made a local copy of KafkaJS and added some extra debug logs to do some digging. Here's what I found:
  
  
  
  • For a given partition, the offset was n.
  
  
  • The messages at offset n have been log compacted away. The next available message is at, say, n + 13.
  
  
  • KafkaJS would fetch the messages at offset n.
  
  
  • It would receive a few messages, but all of them have offset < n. For example, it might receive one message at offset n - 1, or two messages at offset n - 2 and n - 1.
  
  
  • in batch.js these messages are ignored, so the result is treated as an empty batch. (It only uses messagesWithinOffset.)
  
  
  • The "empty" batch is skipped, and KafkaJS repeats the above steps.
  
  
  • My eachBatch callback is never triggered.
  
  

I am not sure what is going wrong here: do I need to change some configuration? (Currently using defaults across the board.) Has the consumer group entered a bad state, somehow? Or could this be a bug?

Advice would be greatly appreciated. Thanks in advance! :slightly_smiling_face:

[JaapRood]

Thanks for the work on that so far! Could you maybe run one of your consumers with logging enabled and post that back? Just be careful that it might contain broker ip's / hostnames, topic names and group names, which you might want to redact!

[jlek]

Sure thing @JaapRood , here you go (might look slightly different than you're used to, due to our logging solution):

{"level":20,"time":1574852754351,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T11:05:54.351Z","message":"Fetching from 1 partitions for 1 out of 1 topics","logger":"kafkajs","topics":"[\"REDACTED_TOPIC\"]","activeTopicPartitions":"[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[6]}]","pausedTopicPartitions":"[]","v":1}
{"level":20,"time":1574852754351,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T11:05:54.351Z","message":"Request Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"fh1-kafka04:9092","clientId":"kafkajs","correlationId":17066,"expectResponse":true,"size":107,"v":1}
{"level":20,"time":1574852754354,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T11:05:54.354Z","message":"Response Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"fh1-kafka04:9092","clientId":"kafkajs","correlationId":17066,"size":562,"data":"[filtered]","v":1}

It repeats these three logs ad infinitum.

[JaapRood]

Thanks!

That seems pretty typical for a consumer that's actively consuming. Perhaps log messages from before it starts repeating those? If it's practical to just dump the whole thing, we might be able to figure out if something in fetching metadata / fetching offsets goes wrong.

[jlek]

Hey @JaapRood, here are the logs from start up:

{"level":20,"time":1574863725693,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Broker","label":"DEBUG","timestamp":"2019-11-27T14:08:45.693Z","message":"Verified support for SaslAuthenticate","logger":"kafkajs","broker":"REDACTED_BROKER","supportAuthenticationProtocol":true,"v":1}
{"level":20,"time":1574863725696,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:45.696Z","message":"Request GroupCoordinator(key: 10, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":0,"expectResponse":true,"size":49,"v":1}
{"level":20,"time":1574863725701,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:45.701Z","message":"Response GroupCoordinator(key: 10, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":0,"size":33,"data":"{\"throttleTime\":0,\"errorCode\":0,\"errorMessage\":null,\"coordinator\":{\"nodeId\":1,\"host\":\"fh1-kafka02\",\"port\":9092}}","v":1}
{"level":20,"time":1574863725701,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Cluster","label":"DEBUG","timestamp":"2019-11-27T14:08:45.701Z","message":"Found group coordinator","logger":"kafkajs","nodeId":1,"v":1}
{"level":20,"time":1574863725707,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:45.707Z","message":"Request JoinGroup(key: 11, version: 2)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":2,"expectResponse":true,"size":131,"v":1}
{"level":20,"time":1574863726841,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.840Z","message":"Response JoinGroup(key: 11, version: 2)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":2,"size":130,"data":"{\"throttleTime\":0,\"errorCode\":0,\"generationId\":7379,\"groupProtocol\":\"RoundRobinAssigner\",\"leaderId\":\"kafkajs-5c8b6dc3-ac78-4544-9189-25f704976041\",\"memberId\":\"kafkajs-f8ec9915-ce7e-4fbd-b3bb-182622057e3b\",\"members\":[]}","v":1}
{"level":20,"time":1574863726846,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.846Z","message":"Request SyncGroup(key: 14, version: 1)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":3,"expectResponse":true,"size":102,"v":1}
{"level":20,"time":1574863726849,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.849Z","message":"Response SyncGroup(key: 14, version: 1)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":3,"size":57,"data":"{\"throttleTime\":0,\"errorCode\":0,\"memberAssignment\":{\"type\":\"Buffer\",\"data\":[0,1,0,0,0,1,0,23,82,101,108,97,116,105,111,110,46,100,98,111,46,80,101,114,115,111,110,72,105,110,116,0,0,0,1,0,0,0,11,0,0,0,0]}}","v":1}
{"level":20,"time":1574863726850,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T14:08:46.850Z","message":"Received assignment","logger":"kafkajs","groupId":"merckx-production-kafkajs","generationId":7379,"memberId":"kafkajs-f8ec9915-ce7e-4fbd-b3bb-182622057e3b","memberAssignment":"{\"REDACTED_TOPIC\":[11]}","v":1}
{"level":30,"time":1574863726852,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Runner","label":"INFO","timestamp":"2019-11-27T14:08:46.852Z","message":"Consumer has joined the group","logger":"kafkajs","groupId":"merckx-production-kafkajs","memberId":"kafkajs-f8ec9915-ce7e-4fbd-b3bb-182622057e3b","leaderId":"kafkajs-5c8b6dc3-ac78-4544-9189-25f704976041","isLeader":false,"memberAssignment":"{\"REDACTED_TOPIC\":[11]}","groupProtocol":"RoundRobinAssigner","duration":1288,"v":1}
{"level":20,"time":1574863726858,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.858Z","message":"Request OffsetFetch(key: 9, version: 3)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":4,"expectResponse":true,"size":85,"v":1}
{"level":20,"time":1574863726860,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.860Z","message":"Response OffsetFetch(key: 9, version: 3)","logger":"kafkajs","broker":"fh1-kafka02:9092","clientId":"kafkajs","correlationId":4,"size":59,"data":"{\"throttleTime\":0,\"responses\":[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[{\"partition\":11,\"offset\":\"65112478\",\"metadata\":\"\",\"errorCode\":0}]}],\"errorCode\":0}","v":1}
{"level":20,"time":1574863726861,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T14:08:46.861Z","message":"Fetching from 1 partitions for 1 out of 1 topics","logger":"kafkajs","topics":"[\"REDACTED_TOPIC\"]","activeTopicPartitions":"[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[11]}]","pausedTopicPartitions":"[]","v":1}
{"level":20,"time":1574863726879,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.879Z","message":"Request Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":1,"expectResponse":true,"size":107,"v":1}
{"level":20,"time":1574863726885,"pid":1,"hostname":"REDACTED_HOST_NAME","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T14:08:46.885Z","message":"Response Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":1,"size":1800,"data":"[filtered]","v":1}

Does this help?

[Nevon]

Thanks for the detailed bug report. What you're saying makes sense. If we indeed request offset N and receive back N-1, N-2, N-3 etc. we would have no way of moving on from N. It reminds me a bit of the issue with control records, where we have to commit the control record offset without the control record even having been handed off to userland code, so that we don't get stuck with a batch full of control records. The difference being that in that case we _do_ get N+1 etc so we know which offset to commit. If we don't get the record at N, we can't commit N.

It would be useful to see the Fetch response from your service, to see what actual records you are getting. Unfortunately it's hard coded to filter out any fetch response. I've opened #563 to allow you to set an environment variable to override this behavior for debugging purposes. If you run with KafkaJS as a local dependency, you could override the row I linked from #563 to log both the parsed fetch response as well as the payload buffer so that we could use that to reproduce the issue. However, that would mean making your fetch response public, which might not be an option if this is production data.

[JaapRood]

[...] fetch response as well as the payload buffer so that we could use that to reproduce the issue. However, that would mean making your fetch response public, which might not be an option if this is production data.

@jlek Kafka _should_ not care about the message payload (message.value, message.key), which are the ones that would hold the specific information. Both are just a sequence of bytes, shouldn't be any logic beyond that. So you could filter that out of Fetch response log, and it _shouldn't_ have an effect on our ability to reproduce the problem.

[jlek]

Hey @Nevon,

Here are the logs you asked for (you're right, they do dump a lot of data):

{"level":30,"time":1574874919365,"pid":1,"hostname":"REDACTED_TOPIC","msg":"initialising database connection pools","v":1}
{"level":30,"time":1574874919378,"pid":1,"hostname":"REDACTED_TOPIC","msg":"merckx-topic-consumer service running on port 3045","v":1}
{"level":30,"time":1574874919379,"pid":1,"hostname":"REDACTED_TOPIC","msg":"About to start Kafka-Consumer with groupId merckx-production","v":1}
{"level":20,"time":1574874919579,"pid":1,"hostname":"REDACTED_TOPIC","msg":"Setting logger level to: debug","v":1}
{"level":20,"time":1574874919579,"pid":1,"hostname":"REDACTED_TOPIC","msg":"Setting logger level to: debug","v":1}
{"level":20,"time":1574874919583,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Broker","label":"DEBUG","timestamp":"2019-11-27T17:15:19.583Z","message":"Verified support for SaslAuthenticate","logger":"kafkajs","broker":"REDACTED_BROKER","supportAuthenticationProtocol":true,"v":1}
{"level":20,"time":1574874919587,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:19.587Z","message":"Request GroupCoordinator(key: 10, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":0,"expectResponse":true,"size":49,"v":1}
{"level":20,"time":1574874919590,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:19.590Z","message":"Response GroupCoordinator(key: 10, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":0,"size":33,"data":"{\"throttleTime\":0,\"errorCode\":0,\"errorMessage\":null,\"coordinator\":{\"nodeId\":1,\"host\":\"fh1-kafka02\",\"port\":9092}}","v":1}
{"level":20,"time":1574874919590,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Cluster","label":"DEBUG","timestamp":"2019-11-27T17:15:19.590Z","message":"Found group coordinator","logger":"kafkajs","nodeId":1,"v":1}
{"level":20,"time":1574874919594,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:19.594Z","message":"Request JoinGroup(key: 11, version: 2)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":2,"expectResponse":true,"size":131,"v":1}
{"level":20,"time":1574874922591,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.591Z","message":"Response JoinGroup(key: 11, version: 2)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":2,"size":130,"data":"{\"throttleTime\":0,\"errorCode\":0,\"generationId\":7396,\"groupProtocol\":\"RoundRobinAssigner\",\"leaderId\":\"kafkajs-b8060803-4c28-4e32-b7e4-9d473b7c5193\",\"memberId\":\"kafkajs-51b12285-0f45-4e71-8597-8ecda551bbce\",\"members\":[]}","v":1}
{"level":20,"time":1574874922596,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.596Z","message":"Request SyncGroup(key: 14, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":3,"expectResponse":true,"size":102,"v":1}
{"level":20,"time":1574874922605,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.605Z","message":"Response SyncGroup(key: 14, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":3,"size":57,"data":"{\"throttleTime\":0,\"errorCode\":0,\"memberAssignment\":{\"type\":\"Buffer\",\"data\":REDACTED_SYNCGROUP_BUFFER}}","v":1}
{"level":20,"time":1574874922606,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T17:15:22.606Z","message":"Received assignment","logger":"kafkajs","groupId":"merckx-production-kafkajs","generationId":7396,"memberId":"kafkajs-51b12285-0f45-4e71-8597-8ecda551bbce","memberAssignment":"{\"REDACTED_TOPIC\":[3]}","v":1}
{"level":30,"time":1574874922606,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Runner","label":"INFO","timestamp":"2019-11-27T17:15:22.606Z","message":"Consumer has joined the group","logger":"kafkajs","groupId":"merckx-production-kafkajs","memberId":"kafkajs-51b12285-0f45-4e71-8597-8ecda551bbce","leaderId":"kafkajs-b8060803-4c28-4e32-b7e4-9d473b7c5193","isLeader":false,"memberAssignment":"{\"REDACTED_TOPIC\":[3]}","groupProtocol":"RoundRobinAssigner","duration":3030,"v":1}
{"level":20,"time":1574874922611,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.611Z","message":"Request OffsetFetch(key: 9, version: 3)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":4,"expectResponse":true,"size":85,"v":1}
{"level":20,"time":1574874922614,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.614Z","message":"Response OffsetFetch(key: 9, version: 3)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":4,"size":59,"data":"{\"throttleTime\":0,\"responses\":[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[{\"partition\":3,\"offset\":\"60444340\",\"metadata\":\"\",\"errorCode\":0}]}],\"errorCode\":0}","v":1}
{"level":20,"time":1574874922615,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T17:15:22.615Z","message":"Fetching from 1 partitions for 1 out of 1 topics","logger":"kafkajs","topics":"[\"REDACTED_TOPIC\"]","activeTopicPartitions":"[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[3]}]","pausedTopicPartitions":"[]","v":1}
{"level":20,"time":1574874922628,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.628Z","message":"Request Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":1,"expectResponse":true,"size":107,"v":1}
{"level":20,"time":1574874922634,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.634Z","message":"Response Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":1,"size":952,"data":"{\"throttleTime\":0,\"errorCode\":0,\"sessionId\":0,\"responses\":[{\"topicName\":\"REDACTED_TOPIC\",\"partitions\":[{\"partition\":3,\"errorCode\":0,\"highWatermark\":\"68627274\",\"lastStableOffset\":\"68627274\",\"lastStartOffset\":\"0\",\"abortedTransactions\":[],\"messages\":[{\"magicByte\":2,\"attributes\":0,\"timestamp\":\"1570632587028\",\"offset\":\"60444339\",\"key\":{\"type\":\"Buffer\",\"data\":REDACTED_KEY_BUFFER},\"value\":{\"type\":\"Buffer\",\"data\":REDACTED_VALUE_BUFFER},\"headers\":{},\"isControlRecord\":false,\"batchContext\":{\"firstOffset\":\"60444337\",\"firstTimestamp\":\"1570632587028\",\"partitionLeaderEpoch\":0,\"inTransaction\":false,\"isControlBatch\":false,\"lastOffsetDelta\":6,\"producerId\":\"-1\",\"producerEpoch\":-1,\"firstSequence\":-1,\"maxTimestamp\":\"1570632587028\",\"magicByte\":2}}]}]}]}","v":1}
{"level":20,"time":1574874922639,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.639Z","message":"Request Heartbeat(key: 12, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":5,"expectResponse":true,"size":98,"v":1}
{"level":20,"time":1574874922642,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.642Z","message":"Response Heartbeat(key: 12, version: 1)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":5,"size":10,"data":"{\"throttleTime\":0,\"errorCode\":0}","v":1}
{"level":20,"time":1574874922642,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"ConsumerGroup","label":"DEBUG","timestamp":"2019-11-27T17:15:22.642Z","message":"Fetching from 1 partitions for 1 out of 1 topics","logger":"kafkajs","topics":"[\"REDACTED_TOPIC\"]","activeTopicPartitions":"[{\"topic\":\"REDACTED_TOPIC\",\"partitions\":[3]}]","pausedTopicPartitions":"[]","v":1}
{"level":20,"time":1574874922643,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.643Z","message":"Request Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":2,"expectResponse":true,"size":107,"v":1}
{"level":20,"time":1574874922645,"pid":1,"hostname":"REDACTED_TOPIC","namespace":"Connection","label":"DEBUG","timestamp":"2019-11-27T17:15:22.645Z","message":"Response Fetch(key: 1, version: 7)","logger":"kafkajs","broker":"REDACTED_BROKER","clientId":"kafkajs","correlationId":2,"size":952,"data":"{\"throttleTime\":0,\"errorCode\":0,\"sessionId\":0,\"responses\":[{\"topicName\":\"REDACTED_TOPIC\",\"partitions\":[{\"partition\":3,\"errorCode\":0,\"highWatermark\":\"68627274\",\"lastStableOffset\":\"68627274\",\"lastStartOffset\":\"0\",\"abortedTransactions\":[],\"messages\":[{\"magicByte\":2,\"attributes\":0,\"timestamp\":\"1570632587028\",\"offset\":\"60444339\",\"key\":{\"type\":\"Buffer\",\"data\":REDACTED_KEY_BUFFER},\"value\":{\"type\":\"Buffer\",\"data\":REDACTED_VALUE_BUFFER},\"headers\":{},\"isControlRecord\":false,\"batchContext\":{\"firstOffset\":\"60444337\",\"firstTimestamp\":\"1570632587028\",\"partitionLeaderEpoch\":0,\"inTransaction\":false,\"isControlBatch\":false,\"lastOffsetDelta\":6,\"producerId\":\"-1\",\"producerEpoch\":-1,\"firstSequence\":-1,\"maxTimestamp\":\"1570632587028\",\"magicByte\":2}}]}]}]}","v":1}

[jlek]

So, any suggestions on how to fix this? @JaapRood , @Nevon ? I'd be happy to take a stab at making a pull request, but I'm not sure what change needs to be made.

If all messages have an offset less than N, can we make the assumption that the message at offset N was log compacted away, and increment the offset by 1?

Or should we make a change to the way messages are fetched from the broker? I'm not at all familiar with the API/protocol, but can we tell it to give us the first messages where the offset >= N?

[tulios]

@jlek are you running version 1.11.0? I think PR #511 fixed this problem, can you try 1.12.0-beta.0 or the latest beta release (1.12.0-beta.8)?

[jlek]

Hi @tulios

Thanks for looking into this. I am indeed running 1.11.0. I tried 1.12.0-beta.0 after reading your suggestion, but alas, it didn't fix the issue. Although I do think it's very closely related to our issue!

I will try to break it down as I understand it. Please let me know if I've misunderstood something. :slightly_smiling_face:

The issue you mentioned (PR #511)

Underlying Cause

KafkaJS could get stuck on an offset marker for a Control Record, because it would ignore the Control Record, handle an empty batch by doing nothing, fetch the same Control Record Batch again, and repeat.

Fix

Fixed an issue by incrementing the offset by 1 when a control batch was fetched.

In the Java Consumer

Useful comment in the Java consumer: https://github.com/apache/kafka/blob/9aa660786e46c1efbf5605a6a69136a1dac6edb9/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L1499-L1505

This issue

Underlying Cause

KafkaJS can get stuck on an offset marker for a compacted Record, because it will fetch an "empty" batch, handle the "empty" batch by doing nothing, and repeat. (The fetch did return some records, but they are all from an earlier offset than requested, so these records are ignored and the batch is treated like an empty batch.)

Hacky Fix

Can be fixed by incrementing the offset by 1 if the all the fetched messages are from before the requested offset. (It might also be worth checking that fetchedOffset is less than highWatermark, _i.e._ that we haven't reached the end of the partition yet.) I confirmed this by copying KafkaJS locally and hacking consumerGroup.js and batch.js a bit.

"Proper" Fix

To match the Java Consumer, we could:

• After a batch has been processed:
• Resolve the offset to the "nextOffset": https://github.com/apache/kafka/blob/9aa660786e46c1efbf5605a6a69136a1dac6edb9/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L1462
• Which is calculated as the "lastOffset" + 1: https://github.com/apache/kafka/blob/9aa660786e46c1efbf5605a6a69136a1dac6edb9/clients/src/main/java/org/apache/kafka/common/record/AbstractRecordBatch.java#L28
• Which is calculated as the "baseOffset" + "lastOffsetDelta": https://github.com/apache/kafka/blob/9aa660786e46c1efbf5605a6a69136a1dac6edb9/clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java#L182

In the Java Consumer

Useful comment in the Java consumer: https://github.com/apache/kafka/blob/9aa660786e46c1efbf5605a6a69136a1dac6edb9/clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java#L1456-L1460

[Nevon]

That is some really quality digging, @jlek. I really appreciate getting this detailed investigation. I think the "proper" fix you mentioned sounds totally doable, although testing it might be a bit of a challenge.

[Nevon]

We wrote a test to try to reproduce this locally, but weren't able to get it stuck. Could you share your topic configuration? We're not using topic compaction ourselves, so maybe we're misconfiguring something.

The topic configuration that we tried was:

await admin.createTopics({
    topics: [
      {
        topic,
        partitions: 1,
        replicationFactor: 1,
        configEntries: [
          { name: 'cleanup.policy', value: 'compact' },
          { name: 'delete.retention.ms', value: '100' },
          { name: 'min.cleanable.dirty.ratio', value: '0.01' },
          { name: 'max.compaction.lag.ms', value: '1' },
          { name: 'segment.bytes', value: '5000' },
          { name: 'max.message.bytes', value: '1000' },
        ],
      },
    ],
  })

Then we produced a bunch of messages with the same key, waited a bit and tried to consume them (which worked). It looks like it did compact the topic, sort of, although we ended up getting 2 messages in our consumer instead of just one:

{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.641Z","logger":"kafkajs","message":"[ConsumerGroup] Fetching from 1 partitions for 1 out of 1 topics","topics":["compacted-topic-test-4"],"activeTopicPartitions":[{"topic":"compacted-topic-test-4","partitions":[0]}],"pausedTopicPartitions":[]}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.650Z","logger":"kafkajs","message":"[Connection] Request Fetch(key: 1, version: 7)","broker":"localhost:9094","clientId":"example-consumer","correlationId":8,"expectResponse":true,"size":115}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.662Z","logger":"kafkajs","message":"[Connection] Response Fetch(key: 1, version: 7)","broker":"localhost:9094","clientId":"example-consumer","correlationId":8,"size":224,"data":"[filtered]"}
🚗 - compacted-topic-test-4[0 | 538] / 1575376830090 same-key-every-time#value-52-2019-12-03T12:40:30.090Z
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.666Z","logger":"kafkajs","message":"[Connection] Request OffsetCommit(key: 8, version: 3)","broker":"localhost:9094","clientId":"example-consumer","correlationId":9,"expectResponse":true,"size":195}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.677Z","logger":"kafkajs","message":"[Connection] Response OffsetCommit(key: 8, version: 3)","broker":"localhost:9094","clientId":"example-consumer","correlationId":9,"size":46,"data":{"throttleTime":0,"responses":[{"topic":"compacted-topic-test-4","partitions":[{"partition":0,"errorCode":0}]}]}}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.677Z","logger":"kafkajs","message":"[ConsumerGroup] Fetching from 1 partitions for 1 out of 1 topics","topics":["compacted-topic-test-4"],"activeTopicPartitions":[{"topic":"compacted-topic-test-4","partitions":[0]}],"pausedTopicPartitions":[]}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.678Z","logger":"kafkajs","message":"[Connection] Request Fetch(key: 1, version: 7)","broker":"localhost:9094","clientId":"example-consumer","correlationId":10,"expectResponse":true,"size":115}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.682Z","logger":"kafkajs","message":"[Connection] Response Fetch(key: 1, version: 7)","broker":"localhost:9094","clientId":"example-consumer","correlationId":10,"size":224,"data":"[filtered]"}
🚗 - compacted-topic-test-4[0 | 539] / 1575376830592 same-key-every-time#value-55-2019-12-03T12:40:30.591Z
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.684Z","logger":"kafkajs","message":"[Connection] Request OffsetCommit(key: 8, version: 3)","broker":"localhost:9094","clientId":"example-consumer","correlationId":11,"expectResponse":true,"size":195}
{"level":"DEBUG","timestamp":"2019-12-03T12:43:08.697Z","logger":"kafkajs","message":"[Connection] Response OffsetCommit(key: 8, version: 3)","broker":"localhost:9094","clientId":"example-consumer","correlationId":11,"size":46,"data":{"throttleTime":0,"responses":[{"topic":"compacted-topic-test-4","partitions":[{"partition":0,"errorCode":0}]}]}}

[jlek]

Hi @Nevon

The topic configuration is:

• 12 partitions
• replication factor 3
• cleanup.policy=compact
• delete.retention.ms=2678400000
• min.cleanable.dirty.ratio=0.01
• min.insync.replicas=2
• segment.bytes=1073741824
• segment.ms=86400000

I have an idea as to why it might be difficult to write a test: the bug might only occur when the compacted message is the last message in a RecordBatch. In our case, since we are working with hundreds of millions of real messages, we're bound to come across this at some point. With test data, it might be more difficult. (Take all of this with a grain of salt: I'm not 100% sure.)

[tulios]

@jlek our plan is to get a buffer and write unit tests, we just want to reproduce it locally. It would be quite difficult to have integration tests for this, but we should be able to have a fixture with a buffer in the right state.

[jlek]

Sounds good! Just let me know if there's any way I can help.

[JaapRood]

It would be quite difficult to have integration tests for this

Just to confirm, I had to try something similar a year back or so, and had to give up on it after a couple of days. Kafka's compaction timing strategy is based on all kinds of heuristics, as well as prone to GC and stuff, some virtually impossible to get that exact state reproducible.

[Nevon]

Could you share your "hacky" solution? We've been tracing through KafkaJS, comparing it to the Java implementation, and from our eyes it looks like we're doing the same thing - but clearly there is some difference.

Also, this is a long shot, but is there any chance that you could share an offending Fetch buffer with us for testing? We can't seem to replicate it locally, which makes debugging challenging.

It would be great if you'd join the KafkaJS Slack channel so we can iterate faster.

[jlek]

Sure thing, here's the hacky solution

In ConsumerGroup#Fetch I did this:

const batch = new Batch(topicName, fetchedOffset, partitionData);

if (batch.isEmptyDueToLogCompactedMessages() && parseInt(batch.offsetLag()) > 0) {
  this.logger.debug(`Incrementing offset to skip past log-compacted messages.`)
  this.resolveOffset({ // Increment the offset by 1
    topic: topicName,
    partition: partitionRequestData.partition,
    offset: fetchedOffset
  });
  this.commitOffsets();
}

return batch;

I also made some changes to Batch#lastOffset and Batch#offsetLag:

lastOffset() {
  return this.isEmptyIncludingFiltered()
    ? this.longFetchedOffset
    : this.unfilteredMessages[this.unfilteredMessages.length - 1].offset
}

/**
  * Returns the lag based on the last offset in the batch (also known as "high")
  */
offsetLag() {
  const lastOffsetOfPartition = Long.fromValue(this.highWatermark).add(-1)
  const lastConsumedOffset = Long.fromValue(this.lastOffset())
  return lastOffsetOfPartition.add(lastConsumedOffset.multiply(-1)).toString()
}

Not sure I can supply a Fetch buffer, because the problem only occurs in Production.

[jlek]

Oops, forgot this new method I added to Batch:

/**
   * With compressed messages, it's possible for the returned messages to have offsets smaller than the starting offset.
   * These messages will be filtered out (i.e. they are not even included in this.unfilteredMessages)
   * If these are the only messages, the batch will appear as an empty batch.
   * 
   * isEmpty() and isEmptyIncludingFiltered() will always return true if the batch is empty,
   * but this method will only return true if the batch is empty due to log compacted messages.
   * 
   * @returns boolean True if the batch is empty, because of log compacted messages in the partition.
   */
  isEmptyDueToLogCompactedMessages() {
    return this.partitionDataMessages.length > 0 && // There was at least one message
           this.isEmptyIncludingFiltered(); // All messages had an offset lower than the requested offset
  }