This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 09a8658d39 [Cherry-pick to branch-1.3] [#10169] fix(core): Make
drop-event log throttling atomic (#12711) (#12742)
09a8658d39 is described below
commit 09a8658d39650d55aff6c408a297708eae45abef
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Aug 31 20:27:57 2026 +0800
[Cherry-pick to branch-1.3] [#10169] fix(core): Make drop-event log
throttling atomic (#12711) (#12742)
**Cherry-pick Information:**
- Original commit: 416ef9bece606eecf843b9c112803c04f1d88893
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: MrSibe <[email protected]>
---
.../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);
+ }
+ }
+ }
+}
