lucasbru commented on code in PR #21803:
URL: https://github.com/apache/kafka/pull/21803#discussion_r3466981850
##########
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();
Review Comment:
Yes, that second part was incorrect.
--
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]
