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 86ae95729ab [Subscription][Pipe] Improve receiver error messages and
diagnostics (#18932)
86ae95729ab is described below
commit 86ae95729ab5bf205aa88221d6e7cff1dc99c75d
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 20 19:18:09 2026 +0800
[Subscription][Pipe] Improve receiver error messages and diagnostics
(#18932)
* [Subscription][Pipe] Improve receiver error messages and diagnostics
* [Subscription] Explain rejected request slices
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 83 +++++++++++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 83 +++++++++++
.../legacy/IoTDBLegacyPipeReceiverAgent.java | 41 ++++--
.../receiver/SubscriptionReceiverV1.java | 157 ++++++++++++++-------
.../legacy/IoTDBLegacyPipeReceiverAgentTest.java | 58 +++++++-
.../receiver/SubscriptionReceiverV1Test.java | 151 +++++++++++++++++++-
6 files changed, 505 insertions(+), 68 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 e676d5c93e9..374a35e1642 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
@@ -2710,4 +2710,87 @@ public final class DataNodePipeMessages {
"[%s] Subscription entry requires %d bytes, above the current per-queue
maximum %d bytes (DataNode budget %d bytes). Reduce the write batch/field size
or increase subscription materialization memory. WAL progress has not
advanced.";
public static final String
LOG_SUBSCRIPTION_FAILED_TO_REPLAY_DELIVERY_BATCH_ARG_FROM_WAL_RETAINING_UNCOMMITTED_STATE_0654EACD
=
"Subscription failed to replay delivery batch {} from WAL; retaining
uncommitted state";
+
+ // ===================== SUBSCRIPTION RECEIVER RESPONSES
=====================
+
+ public static final String
+
MESSAGE_SUBSCRIPTION_MALFORMED_REQUEST_SLICE_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_D91FD262
=
+ "Subscription: malformed request slice. Resend the complete request
starting from slice index 0.";
+ public static final String
+
MESSAGE_SUBSCRIPTION_INVALID_OR_OUT_OF_ORDER_REQUEST_SLICE_ORDERID_ARG_SLICEINDEX_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_BYTES_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_A08D6D8E
=
+ "Subscription: invalid or out-of-order request slice (orderId %d,
sliceIndex %d, sliceCount %d, originBodySize %d bytes). Resend the complete
request starting from slice index 0.";
+ public static final String
+
MESSAGE_SUBSCRIPTION_REQUEST_BODY_SIZE_ARG_BYTES_EXCEEDS_THE_RECEIVER_LIMIT_OF_ARG_BYTES_SPLIT_THE_OPERATION_INTO_SMALLER_REQUESTS_C9513027
=
+ "Subscription: request body size %d bytes exceeds the receiver limit
of %d bytes. Split the operation into smaller requests.";
+
+ public static final String
MESSAGE_MISSING_CONSUMER_CONFIG_PLEASE_HANDSHAKE_FIRST_CE588329 =
+ "Missing consumer config, please handshake first.";
+ public static final String
MESSAGE_UNKNOWN_PIPESUBSCRIBEREQUESTTYPE_ARG_7EBB93F8 =
+ "Unknown PipeSubscribeRequestType %s.";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HANDSHAKING_WITH_REQUEST_ARG_ARG_C1BC903B
=
+ "Subscription: something unexpected happened when handshaking with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HEARTBEAT_WITH_REQUEST_ARG_ARG_790B328F
=
+ "Subscription: something unexpected happened when heartbeat with
request %s: %s";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_FETCH_ENDPOINTS_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_267904CB
=
+ "Subscription: Failed to fetch endpoints for consumer %s in config
node, exception is %s.";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SUBSCRIBING_WITH_REQUEST_ARG_ARG_084648A4
=
+ "Subscription: something unexpected happened when subscribing with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_UNSUBSCRIBING_WITH_REQUEST_ARG_ARG_1C15D8EC
=
+ "Subscription: something unexpected happened when unsubscribing with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_POLLING_WITH_REQUEST_ARG_ARG_DE2CF98B
=
+ "Subscription: something unexpected happened when polling with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_COMMITTING_WITH_REQUEST_ARG_ARG_9AA9866F
=
+ "Subscription: something unexpected happened when committing with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_CLOSING_WITH_REQUEST_ARG_ARG_1E3215D2
=
+ "Subscription: something unexpected happened when closing with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SEEKING_WITH_REQUEST_ARG_ARG_97D0AD1C
=
+ "Subscription: something unexpected happened when seeking with
request %s: %s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_UNSUPPORTED_SEEKTYPE_ARG_FOR_TOPIC_ARG_CONSENSUS_SUBSCRIPTION_ONLY_SUPPORTS_SEEKTOBEGINNING_SEEKTOEND_SEEK_TOPICPROGRESS_AND_SEEKAFTER_TOPICPROGRESS_5A3436C3
=
+ "Subscription: unsupported seekType %s for topic %s. Consensus
subscription only supports seekToBeginning, seekToEnd, seek(topicProgress), and
seekAfter(topicProgress).";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_051923FB
=
+ "Subscription: Failed to create consumer %s in config node, status
is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_6152986E
=
+ "Subscription: Failed to create consumer %s in config node,
exception is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_F31E2A0F
=
+ "Subscription: Failed to close consumer %s in config node, status is
%s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_11845DE5
=
+ "Subscription: Failed to close consumer %s in config node, exception
is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_9F933EBA
=
+ "Subscription: Failed to subscribe topics %s for consumer %s in
config node, status is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_0B6D1355
=
+ "Subscription: Failed to subscribe topics %s for consumer %s in
config node, exception is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_74DF0145
=
+ "Subscription: Failed to unsubscribe topics %s for consumer %s in
config node, status is %s.";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_A92043B5
=
+ "Subscription: Failed to unsubscribe topics %s for consumer %s in
config node, exception is %s.";
+
+ // ===================== 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
index a3949cd283e..d9439de9014 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
@@ -2535,4 +2535,87 @@ public final class DataNodePipeMessages {
"[%s] 订阅 entry 需要 %d 字节，超过当前队列可用上限 %d 字节（DataNode 预算 %d 字节）。请减小写入
batch/字段大小或增加订阅物化内存。WAL 进度未推进。";
public static final String
LOG_SUBSCRIPTION_FAILED_TO_REPLAY_DELIVERY_BATCH_ARG_FROM_WAL_RETAINING_UNCOMMITTED_STATE_0654EACD
=
"Subscription 无法从 WAL 重放交付批次 {}；保留未提交状态";
+
+ // ===================== SUBSCRIPTION RECEIVER RESPONSES
=====================
+
+ public static final String
+
MESSAGE_SUBSCRIPTION_MALFORMED_REQUEST_SLICE_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_D91FD262
=
+ "Subscription：请求分片格式错误。请从分片索引 0 开始重新发送完整请求。";
+ public static final String
+
MESSAGE_SUBSCRIPTION_INVALID_OR_OUT_OF_ORDER_REQUEST_SLICE_ORDERID_ARG_SLICEINDEX_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_BYTES_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_A08D6D8E
=
+ "Subscription：请求分片无效或顺序错误（orderId %d，sliceIndex %d，sliceCount
%d，originBodySize %d 字节）。请从分片索引 0 开始重新发送完整请求。";
+ public static final String
+
MESSAGE_SUBSCRIPTION_REQUEST_BODY_SIZE_ARG_BYTES_EXCEEDS_THE_RECEIVER_LIMIT_OF_ARG_BYTES_SPLIT_THE_OPERATION_INTO_SMALLER_REQUESTS_C9513027
=
+ "Subscription：请求体大小为 %d 字节，超过接收端上限 %d 字节。请将操作拆分为更小的请求。";
+
+ public static final String
MESSAGE_MISSING_CONSUMER_CONFIG_PLEASE_HANDSHAKE_FIRST_CE588329 =
+ "缺少 consumer 配置，请先完成 handshake。";
+ public static final String
MESSAGE_UNKNOWN_PIPESUBSCRIBEREQUESTTYPE_ARG_7EBB93F8 =
+ "未知的 PipeSubscribeRequestType：%s。";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HANDSHAKING_WITH_REQUEST_ARG_ARG_C1BC903B
=
+ "Subscription：处理 handshake 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HEARTBEAT_WITH_REQUEST_ARG_ARG_790B328F
=
+ "Subscription：处理 heartbeat 请求 %s 时发生意外错误：%s";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_FETCH_ENDPOINTS_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_267904CB
=
+ "Subscription：在 ConfigNode 中为 consumer %s 获取节点端点失败，异常：%s。";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SUBSCRIBING_WITH_REQUEST_ARG_ARG_084648A4
=
+ "Subscription：处理 subscribe 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_UNSUBSCRIBING_WITH_REQUEST_ARG_ARG_1C15D8EC
=
+ "Subscription：处理 unsubscribe 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_POLLING_WITH_REQUEST_ARG_ARG_DE2CF98B
=
+ "Subscription：处理 poll 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_COMMITTING_WITH_REQUEST_ARG_ARG_9AA9866F
=
+ "Subscription：处理 commit 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_CLOSING_WITH_REQUEST_ARG_ARG_1E3215D2
=
+ "Subscription：处理 close 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SEEKING_WITH_REQUEST_ARG_ARG_97D0AD1C
=
+ "Subscription：处理 seek 请求 %s 时发生意外错误：%s";
+ public static final String
+
MESSAGE_SUBSCRIPTION_UNSUPPORTED_SEEKTYPE_ARG_FOR_TOPIC_ARG_CONSENSUS_SUBSCRIPTION_ONLY_SUPPORTS_SEEKTOBEGINNING_SEEKTOEND_SEEK_TOPICPROGRESS_AND_SEEKAFTER_TOPICPROGRESS_5A3436C3
=
+ "Subscription：seekType %s 不支持用于 topic %s。Consensus subscription 仅支持
seekToBeginning、seekToEnd、seek(topicProgress) 和 seekAfter(topicProgress)。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_051923FB
=
+ "Subscription：在 ConfigNode 中创建 consumer %s 失败，状态为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_6152986E
=
+ "Subscription：在 ConfigNode 中创建 consumer %s 失败，异常为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_F31E2A0F
=
+ "Subscription：在 ConfigNode 中关闭 consumer %s 失败，状态为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_11845DE5
=
+ "Subscription：在 ConfigNode 中关闭 consumer %s 失败，异常为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_9F933EBA
=
+ "Subscription：在 ConfigNode 中订阅主题 %s 失败，consumer 为 %s，状态为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_0B6D1355
=
+ "Subscription：在 ConfigNode 中订阅主题 %s 失败，consumer 为 %s，异常为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_74DF0145
=
+ "Subscription：在 ConfigNode 中取消订阅主题 %s 失败，consumer 为 %s，状态为 %s。";
+ public static final String
+
EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_A92043B5
=
+ "Subscription：在 ConfigNode 中取消订阅主题 %s 失败，consumer 为 %s，异常为 %s。";
+
+ // ===================== 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 7dbf2db122b..3bad4b45398 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
@@ -164,9 +164,11 @@ public class IoTDBLegacyPipeReceiverAgent {
handleTsFilePipeData(tsFilePipeData, fileDir);
}
} catch (final IOException e) {
- LOGGER.error(DataNodePipeMessages.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
@@ -181,9 +183,12 @@ public class IoTDBLegacyPipeReceiverAgent {
DataNodePipeMessages.LOAD_PIPEDATA_WITH_SERIALIZE_NUMBER_SUCCESSFULLY,
pipeData.getSerialNumber());
} catch (final PipeException e) {
- LOGGER.error(DataNodePipeMessages.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, "");
@@ -273,8 +278,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
@@ -284,8 +295,8 @@ public class IoTDBLegacyPipeReceiverAgent {
}
// step3. append file
+ final int length = buff.remaining();
try (final RandomAccessFile randomAccessFile = new RandomAccessFile(file,
"rw")) {
- final int length = buff.remaining();
randomAccessFile.seek(startIndex);
final byte[] byteArray = new byte[length];
buff.get(byteArray);
@@ -294,8 +305,16 @@ public class IoTDBLegacyPipeReceiverAgent {
LOGGER.debug(
DataNodePipeMessages.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/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index 7e9b94b508e..cd53d9376ab 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -122,7 +122,7 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
new TPipeSubscribeResp(
RpcUtils.getStatus(
TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER,
- "Missing consumer config, please handshake first."),
+
DataNodePipeMessages.MESSAGE_MISSING_CONSUMER_CONFIG_PLEASE_HANDSHAKE_FIRST_CE588329),
PipeSubscribeResponseVersion.VERSION_1.getVersion(),
PipeSubscribeResponseType.ACK.getType());
@@ -141,12 +141,6 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
PipeSubscribeResponseVersion.VERSION_1.getVersion(),
PipeSubscribeResponseType.ACK.getType());
- private static final TPipeSubscribeResp SUBSCRIPTION_SLICE_ERROR_RESP =
- new TPipeSubscribeResp(
- new TSStatus(TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode()),
- PipeSubscribeResponseVersion.VERSION_1.getVersion(),
- PipeSubscribeResponseType.ACK.getType());
-
private final ThreadLocal<ConsumerConfig> consumerConfigThreadLocal = new
ThreadLocal<>();
private final ThreadLocal<PollTimer> pollTimerThreadLocal = new
ThreadLocal<>();
private volatile String authenticatedUsername;
@@ -155,8 +149,9 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
private volatile boolean consumerFenced;
private volatile long lastActivityTimeMs = System.currentTimeMillis();
private final AtomicLong inFlightRequestCount = new AtomicLong(0);
+ private final int maxReassembledRequestBodySize =
getMaxReassembledRequestBodySize();
private final PipeSubscribeSliceReqHandler sliceReqHandler =
- new PipeSubscribeSliceReqHandler(getMaxReassembledRequestBodySize());
+ new PipeSubscribeSliceReqHandler(maxReassembledRequestBodySize);
private long consumerStateVersion;
private static final String SQL_DIALECT_TABLE_VALUE = "table";
@@ -203,7 +198,9 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
final TSStatus status =
RpcUtils.getStatus(
TSStatusCode.SUBSCRIPTION_TYPE_ERROR,
- String.format("Unknown PipeSubscribeRequestType %s.", reqType));
+ String.format(
+
DataNodePipeMessages.MESSAGE_UNKNOWN_PIPESUBSCRIBEREQUESTTYPE_ARG_7EBB93F8,
+ reqType));
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_UNKNOWN_REQUEST_TYPE,
status);
return new TPipeSubscribeResp(
status,
@@ -302,10 +299,27 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
sliceReq = PipeSubscribeSliceReq.fromTPipeSubscribeReq(thriftReq);
} catch (final RuntimeException e) {
sliceReqHandler.clear();
- return SUBSCRIPTION_SLICE_ERROR_RESP;
+ return createSliceErrorResp(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_MALFORMED_REQUEST_SLICE_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_D91FD262);
}
if (!sliceReqHandler.receiveSlice(sliceReq)) {
- return SUBSCRIPTION_SLICE_ERROR_RESP;
+ if (sliceReq.getOriginBodySize() > maxReassembledRequestBodySize) {
+ return createSliceErrorResp(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_REQUEST_BODY_SIZE_ARG_BYTES_EXCEEDS_THE_RECEIVER_LIMIT_OF_ARG_BYTES_SPLIT_THE_OPERATION_INTO_SMALLER_REQUESTS_C9513027,
+ sliceReq.getOriginBodySize(),
+ maxReassembledRequestBodySize));
+ }
+ return createSliceErrorResp(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_INVALID_OR_OUT_OF_ORDER_REQUEST_SLICE_ORDERID_ARG_SLICEINDEX_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_BYTES_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_A08D6D8E,
+ sliceReq.getOrderId(),
+ sliceReq.getSliceIndex(),
+ sliceReq.getSliceCount(),
+ sliceReq.getOriginBodySize()));
}
return sliceReqHandler
.makeReqIfComplete()
@@ -313,6 +327,13 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
.orElse(SUBSCRIPTION_SLICE_ACK_RESP);
}
+ private static TPipeSubscribeResp createSliceErrorResp(final String message)
{
+ return new TPipeSubscribeResp(
+ RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_TYPE_ERROR, message),
+ PipeSubscribeResponseVersion.VERSION_1.getVersion(),
+ PipeSubscribeResponseType.ACK.getType());
+ }
+
private static int getMaxReassembledRequestBodySize() {
final long maxFrameSize =
IoTDBDescriptor.getInstance().getConfig().getThriftMaxFrameSize();
final long desiredSize =
@@ -379,8 +400,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_HANDSHAKING,
req, e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when handshaking
with request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HANDSHAKING_WITH_REQUEST_ARG_ARG_C1BC903B,
+ req,
+ e);
return PipeSubscribeHandshakeResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_HANDSHAKE_ERROR,
exceptionMessage),
-1,
@@ -450,8 +473,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_HEARTBEAT, req,
e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when heartbeat with
request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_HEARTBEAT_WITH_REQUEST_ARG_ARG_790B328F,
+ req,
+ e);
return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_HEARTBEAT_ERROR,
exceptionMessage));
}
@@ -534,8 +559,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
e);
final String exceptionMessage =
String.format(
- "Subscription: Failed to fetch endpoints for consumer %s in
config node, exception is %s.",
- consumerConfig, e);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_FETCH_ENDPOINTS_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_267904CB,
+ consumerConfig,
+ e);
throw new SubscriptionException(exceptionMessage);
}
@@ -570,8 +597,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_SUBSCRIBING,
req, e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when subscribing
with request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SUBSCRIBING_WITH_REQUEST_ARG_ARG_084648A4,
+ req,
+ e);
return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_SUBSCRIBE_ERROR,
exceptionMessage));
}
@@ -626,8 +655,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_UNSUBSCRIBING,
req, e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when unsubscribing
with request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_UNSUBSCRIBING_WITH_REQUEST_ARG_ARG_1C15D8EC,
+ req,
+ e);
return PipeSubscribeUnsubscribeResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_UNSUBSCRIBE_ERROR,
exceptionMessage));
}
@@ -670,8 +701,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_POLLING, req, e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when polling with
request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_POLLING_WITH_REQUEST_ARG_ARG_DE2CF98B,
+ req,
+ e);
return PipeSubscribePollResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_POLL_ERROR,
exceptionMessage),
Collections.emptyList());
@@ -956,8 +989,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_COMMITTING, req,
e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when committing
with request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_COMMITTING_WITH_REQUEST_ARG_ARG_9AA9866F,
+ req,
+ e);
return PipeSubscribeCommitResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_COMMIT_ERROR,
exceptionMessage));
}
@@ -1137,8 +1172,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_CLOSING, req, e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when closing with
request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_CLOSING_WITH_REQUEST_ARG_ARG_1E3215D2,
+ req,
+ e);
return PipeSubscribeCloseResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_CLOSE_ERROR,
exceptionMessage));
}
@@ -1172,8 +1209,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
e);
final String exceptionMessage =
String.format(
- "Subscription: something unexpected happened when seeking with
request %s: %s",
- req, e);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_SEEKING_WITH_REQUEST_ARG_ARG_97D0AD1C,
+ req,
+ e);
return PipeSubscribeSeekResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_SEEK_ERROR,
exceptionMessage));
}
@@ -1231,10 +1270,10 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
} else {
final String errorMessage =
String.format(
- "Subscription: unsupported seekType %s for topic %s. "
- + "Consensus subscription only supports seekToBeginning,
seekToEnd, "
- + "seek(topicProgress), and seekAfter(topicProgress).",
- seekType, topicName);
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_UNSUPPORTED_SEEKTYPE_ARG_FOR_TOPIC_ARG_CONSENSUS_SUBSCRIPTION_ONLY_SUPPORTS_SEEKTOBEGINNING_SEEKTOEND_SEEK_TOPICPROGRESS_AND_SEEKAFTER_TOPICPROGRESS_5A3436C3,
+ seekType,
+ topicName);
LOGGER.warn(errorMessage);
return PipeSubscribeSeekResp.toTPipeSubscribeResp(
RpcUtils.getStatus(TSStatusCode.SUBSCRIPTION_SEEK_ERROR,
errorMessage));
@@ -1327,16 +1366,20 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
consumerConfig);
final String exceptionMessage =
String.format(
- "Subscription: Failed to create consumer %s in config node,
status is %s.",
- consumerConfig, tsStatus);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_051923FB,
+ consumerConfig,
+ tsStatus);
throw new SubscriptionException(exceptionMessage);
}
} catch (final ClientManagerException | TException e) {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_CREATING_CONSUMER,
consumerConfig, e);
final String exceptionMessage =
String.format(
- "Subscription: Failed to create consumer %s in config node,
exception is %s.",
- consumerConfig, e);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_CREATE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_6152986E,
+ consumerConfig,
+ e);
throw new SubscriptionException(exceptionMessage);
}
}
@@ -1357,16 +1400,20 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
consumerConfig);
final String exceptionMessage =
String.format(
- "Subscription: Failed to close consumer %s in config node,
status is %s.",
- consumerConfig, tsStatus);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_F31E2A0F,
+ consumerConfig,
+ tsStatus);
throw new SubscriptionException(exceptionMessage);
}
} catch (final ClientManagerException | TException e) {
LOGGER.warn(DataNodeMiscMessages.SUBSCRIPTION_EXCEPTION_CLOSING_CONSUMER,
consumerConfig, e);
final String exceptionMessage =
String.format(
- "Subscription: Failed to close consumer %s in config node,
exception is %s.",
- consumerConfig, e);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_CLOSE_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_11845DE5,
+ consumerConfig,
+ e);
throw new SubscriptionException(exceptionMessage);
}
@@ -1394,8 +1441,11 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
consumerConfig);
final String exceptionMessage =
String.format(
- "Subscription: Failed to subscribe topics %s for consumer %s
in config node, status is %s.",
- topicNames, consumerConfig, tsStatus);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_9F933EBA,
+ topicNames,
+ consumerConfig,
+ tsStatus);
if (TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode() ==
tsStatus.getCode()) {
throw new SubscriptionPipeTimeoutException(exceptionMessage);
} else {
@@ -1411,8 +1461,11 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
e);
final String exceptionMessage =
String.format(
- "Subscription: Failed to subscribe topics %s for consumer %s in
config node, exception is %s.",
- topicNames, consumerConfig, e);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_SUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_0B6D1355,
+ topicNames,
+ consumerConfig,
+ e);
throw new SubscriptionException(exceptionMessage);
}
}
@@ -1438,8 +1491,11 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
consumerConfig);
final String exceptionMessage =
String.format(
- "Subscription: Failed to unsubscribe topics %s for consumer %s
in config node, status is %s.",
- topicNames, consumerConfig, tsStatus);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_STATUS_IS_ARG_74DF0145,
+ topicNames,
+ consumerConfig,
+ tsStatus);
if (TSStatusCode.SUBSCRIPTION_PIPE_TIMEOUT_ERROR.getStatusCode() ==
tsStatus.getCode()) {
throw new SubscriptionPipeTimeoutException(exceptionMessage);
} else {
@@ -1455,8 +1511,11 @@ public class SubscriptionReceiverV1 implements
SubscriptionReceiver {
e);
final String exceptionMessage =
String.format(
- "Subscription: Failed to unsubscribe topics %s for consumer %s
in config node, exception is %s.",
- topicNames, consumerConfig, e);
+ DataNodePipeMessages
+
.EXCEPTION_SUBSCRIPTION_FAILED_TO_UNSUBSCRIBE_TOPICS_ARG_FOR_CONSUMER_ARG_IN_CONFIG_NODE_EXCEPTION_IS_ARG_A92043B5,
+ topicNames,
+ consumerConfig,
+ e);
throw new SubscriptionException(exceptionMessage);
}
}
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 5ce4df74f7f..111350d7607 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,15 +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.commons.i18n.PipeMessages;
+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.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;
@@ -85,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,
+
String.format(PipeMessages.ILLEGAL_FILENAME_PATH_TRAVERSAL, traversal))));
Assert.assertFalse(Files.exists(syncDir.resolve("pwned.patch")));
}
@@ -102,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";
@@ -110,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,
+
String.format(PipeMessages.ILLEGAL_FILENAME_PATH_TRAVERSAL, traversal))));
}
private Path getFileDataDir() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
index a8c7dc7841f..40482fe1b20 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
@@ -19,17 +19,22 @@
package org.apache.iotdb.db.subscription.receiver;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext;
+import org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeCloseReq;
import
org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeHandshakeReq;
+import
org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeRequestType;
+import
org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeRequestVersion;
import
org.apache.iotdb.rpc.subscription.payload.request.PipeSubscribeSliceReqBuilder;
import
org.apache.iotdb.rpc.subscription.payload.request.SubscriptionHeartbeatReq;
import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeReq;
@@ -40,6 +45,7 @@ import org.junit.Test;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -219,9 +225,57 @@ public class SubscriptionReceiverV1Test {
public void testNeverHandshakenConsumerStillReturnsMissingConsumerStatus() {
final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
+ final TSStatus status =
receiver.handle(SubscriptionHeartbeatReq.toThriftReq()).getStatus();
+
Assert.assertEquals(
- TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER.getStatusCode(),
-
receiver.handle(SubscriptionHeartbeatReq.toThriftReq()).getStatus().getCode());
+ TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+
DataNodePipeMessages.MESSAGE_MISSING_CONSUMER_CONFIG_PLEASE_HANDSHAKE_FIRST_CE588329,
+ status.getMessage());
+ }
+
+ @Test
+ public void testUnknownRequestTypeReturnsLocalizedMessageTemplate() {
+ final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
+ final short requestType = Short.MAX_VALUE;
+
+ final TSStatus status =
+ receiver.handle(new
TPipeSubscribeReq().setType(requestType)).getStatus();
+
+ Assert.assertEquals(TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ String.format(
+
DataNodePipeMessages.MESSAGE_UNKNOWN_PIPESUBSCRIBEREQUESTTYPE_ARG_7EBB93F8,
+ requestType),
+ status.getMessage());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testCloseFailurePreservesRequestAndExceptionInResponse() throws
Exception {
+ final RuntimeException failure = new RuntimeException("expected cleanup
failure");
+ final SubscriptionReceiverV1 receiver =
+ new SubscriptionReceiverV1() {
+ @Override
+ void closeConsumer(final ConsumerConfig consumerConfig) {
+ throw failure;
+ }
+ };
+ final ThreadLocal<ConsumerConfig> consumerConfigThreadLocal =
+ (ThreadLocal<ConsumerConfig>) getField(receiver,
"consumerConfigThreadLocal");
+ consumerConfigThreadLocal.set(createConsumerConfig(1_000L));
+ final PipeSubscribeCloseReq req =
PipeSubscribeCloseReq.toTPipeSubscribeReq();
+
+ final TSStatus status = receiver.handle(req).getStatus();
+
+ Assert.assertEquals(TSStatusCode.SUBSCRIPTION_CLOSE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_SOMETHING_UNEXPECTED_HAPPENED_WHEN_CLOSING_WITH_REQUEST_ARG_ARG_1E3215D2,
+ req,
+ failure),
+ status.getMessage());
}
@Test
@@ -247,6 +301,83 @@ public class SubscriptionReceiverV1Test {
TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER.getStatusCode(),
lastResp.getStatus().getCode());
}
+ @Test
+ public void testMalformedSliceReturnsMessageAndAllowsCompleteRequestRetry()
throws Exception {
+ final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
+ final TPipeSubscribeReq heartbeatReq =
+ SubscriptionHeartbeatReq.toThriftReq(Collections.emptyList());
+ final int bodySizeLimit = 2;
+ final int sliceCount =
PipeSubscribeSliceReqBuilder.getSliceCount(heartbeatReq, bodySizeLimit);
+
+ receiver.handle(
+ PipeSubscribeSliceReqBuilder.buildSliceReq(heartbeatReq, 1, 0,
sliceCount, bodySizeLimit));
+ final TSStatus status =
+ receiver
+ .handle(
+ new TPipeSubscribeReq()
+
.setVersion(PipeSubscribeRequestVersion.VERSION_1.getVersion())
+ .setType(PipeSubscribeRequestType.SLICE.getType())
+ .setBody(ByteBuffer.allocate(1)))
+ .getStatus();
+
+ Assert.assertEquals(TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_MALFORMED_REQUEST_SLICE_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_D91FD262,
+ status.getMessage());
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
+ receiver
+ .handle(
+ PipeSubscribeSliceReqBuilder.buildSliceReq(
+ heartbeatReq, 1, 1, sliceCount, bodySizeLimit))
+ .getStatus()
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ receiver
+ .handle(
+ PipeSubscribeSliceReqBuilder.buildSliceReq(
+ heartbeatReq, 2, 0, sliceCount, bodySizeLimit))
+ .getStatus()
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER.getStatusCode(),
+ receiver
+ .handle(
+ PipeSubscribeSliceReqBuilder.buildSliceReq(
+ heartbeatReq, 2, 1, sliceCount, bodySizeLimit))
+ .getStatus()
+ .getCode());
+ }
+
+ @Test
+ public void testOversizedSliceReturnsRequestedSizeAndReceiverLimit() throws
Exception {
+ final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
+ final int maxBodySize = (int) getField(receiver,
"maxReassembledRequestBodySize");
+ final int requestedBodySize = maxBodySize + 1;
+ final TPipeSubscribeReq heartbeatReq =
+ SubscriptionHeartbeatReq.toThriftReq(Collections.emptyList());
+ final int bodySizeLimit = 2;
+ final int sliceCount =
PipeSubscribeSliceReqBuilder.getSliceCount(heartbeatReq, bodySizeLimit);
+ final TPipeSubscribeReq sliceReq =
+ PipeSubscribeSliceReqBuilder.buildSliceReq(heartbeatReq, 1, 0,
sliceCount, bodySizeLimit);
+ // Advertise an oversized body without allocating it. The size follows the
order ID, version,
+ // and type in the slice header.
+ sliceReq.body.putInt(Integer.BYTES + Byte.BYTES + Short.BYTES,
requestedBodySize);
+
+ final TSStatus status = receiver.handle(sliceReq).getStatus();
+
+ Assert.assertEquals(TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_REQUEST_BODY_SIZE_ARG_BYTES_EXCEEDS_THE_RECEIVER_LIMIT_OF_ARG_BYTES_SPLIT_THE_OPERATION_INTO_SMALLER_REQUESTS_C9513027,
+ requestedBodySize,
+ maxBodySize),
+ status.getMessage());
+ }
+
@Test
public void testNonSliceRequestClearsIncompleteSlices() throws Exception {
final SubscriptionReceiverV1 receiver = new SubscriptionReceiverV1();
@@ -266,14 +397,22 @@ public class SubscriptionReceiverV1Test {
Assert.assertEquals(
TSStatusCode.SUBSCRIPTION_MISSING_CONSUMER.getStatusCode(),
receiver.handle(SubscriptionHeartbeatReq.toThriftReq()).getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
+ final TSStatus status =
receiver
.handle(
PipeSubscribeSliceReqBuilder.buildSliceReq(
heartbeatReq, 1, 1, sliceCount, bodySizeLimit))
- .getStatus()
- .getCode());
+ .getStatus();
+ Assert.assertEquals(TSStatusCode.SUBSCRIPTION_TYPE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertEquals(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_SUBSCRIPTION_INVALID_OR_OUT_OF_ORDER_REQUEST_SLICE_ORDERID_ARG_SLICEINDEX_ARG_SLICECOUNT_ARG_ORIGINBODYSIZE_ARG_BYTES_RESEND_THE_COMPLETE_REQUEST_STARTING_FROM_SLICE_INDEX_0_A08D6D8E,
+ 1,
+ 1,
+ sliceCount,
+ heartbeatReq.body.remaining()),
+ status.getMessage());
}
@Test
