Only fetch for partitions with initialized offsets

[Nevon]

Second attempt at solving the issue with concurrent recovery of offset out of range, this time without locks.

Fixes #555

Closes #578

[tulios]

@JaapRood this is a new approach to the problem describe in the Lock PR.

[Nevon]

To reproduce the issue, I ran the branch replace-fetch-promise-all-with-generator. First I created a topic with the following configuration:

{
  topic,
  numPartitions: 3,
  configEntries: [
    { name: 'delete.retention.ms', value: '100' },
    { name: 'file.delete.delay.ms', value: '200' },
    { name: 'min.cleanable.dirty.ratio', value: '0.01' },
    { name: 'retention.ms', value: '1' },
    { name: 'segment.bytes', value: '50000' },
    { name: 'max.compaction.lag.ms', value: '200' },
  ],
}

I produced a bunch of messages to that topic, and then I started a consumer and consumed a bit. Then I killed my consumer and kept producing until the broker did a cleanup of old segments (this takes minutes btw...) Now when I started the consumer again, it would break because the old committed offset was now invalid, as described in #578.

With this fix, what happened instead was that we detected that some partitions didn't have a valid offset, and didn't fetch for them. Shortly after, each partition had recovered and fetches continued as usual.

[JaapRood]

Is there a reason why we're using committedOffsets here, instead of just resolved? An invalid offset could come from attempting to resume from a committed offset, but also from a consumer.seek. Your great comment touches on how both are cleared, so I'm not sure whether there would be any actual difference in behaviour, but as future changes are made this subtle difference might be harder to spot while becoming more consequential.

[Nevon]

Initially I was actually using resolvedOffsets, but that didn't work because OffsetManager.resolveOffsets actually just sets the initialized consumer offsets in committedOffsets, not in resolvedOffsets (did someone mention that our naming is confusing...? 😅). When the consumer first boots, it doesn't actually have any resolved offsets, so the only source of offsets is the initialized offsets in committedOffsets.

Regarding the seek behavior, I would expect it to work the same way, no? Seek would commit (potentially invalid) offsets and then clear both committedOffsets and resolvedOffsets using OffsetManager.clearOffsets. In the fetch loop we'd get the consumer offsets from the brokers and from there on it's the same.

Maybe I'm missing something?

[JaapRood]

Regarding the seek behavior, I would expect it to work the same way, no? Seek would commit (potentially invalid) offsets

Seeking shouldn't commit, only move the "playhead", see #395, so to rely on that behaviour is probably not the thing we want.

Initially I was actually using resolvedOffsets, but that didn't work because OffsetManager.resolveOffsets actually just sets the initialized consumer offsets in committedOffsets, not in resolvedOffsets (did someone mention that our naming is confusing...? 😅).

I guess having to use committedOffsets is a symptom of there being an issue in there then. Conceptually, it's the resolvedOffsets (which I understand is the "next to consume offset" or "playhead" for reading the log) that should always exist and the committed offset which is optional, as using Kafka for committing offsets is / should be totally optional (see #395).

Since that seems like a different issue, maybe it's an idea we create a separate issue for it and tag that in a comment. Being able to spot outside of the context of these changes that we conceptually want the resolved offsets rather than committed there might be a lot to ask from our future selves (or others) 😅.

[Nevon]

Since that seems like a different issue, maybe it's an idea we create a separate issue for it and tag that in a comment.

That sounds like a good idea. I would prefer to do that kind of holistic refactoring in a PR that doesn't actually change any behavior, rather than squeezing it into a bugfix. Could you create that issue?

[JaapRood]

Created the issue, trying to preserve the context of this conversation properly: #585. To help the audit suggested in there, I'd suggest linking that issue in a comment above where committedOffsets() is called.

[JaapRood]

Pragmatic fix, no locks, 🙌.

[tulios]

The pre-release version 1.12.0-beta.11 was published with this fix.