From 050bebb9a1a971184cb70e4d656742a641493945 Mon Sep 17 00:00:00 2001 From: wudi Date: Thu, 3 Sep 2026 18:11:39 +0800 Subject: [PATCH 1/2] [feature](fe) Add audit logs for S3 streaming insert tasks --- .../insert/streaming/StreamingInsertTask.java | 50 ++++- .../StreamingInsertTaskAuditTest.java | 181 ++++++++++++++++++ 2 files changed, 229 insertions(+), 2 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java index df23f10724049f..e0f45225cde163 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java @@ -20,8 +20,11 @@ import org.apache.doris.analysis.UserIdentity; import org.apache.doris.catalog.Env; import org.apache.doris.common.Config; +import org.apache.doris.common.ErrorCode; import org.apache.doris.common.FeConstants; +import org.apache.doris.common.Pair; import org.apache.doris.common.Status; +import org.apache.doris.common.util.DatasourcePrintableMap; import org.apache.doris.common.util.Util; import org.apache.doris.job.base.Job; import org.apache.doris.job.common.TaskStatus; @@ -30,16 +33,21 @@ import org.apache.doris.job.offset.SourceOffsetProvider; import org.apache.doris.load.loadv2.LoadJob; import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.analyzer.UnboundTVFRelation; import org.apache.doris.nereids.glue.LogicalPlanAdapter; import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.plans.commands.info.BaseViewInfo; import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.qe.AuditLogHelper; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.QueryState; import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.tablefunction.S3TableValuedFunction; import org.apache.doris.thrift.TCell; import org.apache.doris.thrift.TRow; import org.apache.doris.thrift.TStatusCode; +import com.google.common.base.Preconditions; import lombok.Getter; import lombok.extern.log4j.Log4j2; import org.apache.commons.lang3.StringUtils; @@ -49,6 +57,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.TreeMap; @Log4j2 @Getter @@ -61,6 +70,8 @@ public class StreamingInsertTask extends AbstractStreamingTask { private StreamingJobProperties jobProperties; private Map originTvfProps; private String cloudCluster; + private String auditSql; + private final boolean auditEnabled; SourceOffsetProvider offsetProvider; public StreamingInsertTask(long jobId, @@ -79,6 +90,7 @@ public StreamingInsertTask(long jobId, this.jobProperties = jobProperties; this.originTvfProps = originTvfProps; this.cloudCluster = cloudCluster; + this.auditEnabled = S3TableValuedFunction.NAME.equalsIgnoreCase(offsetProvider.getSourceType()); } @Override @@ -100,7 +112,14 @@ public void before() throws Exception { this.runningOffset = offsetProvider.getNextOffset(jobProperties, originTvfProps); log.info("streaming insert task {} get running offset: {}", taskId, runningOffset.toString()); - InsertIntoTableCommand baseCommand = (InsertIntoTableCommand) new NereidsParser().parseSingle(sql); + TreeMap, String> replacements = new TreeMap<>(new Pair.PairComparator<>()); + InsertIntoTableCommand baseCommand; + NereidsParser parser = new NereidsParser(); + if (auditEnabled) { + baseCommand = (InsertIntoTableCommand) parser.parseForEncryption(sql, replacements); + } else { + baseCommand = (InsertIntoTableCommand) parser.parseSingle(sql); + } baseCommand.setJobId(getTaskId()); StmtExecutor baseStmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(baseCommand, ctx.getStatementContext())); @@ -111,6 +130,10 @@ public void before() throws Exception { this.taskCommand = offsetProvider.rewriteTvfParams(baseCommand, runningOffset, getTaskId()); this.taskCommand.setLabelName(Optional.of(labelName)); this.stmtExecutor = new StmtExecutor(ctx, new LogicalPlanAdapter(taskCommand, ctx.getStatementContext())); + if (auditEnabled) { + this.auditSql = getAuditSql(replacements); + ctx.setExecutor(stmtExecutor); + } } @Override @@ -121,6 +144,9 @@ public void run() throws JobException { } log.info("start to run streaming insert task, label {}, offset is {}", labelName, runningOffset.toString()); String errMsg = null; + if (auditEnabled) { + ctx.setStartTime(); + } try { taskCommand.run(ctx, stmtExecutor); if (ctx.getState().getStateType() == QueryState.MysqlStateType.OK) { @@ -130,12 +156,32 @@ public void run() throws JobException { } throw new JobException(errMsg); } catch (Exception e) { + String errorMessage = Util.getRootCauseMessage(e); + if (auditEnabled && ctx.getState().getStateType() != QueryState.MysqlStateType.ERR) { + ctx.getState().setError(ErrorCode.ERR_INTERNAL_ERROR, errorMessage); + } log.warn("execute insert task error, label is {},offset is {}", taskCommand.getLabelName(), runningOffset.toString(), e); - throw new JobException(Util.getRootCauseMessage(e)); + throw new JobException(errorMessage); + } finally { + if (auditEnabled) { + AuditLogHelper.logAuditLog(ctx, auditSql, stmtExecutor.getParsedStmt(), + stmtExecutor.getQueryStatisticsForAuditLog(), true); + } } } + private String getAuditSql(TreeMap, String> replacements) { + List tvfRelations = taskCommand.getAllTVFRelation(); + Preconditions.checkState(replacements.size() == 1 && tvfRelations.size() == 1, + "S3 streaming insert must contain exactly one TVF"); + String rewrittenProperties = new DatasourcePrintableMap<>( + tvfRelations.get(0).getProperties().getMap(), "=", true, false, true).toString(); + Pair tvfPropertiesRange = replacements.firstKey(); + replacements.replace(tvfPropertiesRange, rewrittenProperties); + return BaseViewInfo.rewriteSql(replacements, sql); + } + @Override public List getScanBackendIds() { if (stmtExecutor != null && stmtExecutor.getCoord() != null) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java new file mode 100644 index 00000000000000..49083f0b8569f4 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java @@ -0,0 +1,181 @@ +// 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.job.extensions.insert.streaming; + +import org.apache.doris.analysis.StmtType; +import org.apache.doris.analysis.UserIdentity; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Pair; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.common.profile.SummaryProfile; +import org.apache.doris.datasource.CatalogMgr; +import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.job.exception.JobException; +import org.apache.doris.job.extensions.insert.InsertTask; +import org.apache.doris.job.offset.Offset; +import org.apache.doris.job.offset.SourceOffsetProvider; +import org.apache.doris.job.offset.jdbc.JdbcTvfSourceOffsetProvider; +import org.apache.doris.job.offset.s3.S3Offset; +import org.apache.doris.job.offset.s3.S3SourceOffsetProvider; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.analyzer.UnboundTVFRelation; +import org.apache.doris.nereids.glue.LogicalPlanAdapter; +import org.apache.doris.nereids.parser.NereidsParser; +import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.plugin.AuditEvent; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.resource.workloadschedpolicy.WorkloadRuntimeStatusMgr; + +import org.junit.Assert; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicLong; + +public class StreamingInsertTaskAuditTest { + private static final String ORIGIN_URI = "s3://bucket/input/*.csv"; + private static final String RESOLVED_URI = "s3://bucket/input/{1.csv,2.csv}"; + private static final String S3_SQL = "insert into target_table select * from s3(" + + "\"uri\" = \"" + ORIGIN_URI + "\", " + + "\"s3.secret_key\" = \"private-value\")"; + + @Test + public void testS3RunSubmitsAuditEvent() throws Exception { + AuditEvent auditEvent = runS3Task(null); + + Assert.assertEquals(AuditEvent.EventType.AFTER_QUERY, auditEvent.type); + Assert.assertEquals(StmtType.INSERT.name(), auditEvent.stmtType); + Assert.assertFalse(auditEvent.stmt.contains(ORIGIN_URI)); + Assert.assertTrue(auditEvent.stmt.contains(RESOLVED_URI)); + Assert.assertFalse(auditEvent.stmt.contains("private-value")); + Assert.assertEquals("OK", auditEvent.state); + Assert.assertTrue(auditEvent.isInternal); + } + + @Test + public void testFailedS3RunSubmitsErrorAuditEvent() throws Exception { + AuditEvent auditEvent = runS3Task(new RuntimeException("insert failed")); + + Assert.assertEquals(AuditEvent.EventType.AFTER_QUERY, auditEvent.type); + Assert.assertEquals(StmtType.INSERT.name(), auditEvent.stmtType); + Assert.assertEquals("ERR", auditEvent.state); + Assert.assertTrue(auditEvent.errorMessage.contains("insert failed")); + Assert.assertTrue(auditEvent.isInternal); + } + + @Test + public void testCdcRunDoesNotSubmitAuditEvent() throws Exception { + String sql = "insert into target_table select * from cdc_stream(" + + "\"type\" = \"mysql\", \"jdbc_url\" = \"jdbc:mysql://127.0.0.1:3306\", " + + "\"table\" = \"source_table\", \"offset\" = \"latest\")"; + runTask(null, sql, Mockito.mock(JdbcTvfSourceOffsetProvider.class), Mockito.mock(Offset.class), false); + } + + private AuditEvent runS3Task(RuntimeException commandFailure) throws Exception { + S3Offset offset = new S3Offset(); + offset.setFileLists(RESOLVED_URI); + return runTask(commandFailure, S3_SQL, new S3SourceOffsetProvider(), offset, true); + } + + private AuditEvent runTask(RuntimeException commandFailure, String sql, + SourceOffsetProvider offsetProvider, Offset offset, boolean expectAudit) throws Exception { + Env env = Mockito.mock(Env.class); + CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class); + InternalCatalog catalog = Mockito.mock(InternalCatalog.class); + WorkloadRuntimeStatusMgr statusMgr = Mockito.mock(WorkloadRuntimeStatusMgr.class); + Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr); + Mockito.when(env.getInternalCatalog()).thenReturn(catalog); + Mockito.when(catalogMgr.getCatalog(Mockito.anyString())).thenReturn(catalog); + Mockito.when(catalog.getName()).thenReturn("internal"); + Mockito.when(env.getWorkloadRuntimeStatusMgr()).thenReturn(statusMgr); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + + ConnectContext ctx = InsertTask.makeConnectContext(UserIdentity.ROOT, "test_db"); + ctx.getState().setOk(); + InsertIntoTableCommand command = Mockito.mock(InsertIntoTableCommand.class); + if (expectAudit) { + Map rewrittenTvfProps = new HashMap<>(); + rewrittenTvfProps.put("uri", RESOLVED_URI); + rewrittenTvfProps.put("s3.secret_key", "private-value"); + UnboundTVFRelation tvf = Mockito.mock(UnboundTVFRelation.class); + Mockito.when(tvf.getProperties()).thenReturn(new Properties(rewrittenTvfProps)); + Mockito.when(command.getAllTVFRelation()).thenReturn(Collections.singletonList(tvf)); + } + AtomicLong commandStartTime = new AtomicLong(); + Mockito.doAnswer(invocation -> { + commandStartTime.set(ctx.getStartTime()); + if (commandFailure != null) { + throw commandFailure; + } + return null; + }).when(command).run(Mockito.eq(ctx), Mockito.any(StmtExecutor.class)); + + LogicalPlanAdapter parsedStmt = new LogicalPlanAdapter( + new NereidsParser().parseSingle(sql), new StatementContext()); + StmtExecutor executor = Mockito.mock(StmtExecutor.class); + Mockito.when(executor.getParsedStmt()).thenReturn(parsedStmt); + Mockito.when(executor.getExternalDmlAuditBackendIds()).thenReturn(Collections.emptySet()); + Mockito.when(executor.getSummaryProfile()).thenReturn(Mockito.mock(SummaryProfile.class)); + ctx.setExecutor(executor); + + StreamingInsertTask task = new StreamingInsertTask( + 1L, 2L, sql, offsetProvider, "test_db", null, + Collections.emptyMap(), UserIdentity.ROOT, null); + Deencapsulation.setField(task, "ctx", ctx); + Deencapsulation.setField(task, "taskCommand", command); + Deencapsulation.setField(task, "stmtExecutor", executor); + Deencapsulation.setField(task, "runningOffset", offset); + if (expectAudit) { + TreeMap, String> replacements = + new TreeMap<>(new Pair.PairComparator<>()); + new NereidsParser().parseForEncryption(sql, replacements); + Deencapsulation.setField(task, "auditSql", + Deencapsulation.invoke(task, "getAuditSql", replacements)); + } + + if (commandFailure == null) { + task.run(); + } else { + Assert.assertThrows(JobException.class, task::run); + } + + if (!expectAudit) { + Mockito.verifyNoInteractions(statusMgr); + return null; + } + ArgumentCaptor auditEventCaptor = ArgumentCaptor.forClass(AuditEvent.class); + Mockito.verify(statusMgr).submitFinishQueryToAudit(auditEventCaptor.capture()); + AuditEvent auditEvent = auditEventCaptor.getValue(); + Assert.assertTrue(commandStartTime.get() > 0); + Assert.assertEquals(commandStartTime.get(), auditEvent.timestamp); + return auditEvent; + } finally { + ConnectContext.remove(); + } + } +} From 7771587d3a85c3ed1c60c6bded10af419dbf4a4f Mon Sep 17 00:00:00 2001 From: wudi Date: Fri, 4 Sep 2026 11:23:43 +0800 Subject: [PATCH 2/2] [improvement](fe) Improve S3 streaming insert audit SQL --- .../insert/streaming/StreamingInsertTask.java | 9 +++-- .../job/offset/s3/S3SourceOffsetProvider.java | 2 +- .../StreamingInsertTaskAuditTest.java | 34 +++++++++++++++++-- 3 files changed, 39 insertions(+), 6 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java index e0f45225cde163..2a4799d124a490 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java @@ -24,7 +24,6 @@ import org.apache.doris.common.FeConstants; import org.apache.doris.common.Pair; import org.apache.doris.common.Status; -import org.apache.doris.common.util.DatasourcePrintableMap; import org.apache.doris.common.util.Util; import org.apache.doris.job.base.Job; import org.apache.doris.job.common.TaskStatus; @@ -38,6 +37,7 @@ import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.trees.plans.commands.info.BaseViewInfo; import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; +import org.apache.doris.nereids.util.SqlLiteralUtils; import org.apache.doris.qe.AuditLogHelper; import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.QueryState; @@ -58,6 +58,7 @@ import java.util.Map; import java.util.Optional; import java.util.TreeMap; +import java.util.stream.Collectors; @Log4j2 @Getter @@ -175,8 +176,10 @@ private String getAuditSql(TreeMap, String> replacements) List tvfRelations = taskCommand.getAllTVFRelation(); Preconditions.checkState(replacements.size() == 1 && tvfRelations.size() == 1, "S3 streaming insert must contain exactly one TVF"); - String rewrittenProperties = new DatasourcePrintableMap<>( - tvfRelations.get(0).getProperties().getMap(), "=", true, false, true).toString(); + String rewrittenProperties = tvfRelations.get(0).getProperties().getMap().entrySet().stream() + .map(entry -> SqlLiteralUtils.quoteStringLiteral(entry.getKey()) + " = " + + SqlLiteralUtils.quoteStringLiteral(entry.getValue())) + .collect(Collectors.joining(", ")); Pair tvfPropertiesRange = replacements.firstKey(); replacements.replace(tvfPropertiesRange, rewrittenProperties); return BaseViewInfo.rewriteSql(replacements, sql); diff --git a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java index 37e3d3dfbf0437..2777013471cda7 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java +++ b/fe/fe-core/src/main/java/org/apache/doris/job/offset/s3/S3SourceOffsetProvider.java @@ -129,7 +129,7 @@ public String getShowMaxOffset() { public InsertIntoTableCommand rewriteTvfParams(InsertIntoTableCommand originCommand, Offset runningOffset, long taskId) { S3Offset offset = (S3Offset) runningOffset; - Map props = new HashMap<>(); + Map props = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER); // rewrite plan Plan rewritePlan = originCommand.getParsedPlan().get().rewriteUp(plan -> { if (plan instanceof UnboundTVFRelation) { diff --git a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java index 49083f0b8569f4..b134e528a0bbce 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTaskAuditTest.java @@ -37,6 +37,7 @@ import org.apache.doris.nereids.glue.LogicalPlanAdapter; import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.trees.expressions.Properties; +import org.apache.doris.nereids.trees.plans.RelationId; import org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand; import org.apache.doris.plugin.AuditEvent; import org.apache.doris.qe.ConnectContext; @@ -52,6 +53,7 @@ import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.TreeMap; import java.util.concurrent.atomic.AtomicLong; @@ -60,7 +62,8 @@ public class StreamingInsertTaskAuditTest { private static final String RESOLVED_URI = "s3://bucket/input/{1.csv,2.csv}"; private static final String S3_SQL = "insert into target_table select * from s3(" + "\"uri\" = \"" + ORIGIN_URI + "\", " - + "\"s3.secret_key\" = \"private-value\")"; + + "\"s3.secret_key\" = \"private-value\", " + + "\"enclose\" = \"\\\"\")"; @Test public void testS3RunSubmitsAuditEvent() throws Exception { @@ -94,6 +97,31 @@ public void testCdcRunDoesNotSubmitAuditEvent() throws Exception { runTask(null, sql, Mockito.mock(JdbcTvfSourceOffsetProvider.class), Mockito.mock(Offset.class), false); } + @Test + public void testRewriteS3UriCaseInsensitive() { + Map originProperties = new HashMap<>(); + originProperties.put("URI", ORIGIN_URI); + UnboundTVFRelation originTvf = new UnboundTVFRelation( + new RelationId(1), "s3", new Properties(originProperties)); + InsertIntoTableCommand originCommand = Mockito.mock(InsertIntoTableCommand.class); + Mockito.when(originCommand.getParsedPlan()).thenReturn(Optional.of(originTvf)); + + S3Offset offset = new S3Offset(); + offset.setFileLists(RESOLVED_URI); + Env env = Mockito.mock(Env.class); + Mockito.when(env.isMaster()).thenReturn(false); + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + InsertIntoTableCommand rewritten = new S3SourceOffsetProvider() + .rewriteTvfParams(originCommand, offset, 1L); + + Map rewrittenProperties = + rewritten.getAllTVFRelation().get(0).getProperties().getMap(); + Assert.assertEquals(1, rewrittenProperties.size()); + Assert.assertEquals(RESOLVED_URI, rewrittenProperties.get("URI")); + } + } + private AuditEvent runS3Task(RuntimeException commandFailure) throws Exception { S3Offset offset = new S3Offset(); offset.setFileLists(RESOLVED_URI); @@ -122,6 +150,7 @@ private AuditEvent runTask(RuntimeException commandFailure, String sql, Map rewrittenTvfProps = new HashMap<>(); rewrittenTvfProps.put("uri", RESOLVED_URI); rewrittenTvfProps.put("s3.secret_key", "private-value"); + rewrittenTvfProps.put("enclose", "\""); UnboundTVFRelation tvf = Mockito.mock(UnboundTVFRelation.class); Mockito.when(tvf.getProperties()).thenReturn(new Properties(rewrittenTvfProps)); Mockito.when(command.getAllTVFRelation()).thenReturn(Collections.singletonList(tvf)); @@ -165,7 +194,8 @@ private AuditEvent runTask(RuntimeException commandFailure, String sql, } if (!expectAudit) { - Mockito.verifyNoInteractions(statusMgr); + Mockito.verify(statusMgr, Mockito.never()) + .submitFinishQueryToAudit(Mockito.any(AuditEvent.class)); return null; } ArgumentCaptor auditEventCaptor = ArgumentCaptor.forClass(AuditEvent.class);