Get correct next offset for compacted topics

[Nevon]

FYI @jlek

The intention is that if you get a batch consisting only of messages from before the requested offset (n), we resolve offset n, which means that on next fetch you'll get n+1, allowing you to move forward.

We can't just resolve highwaterMark + 1, because then you'd skip past any messages after the current batch.

I haven't been able to reproduce #562 locally, and I don't have a Fetch buffer fixture to write a unit test from, so @jlek is going to try to validate this against his production workload.

[Nevon]

@jlek verified that the latest fix (stupid mistake on my part, assuming that longFetchedOffset was an instance variable, when it was local to the constructor) works for him.

[tulios]

There is a broker test failing; I think it's a new default configuration from the Kafka 2.3 update

[Nevon]

There is a broker test failing; I think it's a new default configuration from the Kafka 2.3 update

It was just that I hadn't update the azure pipeline conf to use Kafka 2.3. Fixed in 21b8264

[Nevon]

So this works, but if you have N compacted messages to get through, it'll take you N fetches because we only move forward one offset at a time (we resolve the requested offset after all).

It would be better if we calculated the offset to resolve by taking the firstOffset + lastOffsetDelta on the batch level. A naive solution would be something like:

class Batch {
  lastOffset() {
    if (this.isEmptyDueToLogCompactedMessages()) {
      // We know there's at least one message there, because we check it in
      // `isEmptyDueToLogCompactedMessages`
      const { firstOffset, lastOffsetDelta } = this.rawMessages[0].batchContext
      return firstOffset + lastOffsetDelta
    }

    if (this.isEmptyIncludingFiltered()) {
      return Long.fromValue(this.highWatermark)
        .add(-1)
        .toString()
    }

    return this.messagesWithinOffset[this.messagesWithinOffset.length - 1].offset
  }
}

But I'm not sure how to make that work with MessageSets rather than RecordBatches

[Nevon]

This is out in the pre-release version 1.12.0-beta.12.