lucasbru commented on code in PR #22595:
URL: https://github.com/apache/kafka/pull/22595#discussion_r3427883122
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +448,36 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ private void updateTaskOffsetSumSnapshot() {
+ final Map<StreamsRebalanceData.TaskId, Long> snapshot = new
HashMap<>(updatingTasks.size());
+ for (final Task task : updatingTasks.values()) {
+ if (task.changelogPartitions().isEmpty()) {
+ continue;
+ }
+ long sum = 0L;
+ boolean unknownOffsetFound = false;
+ for (final Long offset : task.changelogOffsets().values()) {
+ if (offset == null || offset == OFFSET_UNKNOWN) {
+ unknownOffsetFound = true;
+ continue;
+ }
+ if (sum > Long.MAX_VALUE - offset) {
+ sum = Long.MAX_VALUE;
+ break;
+ }
+ sum += offset;
+ }
+ if (unknownOffsetFound && sum != Long.MAX_VALUE) {
+ sum = 0;
+ }
+ snapshot.put(
+ new
StreamsRebalanceData.TaskId(String.valueOf(task.id().subtopology()),
task.id().partition()),
+ sum
+ );
+ }
+ taskOffsetSumSnapshot.set(Map.copyOf(snapshot));
Review Comment:
`snapshot` is a fresh local that never escapes, so `Map.copyOf` is a second
full copy just to get immutability. Wrapping it once (or publishing the local
directly) would avoid the extra per-iteration allocation.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +448,36 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ private void updateTaskOffsetSumSnapshot() {
+ final Map<StreamsRebalanceData.TaskId, Long> snapshot = new
HashMap<>(updatingTasks.size());
+ for (final Task task : updatingTasks.values()) {
Review Comment:
This iterates only `updatingTasks` — when a topology is paused the tasks
move to `pausedTasks` and drop out of the snapshot entirely. Once this feeds
the heartbeat, a member still owning that state would report the paused task as
absent. Intended, or should `pausedTasks` be included too?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +448,36 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ private void updateTaskOffsetSumSnapshot() {
Review Comment:
The sum + overflow-pin-to-`Long.MAX_VALUE` logic duplicates
`StateDirectory.sumOfChangelogOffsets`, and the two already differ on how they
treat unknown offsets. Could we extract a shared helper so the classic-protocol
and STREAMS-protocol offset sums don't drift apart?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -203,6 +207,8 @@ private void runOnce() {
final long checkpointStartTimeMs = time.milliseconds();
maybeCheckpointTasks(checkpointStartTimeMs);
+ updateTaskOffsetSumSnapshot();
Review Comment:
This rebuilds the full map on every `runOnce()` iteration, but it's only
read at heartbeat cadence. During restore the loop spins quickly and each pass
allocates the snapshot map plus a fresh map per task from `changelogOffsets()`.
Could we recompute only when offsets actually advanced?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +448,36 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ private void updateTaskOffsetSumSnapshot() {
+ final Map<StreamsRebalanceData.TaskId, Long> snapshot = new
HashMap<>(updatingTasks.size());
+ for (final Task task : updatingTasks.values()) {
+ if (task.changelogPartitions().isEmpty()) {
+ continue;
+ }
+ long sum = 0L;
+ boolean unknownOffsetFound = false;
+ for (final Long offset : task.changelogOffsets().values()) {
+ if (offset == null || offset == OFFSET_UNKNOWN) {
+ unknownOffsetFound = true;
+ continue;
+ }
+ if (sum > Long.MAX_VALUE - offset) {
+ sum = Long.MAX_VALUE;
+ break;
+ }
+ sum += offset;
+ }
+ if (unknownOffsetFound && sum != Long.MAX_VALUE) {
Review Comment:
This zeroes the whole task's sum if any single partition is unknown, which
is different from `StateDirectory.sumOfChangelogOffsets` — that one just skips
the unknown partition and keeps the sum of the known ones. Also worth noting
that for real tasks `ProcessorStateManager.changelogOffsets()` maps an unknown
store offset to `0L` (never null or `OFFSET_UNKNOWN`), so this branch only ever
fires for the mocks in the new tests. Is the intent to match the
`StateDirectory` semantics, or is there a reason to report 0 for the whole task
here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
