lucasbru commented on code in PR #22645:
URL: https://github.com/apache/kafka/pull/22645#discussion_r3460019980
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -147,19 +155,101 @@ public StreamsGroupHeartbeatRequestData
buildRequestData() {
data.setActiveTasks(fromStreamsToHeartbeatRequest(Set.of()));
data.setStandbyTasks(fromStreamsToHeartbeatRequest(Set.of()));
data.setWarmupTasks(fromStreamsToHeartbeatRequest(Set.of()));
+
+ // call both methods only once, as they invoke an expensive
`supplier`
+ final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum =
streamsRebalanceData.taskOffsetSum();
+ final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
= streamsRebalanceData.taskEndOffsetSum();
+ data.setTaskOffsets(convertToList(taskOffsetSum));
+ data.setTaskEndOffsets(convertToList(taskEndOffsetSum));
+ // Record what we sent so the first non-joining heartbeat does
not redundantly resend unchanged offsets.
+ lastSentFields.taskOffsets = taskOffsetSum;
+ lastSentFields.taskEndOffsets = taskEndOffsetSum;
} else {
- StreamsRebalanceData.Assignment reconciledAssignment =
streamsRebalanceData.reconciledAssignment();
- if (!reconciledAssignment.equals(lastSentFields.assignment)) {
+ final StreamsRebalanceData.Assignment reconciledAssignment =
streamsRebalanceData.reconciledAssignment();
+ final boolean assignmentChanged =
!reconciledAssignment.equals(lastSentFields.assignment);
+
+ if (assignmentChanged) {
data.setActiveTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.activeTasks()));
data.setStandbyTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.standbyTasks()));
data.setWarmupTasks(fromStreamsToHeartbeatRequest(reconciledAssignment.warmupTasks()));
lastSentFields.assignment = reconciledAssignment;
}
+
+ // call both method only once, as they invoke an expensive
`supplier`
+ final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum =
streamsRebalanceData.taskOffsetSum();
+ final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
= streamsRebalanceData.taskEndOffsetSum();
+
+ if (assignmentChanged || taskOffsetIntervalPassed() ||
hasAtLeastOneHotWarmupTask(taskOffsetSum, taskEndOffsetSum)) {
+ // Task offsets and end-offsets are reported
independently. A null field means "unchanged since the
+ // last heartbeat", so we send each one only when its
value actually changed and leave it null
+ // otherwise. reset() clears the snapshot on any
error/disconnect, forcing a full resend afterwards.
+ if (!taskOffsetSum.equals(lastSentFields.taskOffsets)) {
+ data.setTaskOffsets(convertToList(taskOffsetSum));
+ lastSentFields.taskOffsets = taskOffsetSum;
+ }
+ if
(!taskEndOffsetSum.equals(lastSentFields.taskEndOffsets)) {
+
data.setTaskEndOffsets(convertToList(taskEndOffsetSum));
+ lastSentFields.taskEndOffsets = taskEndOffsetSum;
+ }
+
+ lastTaskOffsetIntervalTs = time.milliseconds();
+ }
}
data.setShutdownApplication(streamsRebalanceData.shutdownRequested());
return data;
}
+ private List<StreamsGroupHeartbeatRequestData.TaskOffset>
convertToList(Map<StreamsRebalanceData.TaskId, Long> offsetsMap) {
+ return offsetsMap.entrySet().stream().map(
+ entry -> new StreamsGroupHeartbeatRequestData.TaskOffset()
+ .setSubtopologyId(entry.getKey().subtopologyId())
+ .setPartition(entry.getKey().partitionId())
+ .setOffset(entry.getValue()))
+ .collect(Collectors.toList());
+ }
+
+ private boolean taskOffsetIntervalPassed() {
+ return lastTaskOffsetIntervalTs +
streamsRebalanceData.taskOffsetIntervalMs() <= time.milliseconds();
+ }
+
+ private boolean hasAtLeastOneHotWarmupTask(
+ final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum,
+ final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
+ ) {
+ final long acceptableRecoveryLag =
streamsRebalanceData.acceptableRecoveryLag();
+
+ // -1 means "unknown" (can happen when talking to older brokers)
+ // we must be conservative and assume that no warmup might be hot
already
+ //
+ // technically, we should never get warmup tasks assigned when
talking to older brokers,
+ // so this is just another safeguard, which should actually be
redundant:
+ // the code futher below should automatically return false if
there are no warmup tasks;
+ // checking `acceptableRecoveryLag` is cheaper though, so it's
also a small micro optimization
+ if (acceptableRecoveryLag < 0) {
+ return false;
+ }
+
+ final Set<StreamsRebalanceData.TaskId> warmupTasks =
streamsRebalanceData.reconciledAssignment().warmupTasks();
+ if (warmupTasks.isEmpty()) {
+ return false;
+ }
+
+ return warmupTasks.stream()
+ .anyMatch(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;
+ }
+
+ return endOffset - offset <= acceptableRecoveryLag;
Review Comment:
When endOffset < offset (log truncation, state corruption), endOffset -
offset is negative, and since acceptableRecoveryLag >= 0 at this point, the
condition is trivially true — the task is falsely promoted from warmup. The
existing assignor code in HighAvailabilityTaskAssignor.acceptable() guards
against this with taskLag >= 0 && taskLag <= acceptableRecoveryLag. Worth
adding the same guard here. Currently dormant since taskEndOffsetSum is wired
to Map::of, but will bite when the real supplier lands.
--
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]
