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 89dbf8f544c [To dev/1.3] [Pipe] Improve legacy receiver error messages
and diagnostics (#18932) (#18949)
89dbf8f544c is described below
commit 89dbf8f544c271b663f897ea967c0a06e63a1409
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 21 16:07:25 2026 +0800
[To dev/1.3] [Pipe] Improve legacy receiver error messages and diagnostics
(#18932) (#18949)
* [To dev/1.3] [Pipe] Improve legacy receiver error messages and diagnostics
Backport the Pipe portion of 86ae95729ab5bf205aa88221d6e7cff1dc99c75d
(#18932), including localized diagnostics and regression coverage.
* [Pipe] Declare runtime Logback dependencies for legacy receiver log
assertions
---
iotdb-core/datanode/pom.xml | 16 ++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 37 ++++++++++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 36 ++++++++++++++
.../legacy/IoTDBLegacyPipeReceiverAgent.java | 42 +++++++++++-----
.../legacy/IoTDBLegacyPipeReceiverAgentTest.java | 57 +++++++++++++++++++++-
5 files changed, 175 insertions(+), 13 deletions(-)
diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml
index ef9e3a03727..d57eb8d9ebb 100644
--- a/iotdb-core/datanode/pom.xml
+++ b/iotdb-core/datanode/pom.xml
@@ -171,6 +171,16 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-core</artifactId>
+ <scope>runtime</scope>
+ </dependency>
<dependency>
<groupId>net.java.dev.jna</groupId>
<artifactId>jna</artifactId>
@@ -542,6 +552,12 @@
<!-- For some reason this plugin missed it being used
for a constant import -->
<ignoredDependency>org.apache.iotdb:isession</ignoredDependency>
</ignoredDependencies>
+ <ignoredNonTestScopedDependencies>
+ <!-- Logback supplies the server's runtime SLF4J
backend and LogbackMetrics.
+ Direct references in this module also support
test log assertions. -->
+
<ignoredNonTestScopedDependency>ch.qos.logback:logback-classic</ignoredNonTestScopedDependency>
+
<ignoredNonTestScopedDependency>ch.qos.logback:logback-core</ignoredNonTestScopedDependency>
+ </ignoredNonTestScopedDependencies>
<usedDependencies>
<!-- This dependency is required at runtime, when
enabling the rest service -->
<usedDependency>org.glassfish.jersey.inject:jersey-hk2</usedDependency>
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
new file mode 100644
index 00000000000..db51ff09b00
--- /dev/null
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -0,0 +1,37 @@
+/*
+ * 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.iotdb.db.i18n;
+
+public final class DataNodePipeMessages {
+
+ private DataNodePipeMessages() {}
+
+ // ===================== LEGACY PIPE RECEIVER RESPONSES =====================
+
+ public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 =
+ "Pipe data transport error, %s";
+ public static final String
MESSAGE_FAIL_TO_LOAD_PIPEDATA_BECAUSE_ARG_49526D0C =
+ "Fail to load pipeData because %s";
+ public static final String
MESSAGE_FAILED_TO_RESOLVE_LEGACY_PIPE_RECEIVER_FILE_ARG_ARG_9B82CF3F =
+ "Failed to resolve legacy pipe receiver file %s: %s";
+ public static final String
+
MESSAGE_FAILED_TO_APPEND_LEGACY_PIPE_RECEIVER_FILE_ARG_AT_OFFSET_ARG_WITH_ARG_BYTES_ARG_57428A06
=
+ "Failed to append legacy pipe receiver file %s at offset %d with %d
bytes: %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
new file mode 100644
index 00000000000..053fc9cf69f
--- /dev/null
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -0,0 +1,36 @@
+/*
+ * 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.iotdb.db.i18n;
+
+public final class DataNodePipeMessages {
+
+ private DataNodePipeMessages() {}
+
+ // ===================== LEGACY PIPE RECEIVER RESPONSES =====================
+
+ public static final String MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238 =
"Pipe 数据传输错误：%s";
+ public static final String
MESSAGE_FAIL_TO_LOAD_PIPEDATA_BECAUSE_ARG_49526D0C =
+ "加载 pipeData 失败，原因：%s";
+ public static final String
MESSAGE_FAILED_TO_RESOLVE_LEGACY_PIPE_RECEIVER_FILE_ARG_ARG_9B82CF3F =
+ "解析 legacy pipe receiver 文件 %s 失败：%s";
+ public static final String
+
MESSAGE_FAILED_TO_APPEND_LEGACY_PIPE_RECEIVER_FILE_ARG_AT_OFFSET_ARG_WITH_ARG_BYTES_ARG_57428A06
=
+ "追加 legacy pipe receiver 文件 %s 失败，偏移量 %d，写入 %d 字节：%s";
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
index 655bd0a1e19..872c967ae01 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgent.java
@@ -24,6 +24,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.pipe.receiver.PipeReceiverFilePathUtils;
import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.db.protocol.session.SessionManager;
@@ -158,9 +159,11 @@ public class IoTDBLegacyPipeReceiverAgent {
handleTsFilePipeData(tsFilePipeData, fileDir);
}
} catch (IOException e) {
- LOGGER.error("Pipe data transport error, {}", e.getMessage());
- return RpcUtils.getStatus(
- TSStatusCode.PIPESERVER_ERROR, "Pipe data transport error, " +
e.getMessage());
+ final String errorMessage =
+ String.format(
+
DataNodePipeMessages.MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238,
e.getMessage());
+ LOGGER.error(errorMessage, e);
+ return RpcUtils.getStatus(TSStatusCode.PIPESERVER_ERROR, errorMessage);
}
// step3. load PipeData
@@ -174,9 +177,12 @@ public class IoTDBLegacyPipeReceiverAgent {
LOGGER.info(
"Load pipeData with serialize number {} successfully.",
pipeData.getSerialNumber());
} catch (PipeException e) {
- LOGGER.error("Fail to load pipeData because {}.", e.getMessage());
- return RpcUtils.getStatus(
- TSStatusCode.PIPESERVER_ERROR, "Fail to load pipeData because " +
e.getMessage());
+ final String errorMessage =
+ String.format(
+
DataNodePipeMessages.MESSAGE_FAIL_TO_LOAD_PIPEDATA_BECAUSE_ARG_49526D0C,
+ e.getMessage());
+ LOGGER.error(errorMessage, e);
+ return RpcUtils.getStatus(TSStatusCode.PIPESERVER_ERROR, errorMessage);
}
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "");
@@ -263,8 +269,14 @@ public class IoTDBLegacyPipeReceiverAgent {
fileWithoutPatch = resolveFileInFileDataDir(fileDir, fileName);
file = resolveFileInFileDataDir(fileDir, fileName + PATCH_SUFFIX);
} catch (final IOException e) {
- LOGGER.warn(e.getMessage());
- return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, e.getMessage());
+ final String errorMessage =
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_FAILED_TO_RESOLVE_LEGACY_PIPE_RECEIVER_FILE_ARG_ARG_9B82CF3F,
+ fileName,
+ e.getMessage());
+ LOGGER.warn(errorMessage, e);
+ return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, errorMessage);
}
// step2. check startIndex
@@ -274,8 +286,8 @@ public class IoTDBLegacyPipeReceiverAgent {
}
// step3. append file
+ final int length = buff.remaining();
try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw"))
{
- int length = buff.remaining();
randomAccessFile.seek(startIndex);
byte[] byteArray = new byte[length];
buff.get(byteArray);
@@ -283,8 +295,16 @@ public class IoTDBLegacyPipeReceiverAgent {
recordStartIndex(fileWithoutPatch, startIndex + length);
LOGGER.debug("Sync {} start at {} to {} is done.", fileName, startIndex,
startIndex + length);
} catch (final IOException e) {
- LOGGER.error(e.getMessage());
- return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, e.getMessage());
+ final String errorMessage =
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_FAILED_TO_APPEND_LEGACY_PIPE_RECEIVER_FILE_ARG_AT_OFFSET_ARG_WITH_ARG_BYTES_ARG_57428A06,
+ fileName,
+ startIndex,
+ length,
+ e.getMessage());
+ LOGGER.error(errorMessage, e);
+ return RpcUtils.getStatus(TSStatusCode.SYNC_FILE_ERROR, errorMessage);
}
return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS, "");
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java
index eaf21aef1fd..2693e182f6e 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/receiver/protocol/legacy/IoTDBLegacyPipeReceiverAgentTest.java
@@ -21,16 +21,21 @@ package org.apache.iotdb.db.pipe.receiver.protocol.legacy;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
import org.apache.commons.io.FileUtils;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.slf4j.LoggerFactory;
import java.io.File;
import java.nio.ByteBuffer;
@@ -86,7 +91,15 @@ public class IoTDBLegacyPipeReceiverAgentTest {
ByteBuffer.wrap("pwned".getBytes(StandardCharsets.UTF_8)));
Assert.assertEquals(TSStatusCode.SYNC_FILE_ERROR.getStatusCode(),
status.getCode());
- Assert.assertTrue(status.getMessage().contains("Illegal fileName"));
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .startsWith(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_FAILED_TO_RESOLVE_LEGACY_PIPE_RECEIVER_FILE_ARG_ARG_9B82CF3F,
+ traversal,
+ "Illegal fileName: " + traversal)));
Assert.assertFalse(Files.exists(syncDir.resolve("pwned.patch")));
}
@@ -103,6 +116,40 @@ public class IoTDBLegacyPipeReceiverAgentTest {
Assert.assertArrayEquals(payload, Files.readAllBytes(patchFile));
}
+ @Test
+ public void testTransportFileWriteFailureIncludesContextAndLogsException()
throws Exception {
+ final String fileName = "1-2-3-4.tsfile";
+ final byte[] payload = "iotdb".getBytes(StandardCharsets.UTF_8);
+ Files.createDirectory(getFileDataDir().resolve(fileName + ".patch"));
+ final Logger logger = (Logger)
LoggerFactory.getLogger(IoTDBLegacyPipeReceiverAgent.class);
+ final ListAppender<ILoggingEvent> appender = new ListAppender<>();
+ appender.start();
+ logger.addAppender(appender);
+ try {
+ final TSStatus status =
+ agent.transportFile(new TSyncTransportMetaInfo(fileName, 0),
ByteBuffer.wrap(payload));
+
+ Assert.assertEquals(TSStatusCode.SYNC_FILE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .startsWith(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_FAILED_TO_APPEND_LEGACY_PIPE_RECEIVER_FILE_ARG_AT_OFFSET_ARG_WITH_ARG_BYTES_ARG_57428A06,
+ fileName,
+ 0,
+ payload.length,
+ "")));
+ final ILoggingEvent failure = appender.list.get(appender.list.size() -
1);
+ Assert.assertEquals(status.getMessage(), failure.getFormattedMessage());
+ Assert.assertNotNull(failure.getThrowableProxy());
+ } finally {
+ logger.detachAppender(appender);
+ appender.stop();
+ }
+ }
+
@Test
public void testTransportPipeDataRejectsPathTraversalTsFileName() throws
Exception {
final String traversal = ".." + File.separator + "evil.tsfile";
@@ -111,7 +158,13 @@ public class IoTDBLegacyPipeReceiverAgentTest {
agent.transportPipeData(ByteBuffer.wrap(new TsFilePipeData("",
traversal, -1).serialize()));
Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(),
status.getCode());
- Assert.assertTrue(status.getMessage().contains("Illegal fileName"));
+ Assert.assertTrue(
+ status
+ .getMessage()
+ .startsWith(
+ String.format(
+
DataNodePipeMessages.MESSAGE_PIPE_DATA_TRANSPORT_ERROR_ARG_ECE6A238,
+ "Illegal fileName: " + traversal)));
}
private Path getFileDataDir() {
