This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 8062f7d5e3f Prevent pipe receiver statement exception log floods
(#18318) (#18325)
8062f7d5e3f is described below
commit 8062f7d5e3fbebc13527dbd29b6f8e508d17d776
Author: Caideyipi <[email protected]>
AuthorDate: Wed Jul 29 10:03:27 2026 +0800
Prevent pipe receiver statement exception log floods (#18318) (#18325)
(cherry picked from commit 11aa5069a6d260394c1db0c17e6a0a2ff1221d8a)
---
.../protocol/thrift/IoTDBDataNodeReceiver.java | 33 +++++---
.../plan/statement/crud/InsertRowsStatement.java | 18 +++++
.../protocol/thrift/IoTDBDataNodeReceiverTest.java | 88 +++++++++++++++++-----
3 files changed, 109 insertions(+), 30 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index f2a05a72798..6ad15376f85 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -117,6 +117,8 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static
org.apache.iotdb.commons.utils.ErrorHandlingCommonUtils.getRootCause;
+
public class IoTDBDataNodeReceiver extends IoTDBFileReceiver {
private static final Logger LOGGER =
LoggerFactory.getLogger(IoTDBDataNodeReceiver.class);
@@ -916,7 +918,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
}
private void logStatementExceptionIfNecessary(final Statement statement,
final Exception e) {
- if (shouldLogStatementException(receiverId.get(), statement, e)) {
+ if (shouldLogStatementException(statement, e)) {
PipeLogger.log(
LOGGER::warn,
e,
@@ -926,16 +928,25 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
}
}
- static boolean shouldLogStatementException(
- final long receiverId, final Statement statement, final Exception e) {
- // Use the reducer cache as a gate. The actual stack trace is logged only
when it passes.
- return LoggerPeriodicalLogReducer.log(
- message -> {},
- "Receiver id = %s, statement = %s, exception = %s, message = %s",
- receiverId,
- Objects.isNull(statement) ? null : statement.getPipeLoggingString(),
- e.getClass().getName(),
- e.getMessage());
+ static boolean shouldLogStatementException(final Statement statement, final
Exception e) {
+ final Throwable rootCause = getRootCause(e);
+ final StackTraceElement[] rootCauseStackTrace = rootCause.getStackTrace();
+ final StackTraceElement rootCauseLocation =
+ rootCauseStackTrace.length > 0 ? rootCauseStackTrace[0] : null;
+
+ // Reduce exceptions raised from the same code location regardless of
receiver, statement
+ // content, or dynamic exception message. Different statement types and
failure locations are
+ // still logged independently.
+ return LoggerPeriodicalLogReducer.shouldLog(
+ IoTDBDataNodeReceiver.class.getName()
+ + '|'
+ + (Objects.isNull(statement)
+ ? Statement.class.getName()
+ : statement.getClass().getName())
+ + '|'
+ + rootCause.getClass().getName()
+ + '|'
+ + rootCauseLocation);
}
private TSStatus executeStatementWithRetryOnDataTypeMismatch(final Statement
statement) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java
index 0e7b65371b5..0a4a84ad032 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/crud/InsertRowsStatement.java
@@ -180,6 +180,24 @@ public class InsertRowsStatement extends
InsertBaseStatement {
: 0);
}
+ @Override
+ public String getPipeLoggingString() {
+ if (Objects.isNull(insertRowStatementList) ||
insertRowStatementList.isEmpty()) {
+ return "InsertRowsStatement{rowCount=0}";
+ }
+
+ final int rowCount = insertRowStatementList.size();
+ return "InsertRowsStatement{"
+ + "rowCount="
+ + rowCount
+ + ", firstRow="
+ + insertRowStatementList.get(0).getPipeLoggingString()
+ + (rowCount > 1
+ ? ", lastRow=" + insertRowStatementList.get(rowCount -
1).getPipeLoggingString()
+ : "")
+ + '}';
+ }
+
@Override
public String toString() {
return "InsertRowsStatement{" + "insertRowStatementList=" +
insertRowStatementList + '}';
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
index 864d0493979..a2710535f70 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiverTest.java
@@ -20,6 +20,8 @@
package org.apache.iotdb.db.pipe.receiver.protocol.thrift;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertRowsStatement;
import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
import org.apache.iotdb.db.storageengine.load.active.ActiveLoadPathHelper;
import org.apache.iotdb.db.storageengine.load.config.LoadTsFileConfigurator;
@@ -29,6 +31,7 @@ import org.junit.Test;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Arrays;
import java.util.Map;
public class IoTDBDataNodeReceiverTest {
@@ -89,25 +92,72 @@ public class IoTDBDataNodeReceiverTest {
}
@Test
- public void testRepeatedStatementExceptionLogIsReduced() throws Exception {
- final Path tsFile = Files.createTempFile("pipe-load-log-reducer",
".tsfile");
- try {
- final LoadTsFileStatement statement =
- IoTDBDataNodeReceiver.buildLoadTsFileStatementForSync(
- "root.test.sg_0", tsFile.toString(), true, true);
- final long receiverId = System.nanoTime();
- final Exception exception = new RuntimeException("repeated receiver
exception " + receiverId);
-
- Assert.assertTrue(
- IoTDBDataNodeReceiver.shouldLogStatementException(receiverId,
statement, exception));
- Assert.assertFalse(
- IoTDBDataNodeReceiver.shouldLogStatementException(receiverId,
statement, exception));
- Assert.assertTrue(
- IoTDBDataNodeReceiver.shouldLogStatementException(
- receiverId, statement, new RuntimeException("another receiver
exception")));
- } finally {
- Files.deleteIfExists(tsFile);
- }
+ public void testStatementExceptionLogIsReducedByFailureLocation() {
+ final InsertRowStatement firstStatement = new InsertRowStatement();
+ firstStatement.setTime(1);
+ firstStatement.setValues(new Object[] {"first statement value"});
+ final InsertRowStatement secondStatement = new InsertRowStatement();
+ secondStatement.setTime(2);
+ secondStatement.setValues(new Object[] {"second statement value"});
+
+ final String testId = Long.toString(System.nanoTime());
+ final String sameFailureLocation = "sameFailureLocation" + testId;
+ Assert.assertTrue(
+ IoTDBDataNodeReceiver.shouldLogStatementException(
+ firstStatement, newStatementException("first message",
sameFailureLocation)));
+ Assert.assertFalse(
+ IoTDBDataNodeReceiver.shouldLogStatementException(
+ secondStatement, newStatementException("second message",
sameFailureLocation)));
+ Assert.assertTrue(
+ IoTDBDataNodeReceiver.shouldLogStatementException(
+ secondStatement,
+ newStatementException("second message", "differentFailureLocation"
+ testId)));
+ Assert.assertTrue(
+ IoTDBDataNodeReceiver.shouldLogStatementException(
+ new InsertRowsStatement(),
+ newStatementException("second message", sameFailureLocation)));
+ }
+
+ @Test
+ public void testInsertRowsPipeLoggingStringIsCompact() {
+ final InsertRowStatement firstStatement = new InsertRowStatement();
+ firstStatement.setTime(1);
+ firstStatement.setValues(new Object[] {"first secret value"});
+ final InsertRowStatement middleStatement = new InsertRowStatement();
+ middleStatement.setTime(2);
+ middleStatement.setValues(new Object[] {"middle secret value"});
+ final InsertRowStatement lastStatement = new InsertRowStatement();
+ lastStatement.setTime(3);
+ lastStatement.setValues(new Object[] {"last secret value"});
+
+ final InsertRowsStatement statement = new InsertRowsStatement();
+ statement.setInsertRowStatementList(
+ Arrays.asList(firstStatement, middleStatement, lastStatement));
+
+ final String pipeLoggingString = statement.getPipeLoggingString();
+ Assert.assertTrue(pipeLoggingString.contains("rowCount=3"));
+ Assert.assertTrue(pipeLoggingString.contains("firstRow="));
+ Assert.assertTrue(pipeLoggingString.contains("time=1"));
+ Assert.assertTrue(pipeLoggingString.contains("lastRow="));
+ Assert.assertTrue(pipeLoggingString.contains("time=3"));
+ Assert.assertFalse(pipeLoggingString.contains("time=2"));
+ Assert.assertFalse(pipeLoggingString.contains("first secret value"));
+ Assert.assertFalse(pipeLoggingString.contains("middle secret value"));
+ Assert.assertFalse(pipeLoggingString.contains("last secret value"));
+ }
+
+ private static Exception newStatementException(
+ final String message, final String failureLocation) {
+ final NullPointerException rootCause = new NullPointerException(message);
+ rootCause.setStackTrace(
+ new StackTraceElement[] {
+ new StackTraceElement(
+ IoTDBDataNodeReceiverTest.class.getName(),
+ failureLocation,
+ "IoTDBDataNodeReceiverTest.java",
+ 1)
+ });
+ return new RuntimeException("wrapper " + message, rootCause);
}
@Test
