Copilot commented on code in PR #22608:
URL: https://github.com/apache/kafka/pull/22608#discussion_r3459430723
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java:
##########
@@ -432,6 +432,23 @@ public Set<TopicPartition> completedChangelogs() {
.collect(Collectors.toSet());
}
+ // report a conservative value (higher as real value) as fallback
+ // only "source topic optimized" changelogs have a logical offset from
`ChangelogMetadata` (the committed offset on the source topic)
+ // -> if not known (not set yet, or no "source topic optimization"), fall
back to physical endOffset
Review Comment:
The new comment above `logicalChangelogEndOffsets()` has a couple of
grammatical issues that make it harder to read (e.g., “higher as real value”).
Consider rephrasing for clarity.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -442,6 +447,43 @@ private void
addToExceptionsAndFailedTasksThenClearUpdatingAndPausedTasks(final
}
}
+ // Current offset sums are sourced from the StateDirectory (see
TaskManager#maybeUpdateTaskOffsetSumSnapshot);
+ // end-offsets are not tracked there, so the end-offset-sum is
computed here from the changelog reader's
+ // (logical) end-offsets for the tasks the state updater is currently
restoring/updating.
+ private void updateTaskEndOffsetSumSnapshot() {
+ final Map<TopicPartition, Long> changelogEndOffsets;
+ if (!updatingTasks.isEmpty()) {
+ changelogEndOffsets =
changelogReader.logicalChangelogEndOffsets();
+ } else {
+ changelogEndOffsets = Collections.emptyMap();
+ }
+
+ final Map<StreamsRebalanceData.TaskId, Long> endOffsetSnapshot =
new HashMap<>(updatingTasks.size());
+
+ for (final Task task : updatingTasks.values()) {
+ long endSum = 0L; // ok to init with zero, as we have at least
one changelog topic partition
+ for (final TopicPartition partition :
task.changelogPartitions()) {
+ final Long endOffset = changelogEndOffsets.get(partition);
+ if (endOffset == null) {
+ endSum = Long.MAX_VALUE;
+ break;
+ }
+ if (endSum > Long.MAX_VALUE - endOffset) {
+ endSum = Long.MAX_VALUE;
+ break;
+ }
+ endSum += endOffset;
+ }
+
+ endOffsetSnapshot.put(
+ new
StreamsRebalanceData.TaskId(String.valueOf(task.id().subtopology()),
task.id().partition()),
+ endSum
+ );
+ }
+
+ taskEndOffsetSumSnapshot.set(endOffsetSnapshot);
Review Comment:
`taskEndOffsetSumSnapshot` is published as a mutable `HashMap`. Since this
snapshot is explicitly meant to be safe to access from any thread (and is
exposed via a public accessor), returning an unmodifiable map helps prevent
accidental mutation by callers and matches the existing `taskOffsetSumSnapshot`
behavior in `TaskManager#maybeUpdateTaskOffsetSumSnapshot()`.
--
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]
