This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 11aa5069a6d Prevent pipe receiver statement exception log floods
(#18318)
11aa5069a6d is described below
commit 11aa5069a6d260394c1db0c17e6a0a2ff1221d8a
Author: Caideyipi <[email protected]>
AuthorDate: Tue Jul 28 10:18:16 2026 +0800
Prevent pipe receiver statement exception log floods (#18318)
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 -
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 -
.../protocol/thrift/IoTDBDataNodeReceiver.java | 31 +++++---
.../plan/statement/crud/InsertRowsStatement.java | 18 +++++
.../protocol/thrift/IoTDBDataNodeReceiverTest.java | 85 +++++++++++++++++-----
5 files changed, 104 insertions(+), 34 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index c56efafbcd9..b75097a50cd 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -1318,8 +1318,6 @@ public final class DataNodePipeMessages {
"Receiver id = {}: Failure status encountered while executing statement
{}: {}";
public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
"Receiver id = {}: Exception encountered while executing statement {}: ";
- public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
- "Receiver id = {}, statement = {}, exception = {}, message = {}";
public static final String UNKNOWN_PIPEREQUESTTYPE = "Unknown
PipeRequestType %s.";
public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
"Exception %s encountered while handling request %s.";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 3c4cc129596..f03d649ac81 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -1237,8 +1237,6 @@ public final class DataNodePipeMessages {
"Receiver id = {}: 执行 statement {} 时遇到失败状态：{}";
public static final String RECEIVER_ID_EXCEPTION_WHILE_EXECUTING_STATEMENT =
"Receiver id = {}: 执行 statement {} 时遇到异常：";
- public static final String RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE =
- "Receiver id = {}，statement = {}，exception = {}，message = {}";
public static final String UNKNOWN_PIPEREQUESTTYPE = "未知 PipeRequestType
%s。";
public static final String EXCEPTION_ENCOUNTERED_WHILE_HANDLING_REQUEST =
"遇到异常 %s，处理请求 %s 时。";
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 4cca55d705f..46cbf4f1b82 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
@@ -1080,7 +1080,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,
@@ -1090,16 +1090,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 -> {},
- DataNodePipeMessages.RECEIVER_ID_STATEMENT_EXCEPTION_MESSAGE,
- 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
executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(
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 f6fe91016ba..7a2ba2ef0f4 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
@@ -246,6 +246,24 @@ public class InsertRowsStatement extends
InsertBaseStatement {
insertRowStatementList.forEach(InsertBaseStatement::removeAttributeColumns);
}
+ @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 8f2e86c62d0..59b247d4d34 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
@@ -95,25 +95,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
