nicktelford commented on code in PR #23082:
URL: https://github.com/apache/kafka/pull/23082#discussion_r3719804858
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1323,11 +1327,20 @@ public Map<TaskId, Long> taskOffsetSums() {
final Map<TaskId, Long> taskOffsetSums =
stateDirectory.taskOffsetSums(lockedTaskDirectoriesOfNonOwnedTasksAndClosedAndCreatedTasks);
- // overlay latest offsets from assigned tasks
+ // Overlay latest offsets from assigned tasks.
+ // `stateDirectory.taskOffsetSums` above only cover what is
recoverable from disk (potentially stale),
+ // including offsets from persistent stores which tasks are owned by
sibling thread, as well as dormant tasks;
+ // We update this (potentially state) information with latest
changelog offset inforamtion for all other
+ // tasks assigned to this thread; this step also adds offset-sum
information for in-memory state stores
for (final Task task : tasks.values()) {
// exclude stateless and non-logged tasks
if (task.isActive() && task.state() == State.RUNNING &&
!task.changelogPartitions().isEmpty()) {
taskOffsetSums.put(task.id(), Task.LATEST_OFFSET);
Review 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.
##########
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:
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?
--
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]
