lucasbru commented on code in PR #23987:
URL: https://github.com/apache/kafka/pull/23987#discussion_r4413849665
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -225,22 +227,57 @@ private boolean hasAtLeastOneHotWarmupTask(
return false;
}
+ return warmupTasks.stream().anyMatch(taskId -> isHot(taskId,
taskOffsetSum, taskEndOffsetSum));
+ }
+
+ /**
+ * Whether an active task is restoring with a lag within the
acceptable recovery lag: a client reports the
+ * offsets of an active task only while restoring it, so a running
active task is never hot here.
+ */
+ private boolean hasAtLeastOneHotRestoringActiveTask(
+ final Set<StreamsRebalanceData.TaskId> activeTasks,
+ final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum,
+ final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
+ ) {
+ return activeTasks.stream().anyMatch(taskId -> isHot(taskId,
taskOffsetSum, taskEndOffsetSum));
+ }
+
+ private boolean isHot(
+ final StreamsRebalanceData.TaskId taskId,
+ final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum,
+ final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
+ ) {
final long acceptableRecoveryLag =
streamsRebalanceData.acceptableRecoveryLag();
- return warmupTasks.stream()
- .anyMatch(taskId -> {
- final Long offset = taskOffsetSum.get(taskId);
- final Long endOffset = taskEndOffsetSum.get(taskId);
+ final Long offset = taskOffsetSum.get(taskId);
+ final Long endOffset = taskEndOffsetSum.get(taskId);
- // offset and endOffset might not be known,
- // or be capped at MAX_VALUE due to overflow
- if (offset == null || offset == Long.MAX_VALUE
- || endOffset == null || endOffset == Long.MAX_VALUE) {
- return false;
- }
+ // offset and endOffset might not be known,
+ // or be capped at MAX_VALUE due to overflow
+ if (offset == null || offset == Long.MAX_VALUE
+ || endOffset == null || endOffset == Long.MAX_VALUE) {
+ return false;
+ }
- return endOffset - offset <= acceptableRecoveryLag;
- });
+ return endOffset - offset <= acceptableRecoveryLag;
+ }
+
+ /**
+ * Whether an active task is in the last sent task offsets but not in
the current ones: a client reports the
+ * offsets of an active task only while restoring it, so the task
finished restoring. Until the broker receives
+ * offsets without the task, it takes the task's owner to still be
restoring it.
+ */
+ private boolean hasAtLeastOneActiveTaskFinishedRestoring(
Review Comment:
This only catches the restoring -> running direction. What about an active
task that goes back into restoration without an assignment change? E.g. on a
`TaskCorruptedException`, `closeDirtyAndRevive` re-initializes the task under
the same assignment, so its offsets reappear in `taskOffsetSum`, but
`assignmentChanged` is false and the lag is usually above
`acceptable.recovery.lag`. Until the next interval the broker still thinks the
owner is processing the task and treats it as hot in the refiner. Could we make
the check symmetric, i.e. `lastSentTaskOffsets.containsKey(taskId) !=
taskOffsetSum.containsKey(taskId)`?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -183,6 +183,8 @@ public StreamsGroupHeartbeatRequestData buildRequestData() {
if (assignmentChanged
|| taskOffsetIntervalPassed(now)
||
hasAtLeastOneHotWarmupTask(reconciledAssignment.warmupTasks(), taskOffsetSum,
taskEndOffsetSum)
+ ||
hasAtLeastOneHotRestoringActiveTask(reconciledAssignment.activeTasks(),
taskOffsetSum, taskEndOffsetSum)
Review Comment:
Not blocking, more a question on the direction: the refiner only cares about
a per-task status (not reported / no offsets yet / restoring cold / restoring
hot). Would it be simpler to compute that status per task and send whenever it
differs from what we last sent, instead of adding a trigger per case? That
would also cover hot -> cold and the re-restore case, and avoid resending the
whole offset map on every heartbeat while a task stays hot.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -225,22 +227,57 @@ private boolean hasAtLeastOneHotWarmupTask(
return false;
}
+ return warmupTasks.stream().anyMatch(taskId -> isHot(taskId,
taskOffsetSum, taskEndOffsetSum));
Review Comment:
nit: now that `acceptableRecoveryLag` is read in `isHot`, the `isEmpty()`
early return above is redundant.
##########
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/StreamsGroupWarmupTaskLifecycleIntegrationTest.java:
##########
@@ -290,8 +321,6 @@ && targetActiveTasks(group, c).equals(tasksOfLeaver)
&& activeTasks(group, c).equals(tasksOfLeaver)
&& activeTasks(group, owner).equals(tasksOfOwner)
&& totalWarmupTasks(group) == 0);
- // Observed through the standby update listener rather than the thread
metadata, which a stream thread only
- // refreshes once all its active tasks are restored -- and c's gate
holds back the restore of its new tasks.
TestUtils.waitForCondition(
Review Comment:
Why was the comment here removed?
--
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]
