KAFKA-20893: Avoid reporting incorrect task-offsets to Kafka Streams assignor - #23082
Conversation
…assignor The `StateDirectory` offet cache should only contain data for persitent state store. This PR avoids the offset from in-memory state stores are added to the cache, and ensure that the read path fills in missing inforamtion for other state stores (in particular in-memory ones) on the fly.
|
|
nicktelford
left a comment
nicktelford
left a comment
There was a problem hiding this comment.
LGTM. My only comments are requests for clarification.
| if (task.isActive() && task.state() == State.RUNNING && !task.changelogPartitions().isEmpty()) { | ||
| taskOffsetSums.put(task.id(), Task.LATEST_OFFSET); |
There was a problem hiding this comment.
Since this branch was already unconditionally using LATEST_OFFSET for all running active tasks (with changelogs), this should narrow the scope of the bug to only tasks with in-memory stores that are currently assigned as standbys, or as actives that are not currently running (i.e. restoring state).
Still worth fixing, but I wanted to clarify the scope of the bug here.
There was a problem hiding this comment.
The main point of the fix is actually dormant tasks, ie, not owned any longer. For persistent stores, we still have the state in the state directory, so if the state-directory caches them that's correct. However, for in-memory stores, when a task is revoked we don't have any local state left, but the state-store cache still reported these offsets.
For assigned in-memory tasks (active or standby) it's ok if we report offsets, and not much changes for them with this PR -- before the change, the in-memory offsets are pushed into the cache and read from the cache for reporting. With the change, we won't add these offset into the cache any longer, and thus need to add them back on the read path for reporting.
The point is, that we don't report offset of dormet/previously assigned tasks with in-memory offsets any longer -- they don't make it into the cache any more, and on-read we only go over assigned tasks.
| } else if (task.state() != State.CREATED && task.state() != State.CLOSED) { | ||
| final Map<TopicPartition, Long> changelogOffsets = task.changelogOffsets(); | ||
| if (!changelogOffsets.isEmpty()) { | ||
| offsetSums.put(task.id(), StateDirectory.sumOfChangelogOffsets(task.id(), changelogOffsets)); | ||
| } |
There was a problem hiding this comment.
If I'm reading this right, I think this branch is actually fixing a slightly different bug: a failure to include offsets for standby tasks, or active tasks that are not yet running (i.e. restoring). Is that right, or did I miss how this relates to the in-memory stores issue?
There was a problem hiding this comment.
No, both of these were already included in the state-directory cache previously. -- So technically, we could limit this branch to only all offsets from in-memory stores (which we don't get from the state-directory cache any longer), but we just include all owned stores for the benefit of getting slightly more up-to-date offsets (compare to what we got from the state-directory cache). The main benefit is really simpler code/logic: we just go over tasks, and don't need to check if state stores are persistent or in-memory.
Since KIP-1035 reverted the guard was originally added by KAFKA-10249 #8996 I don't think that will be too difficult |
|
Kicked off a system test run - will post results once it completes |
|
Back-port to |
…assignor (#23082) (#23089) The `StateDirectory` offset cache should only contain data for persistent state store. This PR avoids the offset from in-memory state stores are added to the cache, and ensure that the read path fills in missing information for other state stores (in particular in-memory ones) on the fly. Reviewers: Nick Telford <nick.telford@gmail.com>, Bill Bejeck <bbejeck@apache.org>
3 participants
The
StateDirectoryoffset cache should only contain data forpersistent state store. This PR avoids the offset from in-memory state
stores are added to the cache, and ensure that the read path fills in
missing information for other state stores (in particular in-memory
ones) on the fly.
Reviewers: Nick Telford nick.telford@gmail.com, Bill Bejeck
bbejeck@apache.org