dajac commented on code in PR #15534:
URL: https://github.com/apache/kafka/pull/15534#discussion_r1533486994
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -1191,30 +1212,37 @@ public void onHighWatermarkUpdated(
long offset
) {
log.debug("High watermark of {} incremented to {}.", tp, offset);
- scheduleInternalOperation("HighWatermarkUpdated(tp=" + tp + ",
offset=" + offset + ")", tp, () -> {
- CoordinatorContext context = coordinators.get(tp);
- if (context != null) {
- context.lock.lock();
- try {
- if (context.state == CoordinatorState.ACTIVE) {
- // The updated high watermark can be applied to
the coordinator only if the coordinator
- // exists and is in the active state.
- log.debug("Updating high watermark of {} to {}.",
tp, offset);
-
context.coordinator.updateLastCommittedOffset(offset);
- context.deferredEventQueue.completeUpTo(offset);
- coordinatorMetrics.onUpdateLastCommittedOffset(tp,
offset);
- } else {
- log.debug("Ignored high watermark updated for {}
to {} because the coordinator is not active.",
- tp, offset);
+ if (lastHighWatermark.getAndSet(offset) == NO_OFFSET) {
+ // An event to apply the new high watermark is pushed to the
front of the
+ // queue only if the previous value was -1L. If it was not, it
means that
+ // there is already an event waiting to process the last value.
Review Comment:
Good point. I added a new unit test to better cover this.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
