mjsax commented on code in PR #23082:
URL: https://github.com/apache/kafka/pull/23082#discussion_r3723065416
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1270,26 +1270,30 @@ public Map<StreamsRebalanceData.TaskId, Long>
taskOffsetSumSnapshot() {
}
/**
- * Recomputes the offset-sum snapshot reported to the streams-group
coordinator from the offset sums maintained in
- * the {@link StateDirectory}, which already cover all stateful tasks with
state on disk (standby, warmup,
- * restoring-active and dormant) using a conservative (per-partition
lower-bound) sum. Running-active tasks are
- * excluded: their assignment is not offset-driven and they are caught up
by definition.
+ * Recomputes the offset-sum snapshot reported to the streams-group
coordinator. The {@link StateDirectory} sums
+ * cover every task with state on disk, including tasks owned by a sibling
stream thread and dormant ones.
+ * We include these, because we could take over such a task re-using the
local state on disk.
+ * Running-active tasks are excluded: their assignment is not
offset-driven and they are caught up by definition.
+ * For all other assigned tasks this thread owns, we overwrite the
(potentially) state offset-sum from the
+ * state directory with the latest changelog offset information. This step
also add offset-sums for in-memory
+ * state stores.
*/
public void maybeUpdateTaskOffsetSumSnapshot() {
- final Set<TaskId> runningActiveTasks = new HashSet<>();
+ final Map<TaskId, Long> offsetSums = new
HashMap<>(stateDirectory.taskOffsetSums());
for (final Task task : allTasks().values()) {
if (task.isActive() && task.state() == State.RUNNING) {
- runningActiveTasks.add(task.id());
+ offsetSums.remove(task.id());
+ } 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));
+ }
Review 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.
--
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]
