mjsax commented on code in PR #22715:
URL: https://github.com/apache/kafka/pull/22715#discussion_r3503815504
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -179,23 +179,24 @@ public StreamsGroupHeartbeatRequestData
buildRequestData() {
final Map<StreamsRebalanceData.TaskId, Long> taskOffsetSum =
streamsRebalanceData.taskOffsetSum();
final Map<StreamsRebalanceData.TaskId, Long> taskEndOffsetSum
= streamsRebalanceData.taskEndOffsetSum();
+ final long now = time.milliseconds();
if (assignmentChanged
- || taskOffsetIntervalPassed()
+ || taskOffsetIntervalPassed(now)
||
hasAtLeastOneHotWarmupTask(reconciledAssignment.warmupTasks(), 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.
+ // otherwise. reset() clears the snapshot on any
error/disconnect, forcing a full resend afterward.
if (!taskOffsetSum.equals(lastSentFields.taskOffsets)) {
data.setTaskOffsets(convertToList(taskOffsetSum));
lastSentFields.taskOffsets = taskOffsetSum;
+ lastTaskOffsetIntervalTs = now;
}
if
(!taskEndOffsetSum.equals(lastSentFields.taskEndOffsets)) {
data.setTaskEndOffsets(convertToList(taskEndOffsetSum));
lastSentFields.taskEndOffsets = taskEndOffsetSum;
+ lastTaskOffsetIntervalTs = now;
}
-
- lastTaskOffsetIntervalTs = time.milliseconds();
Review Comment:
This is the actual fix. We should not set this unconditionally.
--
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]
