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 c56efafbcd9d8..b75097a50cda7 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 3c4cc12959622..f03d649ac81d8 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 4cca55d705f63..46cbf4f1b8279 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 @@ private TSStatus executeStatementAndClassifyExceptions( } 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 @@ private void logStatementExceptionIfNecessary(final Statement statement, final E } } - 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 f6fe91016baf1..7a2ba2ef0f4b2 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 @@ protected void subRemoveAttributeColumns(List columnsToKeep) { 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 8f2e86c62d057..59b247d4d3431 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 void testLoadTsFileSyncStatementKeepsDefaultDatabaseLevelWhenDatabaseName } @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