mjsax commented on code in PR #23987:
URL: https://github.com/apache/kafka/pull/23987#discussion_r4415127294
##########
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:
Good catch!
--
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]
