dajac commented on code in PR #15534:
URL: https://github.com/apache/kafka/pull/15534#discussion_r1532739687
##########
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:
> The first HWM advancement to h1 will set lastHighWatermark to NO_OFFSET
and enqueueFirst() HWM update event.
Hum... My understanding is that the code will actually set lastHighWatermark
from NO_OFFSET to h1 and push the event in this case.
> Before the first event runs, let's say the HWM advances to h2. this will
see that lastHighWatermark is NO_OFFSET and will skip enqueueFirst().
It will update lastHighWatermark to h2 and, as the previous value is not
NO_OFFSET, it does not push the event this time.
> I wonder if we can:
> * keep track of highest HWM updated
> * only enqueueFirst if the offset to update is greater than highest HWM
recorded
Isn't it more or less what my change does? It does not enforce that the HWM
is greater than the previous one though but this should not happen.
--
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
