Copilot commented on code in PR #22645:
URL: https://github.com/apache/kafka/pull/22645#discussion_r3459955076
##########
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();
+ }
Review Comment:
`taskOffsetIntervalMs()` is documented to return `-1` when not yet set. With
the current math, an interval of `-1` makes `taskOffsetIntervalPassed()` always
return true, which can cause the offset/end-offset reporting block to run on
every heartbeat (and update `lastTaskOffsetIntervalTs`) even before the broker
has provided a valid interval.
##########
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
Review Comment:
Typo in comment: "futher" → "further".
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -609,8 +609,6 @@ private static void
throwIfStreamsGroupHeartbeatRequestIsInvalid(
private static void
throwIfStreamsGroupHeartbeatRequestIsUsingUnsupportedFeatures(
StreamsGroupHeartbeatRequestData request
) throws InvalidRequestException {
- throwIfNotNull(request.taskOffsets(), "TaskOffsets are not supported
yet.");
- throwIfNotNull(request.taskEndOffsets(), "TaskEndOffsets are not
supported yet.");
throwIfNotNullOrEmpty(request.warmupTasks(), "WarmupTasks are not
supported yet.");
if (request.topology() != null) {
Review Comment:
This method no longer rejects `taskOffsets`/`taskEndOffsets` as "not
supported". There is an existing unit test
(`GroupCoordinatorServiceTest#testStreamsGroupHeartbeatFailsForUnsupportedFeatures`)
that asserts INVALID_REQUEST for these fields; with this change it will fail
and should be updated alongside the behavior change.
--
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]
