mjsax commented on code in PR #21803:
URL: https://github.com/apache/kafka/pull/21803#discussion_r3463031525
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -147,19 +151,87 @@ public StreamsGroupHeartbeatRequestData
buildRequestData() {
data.setActiveTasks(fromStreamsToHeartbeatRequest(Set.of()));
data.setStandbyTasks(fromStreamsToHeartbeatRequest(Set.of()));
data.setWarmupTasks(fromStreamsToHeartbeatRequest(Set.of()));
+
data.setTaskOffsets(convertToList(streamsRebalanceData.taskOffsetSum()));
+
data.setTaskEndOffsets(convertToList(streamsRebalanceData.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)) {
+
+ // TODO: send only if changed this last time
+ data.setTaskOffsets(convertToList(taskOffsetSum));
+ data.setTaskEndOffsets(convertToList(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:
I guess this would be logic we would need on the broker? This PR is client
side, so it might actually be correct to return `true` for this case, and
report task-offset-sum and task-end-offset-sum to the broker right away? If we
add a check and return `false`, we would only not report the offsets to the
broker until next `task.offset.interval.ms` passed; don't think we gain
anything, contrary -- if we detect a corrupted warmup task, we should surface
it to the broker quickly, so the broker can compute a different assignment?
--
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]
