This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f7ff7831e88 branch-4.1: [fix](audit) serialize audit loader batch
assembly #65107 (#65132)
f7ff7831e88 is described below
commit f7ff7831e8806c8d889210dfe67985a1efba7645
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Jul 2 15:06:29 2026 +0800
branch-4.1: [fix](audit) serialize audit loader batch assembly #65107
(#65132)
Cherry-picked from #65107
Co-authored-by: shuke <[email protected]>
---
.../org/apache/doris/plugin/audit/AuditLoader.java | 3 +-
.../apache/doris/plugin/audit/AuditLoaderTest.java | 82 ++++++++++++++++++++++
2 files changed, 83 insertions(+), 2 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
index b4b8195b608..dab8fcc26fd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java
@@ -140,7 +140,7 @@ public class AuditLoader extends Plugin implements
AuditPlugin {
}
}
- private void assembleAudit(AuditEvent event) {
+ private synchronized void assembleAudit(AuditEvent event) {
fillLogBuffer(event, auditLogBuffer);
++auditLogNum;
}
@@ -286,4 +286,3 @@ public class AuditLoader extends Plugin implements
AuditPlugin {
}
}
}
-
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditLoaderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditLoaderTest.java
new file mode 100644
index 00000000000..63bab9c3f95
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditLoaderTest.java
@@ -0,0 +1,82 @@
+// 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.doris.plugin.audit;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.plugin.AuditEvent;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class AuditLoaderTest {
+
+ @Test
+ public void testAssembleAuditIsSerializedWithLoadLock() throws Exception {
+ AuditLoader auditLoader = new AuditLoader();
+ AuditEvent auditEvent = new AuditEvent.AuditEventBuilder()
+ .setQueryId("query-in-shared-monitor-test")
+ .setTimestamp(1L)
+ .setStmt("select 1")
+ .build();
+
+ CountDownLatch started = new CountDownLatch(1);
+ AtomicReference<Throwable> error = new AtomicReference<>();
+ Thread assembleThread = new Thread(() -> {
+ started.countDown();
+ try {
+ Deencapsulation.invoke(auditLoader, "assembleAudit",
auditEvent);
+ } catch (Throwable t) {
+ error.set(t);
+ }
+ });
+
+ synchronized (auditLoader) {
+ assembleThread.start();
+ Assert.assertTrue(started.await(5, TimeUnit.SECONDS));
+ Assert.assertTrue(waitForBlocked(assembleThread));
+
Assert.assertFalse(getAuditLogBuffer(auditLoader).contains(auditEvent.queryId));
+ }
+
+ assembleThread.join(5000);
+ Assert.assertFalse(assembleThread.isAlive());
+ if (error.get() != null) {
+ throw new AssertionError("failed to assemble audit event",
error.get());
+ }
+
Assert.assertTrue(getAuditLogBuffer(auditLoader).contains(auditEvent.queryId));
+ }
+
+ private boolean waitForBlocked(Thread thread) throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 5000;
+ while (System.currentTimeMillis() < deadline) {
+ if (thread.getState() == Thread.State.BLOCKED) {
+ return true;
+ }
+ Thread.sleep(10);
+ }
+ return false;
+ }
+
+ private String getAuditLogBuffer(AuditLoader auditLoader) {
+ StringBuilder buffer = Deencapsulation.getField(auditLoader,
"auditLogBuffer");
+ return buffer.toString();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
