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 1fdcb123ac9 [To dev/1.3] Preserve LOAD failure details in logs and
responses (#18935) (#18940)
1fdcb123ac9 is described below
commit 1fdcb123ac9f045b014c52eabaa5c9d1707ad936
Author: Caideyipi <[email protected]>
AuthorDate: Tue Oct 27 11:41:11 2026 +0800
[To dev/1.3] Preserve LOAD failure details in logs and responses (#18935)
(#18940)
* [To dev/1.3] Preserve LOAD failure details in logs and responses (#18935)
* Fix LoadTsFile analyzer diagnostics test reflection
* Fix unused PowerMock dependency
* Fix PowerMock access to Java reflection internals
* Open Java stream package for PowerMock tests
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 23 ++++
.../iotdb/db/i18n/DataNodeQueryMessages.java | 23 ++++
.../plan/analyze/load/LoadTsFileAnalyzer.java | 55 +++++---
.../scheduler/load/LoadTsFileDispatcherImpl.java | 57 ++++++--
.../plan/scheduler/load/LoadTsFileScheduler.java | 41 ++++--
.../load/LoadTsFileAnalyzerDiagnosticsTest.java | 147 +++++++++++++++++++++
.../load/LoadTsFileDispatcherImplTest.java | 84 +++++++++++-
pom.xml | 2 +
8 files changed, 389 insertions(+), 43 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 1c4433d3b55..4d04dbaf1d2 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -23,6 +23,29 @@ public final class DataNodeQueryMessages {
private DataNodeQueryMessages() {}
+ public static final String
EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
+ "Failed to convert TsFile %s to tablets. Status: %s";
+ public static final String
EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =
+ "Failed to convert TsFile %s to tablets: %s";
+ public static final String
MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3 =
+ "Tablet conversion returned no status";
+ public static final String
+
EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5 =
+ "Failed to load TsFiles after tablet conversion. Details: %s";
+ public static final String
+ LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E =
+ "LOAD rejected: uuid={}, region={}, endpoint={}, status={}";
+ public static final String
LOG_LOAD_COMMAND_REJECTED_UUID_ARG_ENDPOINT_ARG_STATUS_ARG_E8D0031C =
+ "LOAD command rejected: uuid={}, endpoint={}, status={}";
+ public static final String
+ MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645
=
+ "Target node rejected LOAD without a status or a message";
+ public static final String
+
MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A =
+ "Failed to dispatch LOAD %s for region %s to node %s: %s";
+ public static final String
MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_ARG_53D2BF0B =
+ "Failed to dispatch LOAD command %s to node %s: %s";
+
public static final String RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT =
"The failed memory reservation exceeds available memory by the
equivalent of at least "
+ "%,d columns, estimated from the observed average column size. ";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 5b151ce3e35..e84e99433f5 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -23,6 +23,29 @@ public final class DataNodeQueryMessages {
private DataNodeQueryMessages() {}
+ public static final String
EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477 =
+ "将 TsFile %s 转换为 tablet 失败。状态：%s";
+ public static final String
EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD =
+ "将 TsFile %s 转换为 tablet 失败：%s";
+ public static final String
MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3 =
+ "tablet 转换未返回状态";
+ public static final String
+
EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5 =
+ "将 TsFile 转换为 tablet 后加载仍失败。详情：%s";
+ public static final String
+ LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E =
+ "LOAD 被拒绝：uuid={}，region={}，endpoint={}，status={}";
+ public static final String
+ MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645
=
+ "目标节点拒绝 LOAD，且未返回状态或消息";
+ public static final String
LOG_LOAD_COMMAND_REJECTED_UUID_ARG_ENDPOINT_ARG_STATUS_ARG_E8D0031C =
+ "LOAD command 被拒绝：uuid={}，endpoint={}，status={}";
+ public static final String
+
MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A =
+ "分发 LOAD %s（region %s）到节点 %s 失败：%s";
+ public static final String
MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_ARG_53D2BF0B =
+ "分发 LOAD command %s 到节点 %s 失败：%s";
+
public static final String RESULT_SET_COLUMN_MEMORY_SHORTAGE_EQUIVALENT =
"本次失败的内存申请超出可用内存，按已记录列的平均内存估算，至少超出相当于 %,d 列的容量。";
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
index 3e25e033626..4cdf80d5416 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzer.java
@@ -44,6 +44,7 @@ import
org.apache.iotdb.db.exception.load.LoadEmptyFileException;
import org.apache.iotdb.db.exception.load.LoadFileException;
import org.apache.iotdb.db.exception.load.LoadRuntimeOutOfMemoryException;
import org.apache.iotdb.db.exception.sql.SemanticException;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
import org.apache.iotdb.db.protocol.client.ConfigNodeInfo;
@@ -544,23 +545,43 @@ public class LoadTsFileAnalyzer implements AutoCloseable {
final LoadTsFileDataTypeConverter loadTsFileDataTypeConverter =
new LoadTsFileDataTypeConverter(context, isGeneratedByPipe);
- final TSStatus status =
- loadTsFileStatement.isConvertOnTypeMismatch()
- ?
loadTsFileDataTypeConverter.convertForTreeModel(loadTsFileStatement).orElse(null)
- : null;
-
- if (status == null) {
- LOGGER.warn(
- "Load: Failed to convert to tablets from statement {}. Status is
null.",
- loadTsFileStatement);
- analysis.setFailStatus(
- new
TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage()));
- } else if (!loadTsFileDataTypeConverter.isSuccessful(status)) {
- LOGGER.warn(
- "Load: Failed to convert to tablets from statement {}. Status: {}",
- loadTsFileStatement,
- status);
- analysis.setFailStatus(status);
+ final String filePaths =
+ loadTsFileStatement.getTsFiles().stream()
+ .map(File::getPath)
+ .collect(Collectors.joining(", "));
+ try {
+ final TSStatus status =
+ loadTsFileStatement.isConvertOnTypeMismatch()
+ ?
loadTsFileDataTypeConverter.convertForTreeModel(loadTsFileStatement).orElse(null)
+ : null;
+
+ if (status == null) {
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePaths,
+
DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3);
+ LOGGER.warn(message);
+
analysis.setFailStatus(RpcUtils.getStatus(TSStatusCode.LOAD_FILE_ERROR,
message));
+ } else if (!loadTsFileDataTypeConverter.isSuccessful(status)) {
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477,
+ filePaths,
+ status);
+ LOGGER.warn(message);
+ analysis.setFailStatus(status.deepCopy().setMessage(message));
+ }
+ } catch (final Exception conversionException) {
+ final String message =
+ String.format(
+
DataNodeQueryMessages.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePaths,
+ conversionException);
+ LOGGER.warn(message, conversionException);
+ analysis.setFailStatus(RpcUtils.getStatus(TSStatusCode.LOAD_FILE_ERROR,
message));
}
analysis.setFinishQueryAfterAnalyze(true);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
index 542d726186f..7e2a3ff0029 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImpl.java
@@ -35,6 +35,7 @@ import
org.apache.iotdb.commons.consensus.index.ProgressIndexType;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.load.LoadFileException;
import org.apache.iotdb.db.exception.mpp.FragmentInstanceDispatchException;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.queryengine.plan.planner.plan.SubPlan;
@@ -317,18 +318,39 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher, AutoCl
for (final TTsFilePieceReq loadTsFileReq : loadTsFileReqs) {
final TLoadResp loadResp = client.sendTsFilePieceNode(loadTsFileReq);
if (!loadResp.isAccepted()) {
- LOGGER.warn(loadResp.message);
- throw new FragmentInstanceDispatchException(loadResp.status);
+ final TSStatus failureStatus =
+ loadResp.isSetStatus()
+ ? loadResp.getStatus()
+ : RpcUtils.getStatus(
+ TSStatusCode.DISPATCH_ERROR,
+ loadResp.isSetMessage()
+ ? loadResp.getMessage()
+ : DataNodeQueryMessages
+
.MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645);
+ LOGGER.warn(
+ DataNodeQueryMessages
+
.LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E,
+ uuid,
+ consensusGroupId,
+ endPoint,
+ failureStatus);
+ throw new FragmentInstanceDispatchException(failureStatus);
}
}
}
+ } catch (final FragmentInstanceDispatchException e) {
+ throw e;
} catch (Exception e) {
adjustTimeoutIfNecessary(e);
final String exceptionMessage =
String.format(
- "failed to dispatch load command %s to node %s because of
exception: %s",
- uuid, endPoint, e);
+ DataNodeQueryMessages
+
.MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A,
+ uuid,
+ consensusGroupId,
+ endPoint,
+ e);
LOGGER.warn(exceptionMessage, e);
throw new FragmentInstanceDispatchException(
new TSStatus()
@@ -413,16 +435,35 @@ public class LoadTsFileDispatcherImpl implements
IFragInstanceDispatcher, AutoCl
final TLoadResp loadResp = client.sendLoadCommand(loadCommandReq);
if (!loadResp.isAccepted()) {
- LOGGER.warn(loadResp.message);
- throw new FragmentInstanceDispatchException(loadResp.status);
+ final TSStatus failureStatus =
+ loadResp.isSetStatus()
+ ? loadResp.getStatus()
+ : RpcUtils.getStatus(
+ TSStatusCode.DISPATCH_ERROR,
+ loadResp.isSetMessage()
+ ? loadResp.getMessage()
+ : DataNodeQueryMessages
+
.MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645);
+ LOGGER.warn(
+ DataNodeQueryMessages
+
.LOG_LOAD_COMMAND_REJECTED_UUID_ARG_ENDPOINT_ARG_STATUS_ARG_E8D0031C,
+ loadCommandReq.uuid,
+ endPoint,
+ failureStatus);
+ throw new FragmentInstanceDispatchException(failureStatus);
}
+ } catch (final FragmentInstanceDispatchException e) {
+ throw e;
} catch (Exception e) {
adjustTimeoutIfNecessary(e);
final String exceptionMessage =
String.format(
- "failed to dispatch load command %s to node %s because of
exception: %s",
- loadCommandReq, endPoint, e);
+ DataNodeQueryMessages
+
.MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_ARG_53D2BF0B,
+ loadCommandReq,
+ endPoint,
+ e);
LOGGER.warn(exceptionMessage, e);
throw new FragmentInstanceDispatchException(
new TSStatus()
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
index 9c6946be6ab..d1f1f3fb370 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileScheduler.java
@@ -43,6 +43,7 @@ import org.apache.iotdb.db.exception.load.LoadFileException;
import org.apache.iotdb.db.exception.load.LoadReadOnlyException;
import org.apache.iotdb.db.exception.load.RegionReplicaSetChangedException;
import org.apache.iotdb.db.exception.mpp.FragmentInstanceDispatchException;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
@@ -559,6 +560,7 @@ public class LoadTsFileScheduler implements IScheduler {
private void convertFailedTsFilesToTabletsAndRetry() {
final LoadTsFileDataTypeConverter loadTsFileDataTypeConverter =
new LoadTsFileDataTypeConverter(queryContext, isGeneratedByPipe);
+ final List<String> failureDetails = new ArrayList<>();
final Iterator<Integer> iterator = failedTsFileNodeIndexes.listIterator();
while (iterator.hasNext()) {
@@ -579,17 +581,30 @@ public class LoadTsFileScheduler implements IScheduler {
"Load: Successfully converted TsFile {} into tablets and
inserted.",
failedNode.getTsFileResource().getTsFilePath());
} else {
- LOGGER.warn(
- "Load: Failed to convert to tablets from TsFile {}. Status: {}",
- failedNode.getTsFileResource().getTsFilePath(),
- status);
+ final String message =
+ status == null
+ ? String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+
DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3)
+ : String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477,
+ filePath,
+ status);
+ failureDetails.add(message);
+ LOGGER.warn(message);
}
} catch (final Exception e) {
- LOGGER.warn(
- "Load: Failed to convert to tablets from TsFile {}. Exception: {}",
- failedNode.getTsFileResource().getTsFilePath(),
- e.getMessage(),
- e);
+ final String message =
+ String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD,
+ filePath,
+ e);
+ failureDetails.add(message);
+ LOGGER.warn(message, e);
}
}
@@ -600,10 +615,10 @@ public class LoadTsFileScheduler implements IScheduler {
stateMachine.transitionToFinished();
} else {
final String errorMsg =
- "Load: failed to load some TsFiles by converting them into tablets.
Failed TsFiles: "
- + failedTsFileNodeIndexes.stream()
- .map(i ->
tsFileNodeList.get(i).getTsFileResource().getTsFilePath())
- .collect(Collectors.joining(", "));
+ String.format(
+ DataNodeQueryMessages
+
.EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5,
+ String.join("; ", failureDetails));
LOGGER.warn(errorMsg);
stateMachine.transitionToFailed(new LoadFileException(errorMsg));
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java
new file mode 100644
index 00000000000..816b79bd013
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.queryengine.plan.analyze.load;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.db.exception.LoadAnalyzeException;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
+import org.apache.iotdb.db.queryengine.common.MPPQueryContext;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.plan.analyze.Analysis;
+import org.apache.iotdb.db.queryengine.plan.analyze.IPartitionFetcher;
+import org.apache.iotdb.db.queryengine.plan.analyze.schema.ISchemaFetcher;
+import org.apache.iotdb.db.queryengine.plan.statement.crud.LoadTsFileStatement;
+import
org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.File;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.Optional;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*",
"javax.management.*"})
+@PrepareForTest(LoadTsFileAnalyzer.class)
+public class LoadTsFileAnalyzerDiagnosticsTest {
+
+ @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ private LoadTsFileDataTypeConverter converter;
+
+ @Before
+ public void setUp() throws Exception {
+ converter = Mockito.mock(LoadTsFileDataTypeConverter.class);
+ when(converter.isSuccessful(any())).thenCallRealMethod();
+ PowerMockito.whenNew(LoadTsFileDataTypeConverter.class)
+ .withAnyArguments()
+ .thenReturn(converter);
+ }
+
+ @Test
+ public void testConversionExceptionReportsCurrentError() throws Exception {
+ when(converter.convertForTreeModel(any()))
+ .thenThrow(new IllegalStateException("tree conversion failed"));
+
+ final File file = temporaryFolder.newFile("conversion-error.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(status.getMessage().contains("tree conversion failed"));
+ Assert.assertFalse(status.getMessage().contains("original analysis
failure"));
+ }
+
+ @Test
+ public void testMissingConversionStatusHasExplicitReason() throws Exception {
+ when(converter.convertForTreeModel(any())).thenReturn(Optional.empty());
+
+ final File file = temporaryFolder.newFile("no-status.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(),
status.getCode());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(
+ status
+ .getMessage()
+
.contains(DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3));
+ }
+
+ @Test
+ public void testConversionStatusKeepsMetadataWithoutMutatingOriginal()
throws Exception {
+ final TSStatus original =
+ new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode())
+ .setMessage("tablet insertion failed")
+ .setSubStatus(
+ Collections.singletonList(
+ new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
+ .setMessage("missing WRITE_DATA privilege")))
+ .setRedirectNode(new TEndPoint("127.0.0.2", 10730))
+ .setNeedRetry(true);
+ final TSStatus snapshot = original.deepCopy();
+
when(converter.convertForTreeModel(any())).thenReturn(Optional.of(original));
+
+ final File file = temporaryFolder.newFile("denied.tsfile");
+ final TSStatus status = convertAfterAnalysisFailure(file);
+
+ Assert.assertNotSame(original, status);
+ Assert.assertEquals(snapshot, original);
+ Assert.assertEquals(original.getCode(), status.getCode());
+ Assert.assertEquals(original.getSubStatus(), status.getSubStatus());
+ Assert.assertEquals(original.getRedirectNode(), status.getRedirectNode());
+ Assert.assertEquals(original.isNeedRetry(), status.isNeedRetry());
+ Assert.assertTrue(status.getMessage().contains(file.getPath()));
+ Assert.assertTrue(status.getMessage().contains("missing WRITE_DATA
privilege"));
+ }
+
+ private TSStatus convertAfterAnalysisFailure(final File file) throws
Exception {
+ try (final LoadTsFileAnalyzer analyzer =
+ new LoadTsFileAnalyzer(
+
LoadTsFileStatement.createUnchecked(file.getPath()).setConvertOnTypeMismatch(true),
+ new MPPQueryContext(new QueryId("load_diagnostics")),
+ Mockito.mock(IPartitionFetcher.class),
+ Mockito.mock(ISchemaFetcher.class))) {
+ final Analysis analysis = new Analysis();
+ final Method conversionMethod =
+ LoadTsFileAnalyzer.class.getDeclaredMethod(
+ "executeTabletConversionOnException", Analysis.class,
LoadAnalyzeException.class);
+ conversionMethod.setAccessible(true);
+ conversionMethod.invoke(
+ analyzer, analysis, new LoadAnalyzeException("original analysis
failure"));
+ Assert.assertTrue(analysis.isFinishQueryAfterAnalyze());
+ return analysis.getFailStatus();
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
index b9082a54c75..b3b6653371c 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadTsFileDispatcherImplTest.java
@@ -24,23 +24,27 @@ import
org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.IClientManager;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.partition.StorageExecutor;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
import org.apache.iotdb.db.queryengine.plan.planner.plan.FragmentInstance;
import org.apache.iotdb.db.queryengine.plan.planner.plan.PlanFragment;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
+import
org.apache.iotdb.db.queryengine.plan.scheduler.FragInstanceDispatchResult;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
import org.apache.iotdb.mpp.rpc.thrift.TLoadResp;
import org.apache.iotdb.mpp.rpc.thrift.TTsFilePieceReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TElasticFramedTransport;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
@@ -246,15 +250,85 @@ public class LoadTsFileDispatcherImplTest {
new LoadTsFilePieceNode(new PlanNodeId("piece"), new
File("test.tsfile"));
try (LoadTsFileDispatcherImpl dispatcher = new
LoadTsFileDispatcherImpl(clientManager, false)) {
dispatcher.setUuid("test-uuid");
- Assert.assertFalse(
- dispatcher
- .dispatch(null,
Collections.singletonList(createFragmentInstance(pieceNode)))
- .get(10, TimeUnit.SECONDS)
- .isSuccessful());
+ final FragmentInstance instance = createFragmentInstance(pieceNode);
+ final FragInstanceDispatchResult result =
+ dispatcher.dispatch(null,
Collections.singletonList(instance)).get(10, TimeUnit.SECONDS);
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(),
result.getFailureStatus().getCode());
+
Assert.assertTrue(result.getFailureStatus().getMessage().contains("test-uuid"));
+ Assert.assertTrue(
+ result
+ .getFailureStatus()
+ .getMessage()
+
.contains(instance.getRegionReplicaSet().getRegionId().toString()));
+
Assert.assertTrue(result.getFailureStatus().getMessage().contains("127.0.0.1"));
}
Mockito.verify(client, Mockito.never()).sendTsFilePieceNode(Mockito.any());
}
+ @Test
+ public void testRemoteRejectionPreservesOriginalStatus() throws Exception {
+ final TSStatus status =
+ new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode())
+ .setMessage("tablet insertion rejected")
+ .setSubStatus(
+ Collections.singletonList(
+ new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode())
+ .setMessage("missing WRITE_DATA privilege")))
+ .setRedirectNode(new TEndPoint("127.0.0.2", 10730))
+ .setNeedRetry(true);
+
+ final FragInstanceDispatchResult result =
+ dispatchWithResponse(new TLoadResp(false).setStatus(status));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(status, result.getFailureStatus());
+ }
+
+ @Test
+ public void testRemoteRejectionWithoutStatusPreservesMessage() throws
Exception {
+ final FragInstanceDispatchResult result =
+ dispatchWithResponse(new TLoadResp(false).setMessage("receiver disk is
full"));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(),
result.getFailureStatus().getCode());
+ Assert.assertEquals("receiver disk is full",
result.getFailureStatus().getMessage());
+ }
+
+ @Test
+ public void testRemoteRejectionWithoutDetailsHasExplicitReason() throws
Exception {
+ final FragInstanceDispatchResult result = dispatchWithResponse(new
TLoadResp(false));
+
+ Assert.assertFalse(result.isSuccessful());
+ Assert.assertEquals(
+ TSStatusCode.DISPATCH_ERROR.getStatusCode(),
result.getFailureStatus().getCode());
+ Assert.assertEquals(
+ DataNodeQueryMessages
+
.MESSAGE_TARGET_NODE_REJECTED_LOAD_WITHOUT_A_STATUS_OR_A_MESSAGE_F4E4C645,
+ result.getFailureStatus().getMessage());
+ }
+
+ private static FragInstanceDispatchResult dispatchWithResponse(final
TLoadResp response)
+ throws Exception {
+ final IClientManager<TEndPoint, SyncDataNodeInternalServiceClient>
clientManager =
+ Mockito.mock(IClientManager.class);
+ final SyncDataNodeInternalServiceClient client =
+ Mockito.mock(SyncDataNodeInternalServiceClient.class);
+ Mockito.when(clientManager.borrowClient(Mockito.any())).thenReturn(client);
+ Mockito.when(client.getThriftMaxFrameSize()).thenReturn(4096);
+
Mockito.when(client.sendTsFilePieceNode(Mockito.any())).thenReturn(response);
+ final LoadTsFilePieceNode pieceNode =
+ new LoadTsFilePieceNode(new PlanNodeId("piece"), new
File("test.tsfile"));
+ try (LoadTsFileDispatcherImpl dispatcher = new
LoadTsFileDispatcherImpl(clientManager, false)) {
+ dispatcher.setUuid("test-uuid");
+ return dispatcher
+ .dispatch(null,
Collections.singletonList(createFragmentInstance(pieceNode)))
+ .get(10, TimeUnit.SECONDS);
+ }
+ }
+
private static void assertFitsBothTransports(
final TTsFilePieceReq request, final int senderMaxFrameSize, final int
receiverMaxFrameSize)
throws Exception {
diff --git a/pom.xml b/pom.xml
index 082fa90a23e..09ad02ff17f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -1518,7 +1518,9 @@
-->
<argLine><