This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 416ef9bece [#10169] fix(core): Make drop-event log throttling atomic
(#12711)
416ef9bece is described below
commit 416ef9bece606eecf843b9c112803c04f1d88893
Author: MrSibe <[email protected]>
AuthorDate: Mon Aug 31 16:26:50 2026 +0800
[#10169] fix(core): Make drop-event log throttling atomic (#12711)
### What changes were proposed in this pull request?
- Guard the drop-event throttle decision and state updates with a lock.
- Update the drop counter and timestamp atomically within the critical
section.
- Capture the current time once and perform log I/O outside the critical
section.
- Add a deterministic concurrency test that verifies only one warning is
emitted within the throttle window.
### Why are the changes needed?
`AsyncQueueListener.logDropEventsIfNecessary()` previously updated the
drop counter and throttle timestamp independently.
Under concurrent drops, another thread could observe the updated counter
before the timestamp was updated and emit a duplicate warning within the
same 60-second throttle window.
The lock makes the time check, counter update, and timestamp update a
single atomic operation.
Fix: #10169
### Does this PR introduce _any_ user-facing change?
No. This change does not modify user-facing APIs or configuration
properties.
It only fixes duplicate drop-event warnings under concurrent queue
overflow.
### How was this patch tested?
Added `TestAsyncQueueListener.testDropEventLogThrottlingIsAtomic`.
The test blocks the first warning while a second thread enters the drop
path. It verifies that the second concurrent drop does not emit another
warning within the same throttle window.
The following Gradle tasks passed locally with Gradle 8.2:
- `:core:spotlessApply`
- `:core:test --tests
org.apache.gravitino.listener.TestAsyncQueueListener -PskipITs`
---
.../gravitino/listener/AsyncQueueListener.java | 37 +++---
.../gravitino/listener/TestAsyncQueueListener.java | 127 +++++++++++++++++++++
2 files changed, 148 insertions(+), 16 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
b/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
index 064a9a63c9..fd5387b40b 100644
--- a/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
+++ b/core/src/main/java/org/apache/gravitino/listener/AsyncQueueListener.java
@@ -52,7 +52,8 @@ public class AsyncQueueListener implements
EventListenerPlugin {
private final int dispatcherJoinSeconds;
private final AtomicBoolean stopped = new AtomicBoolean(false);
private final AtomicLong dropEventCounters = new AtomicLong(0);
- private final AtomicLong lastDropEventCounters = new AtomicLong(0);
+ private final Object dropEventLogLock = new Object();
+ private long lastDropEventCounters;
private Instant lastRecordDropEventTime = Instant.EPOCH;
private final String asyncQueueListenerName;
private final int highWatermarkThreshold;
@@ -144,23 +145,27 @@ public class AsyncQueueListener implements
EventListenerPlugin {
}
private void logDropEventsIfNecessary() {
- long currentDropEvents = dropEventCounters.incrementAndGet();
- long lastDropEvents = lastDropEventCounters.get();
- // dropEvents may less than zero in such conditions:
- // 1. Thread A increment dropEventCounters
- // 2. Thread B increment dropEventCounters and update lastDropEventCounters
- // 3. Thread A get lastDropEventCounters
- long dropEvents = currentDropEvents - lastDropEvents;
- if (dropEvents > 0 &&
Instant.now().isAfter(lastRecordDropEventTime.plusSeconds(60))) {
- if (lastDropEventCounters.compareAndSet(lastDropEvents,
currentDropEvents)) {
- LOG.warn(
- "{} drop {} events since {}",
- asyncQueueListenerName,
- dropEvents,
- lastRecordDropEventTime);
- lastRecordDropEventTime = Instant.now();
+ long dropEventsToLog = 0;
+ Instant previousRecordTime = null;
+ synchronized (dropEventLogLock) {
+ long currentDropEvents = dropEventCounters.incrementAndGet();
+ long lastDropEvents = lastDropEventCounters;
+ long dropEvents = currentDropEvents - lastDropEvents;
+ Instant now = Instant.now();
+ if (dropEvents > 0 &&
now.isAfter(lastRecordDropEventTime.plusSeconds(60))) {
+ lastDropEventCounters = currentDropEvents;
+ dropEventsToLog = dropEvents;
+ previousRecordTime = lastRecordDropEventTime;
+ lastRecordDropEventTime = now;
}
}
+ if (dropEventsToLog > 0) {
+ LOG.warn(
+ "{} drop {} events since {}",
+ asyncQueueListenerName,
+ dropEventsToLog,
+ previousRecordTime);
+ }
}
private void enqueueEvent(BaseEvent baseEvent) {
diff --git
a/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
b/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
new file mode 100644
index 0000000000..a5d578a7c0
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/listener/TestAsyncQueueListener.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.gravitino.listener;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.gravitino.listener.api.event.Event;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.AbstractConfiguration;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.core.layout.PatternLayout;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class TestAsyncQueueListener {
+ private BlockingAppender blockingAppender;
+ private LoggerContext loggerContext;
+
+ @BeforeEach
+ void setUp() {
+ loggerContext =
+ (LoggerContext)
LogManager.getContext(AsyncQueueListener.class.getClassLoader(), false);
+ Configuration configuration = loggerContext.getConfiguration();
+
+ blockingAppender = new BlockingAppender("asyncQueueListenerCapture");
+ blockingAppender.start();
+ configuration.addAppender(blockingAppender);
+
+ LoggerConfig loggerConfig =
+ new LoggerConfig(AsyncQueueListener.class.getName(), Level.WARN,
false);
+ loggerConfig.addAppender(blockingAppender, Level.WARN, null);
+ configuration.addLogger(AsyncQueueListener.class.getName(), loggerConfig);
+ loggerContext.updateLoggers();
+ }
+
+ @AfterEach
+ void tearDown() {
+ blockingAppender.releaseFirstLog.countDown();
+ AbstractConfiguration configuration = (AbstractConfiguration)
loggerContext.getConfiguration();
+ configuration.removeLogger(AsyncQueueListener.class.getName());
+ blockingAppender.stop();
+ configuration.removeAppender(blockingAppender.getName());
+ loggerContext.updateLoggers();
+ }
+
+ @Test
+ void testDropEventLogThrottlingIsAtomic() throws Exception {
+ AsyncQueueListener listener = new AsyncQueueListener(List.of(), "test", 1,
1);
+ Event event = mock(Event.class);
+ listener.onPostEvent(event);
+
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ Future<?> firstDrop = executor.submit(() -> listener.onPostEvent(event));
+ assertTrue(blockingAppender.firstLogEntered.await(30, TimeUnit.SECONDS));
+
+ Future<?> secondDrop = executor.submit(() ->
listener.onPostEvent(event));
+ secondDrop.get(30, TimeUnit.SECONDS);
+
+ blockingAppender.releaseFirstLog.countDown();
+ firstDrop.get(30, TimeUnit.SECONDS);
+
+ assertEquals(1, blockingAppender.logCount.get());
+ } finally {
+ blockingAppender.releaseFirstLog.countDown();
+ executor.shutdownNow();
+ assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
+ private static class BlockingAppender extends AbstractAppender {
+ private final AtomicInteger logCount = new AtomicInteger();
+ private final CountDownLatch firstLogEntered = new CountDownLatch(1);
+ private final CountDownLatch releaseFirstLog = new CountDownLatch(1);
+
+ BlockingAppender(String name) {
+ super(name, null, PatternLayout.createDefaultLayout(), true, null);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ if (logCount.incrementAndGet() != 1) {
+ return;
+ }
+
+ firstLogEntered.countDown();
+ try {
+ assertTrue(releaseFirstLog.await(30, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError("Interrupted while waiting to release the
first log", e);
+ }
+ }
+ }
+}
