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 1c4433d3b552e..4d04dbaf1d2e8 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 5b151ce3e351b..e84e99433f5f3 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 3e25e033626e2..4cdf80d5416c6 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.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 @@ private Analysis executeTabletConversionOnException( 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 542d726186f33..7e2a3ff002908 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.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 @@ private void dispatchRemote( 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 @@ private void dispatchRemote(TLoadCommandReq loadCommandReq, TEndPoint endPoint) 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 9c6946be6ab48..d1f1f3fb37058 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.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 @@ private boolean loadLocally(LoadSingleTsFileNode node) throws IoTDBException { private void convertFailedTsFilesToTabletsAndRetry() { final LoadTsFileDataTypeConverter loadTsFileDataTypeConverter = new LoadTsFileDataTypeConverter(queryContext, isGeneratedByPipe); + final List failureDetails = new ArrayList<>(); final Iterator iterator = failedTsFileNodeIndexes.listIterator(); while (iterator.hasNext()) { @@ -579,17 +581,30 @@ private void convertFailedTsFilesToTabletsAndRetry() { "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 @@ private void convertFailedTsFilesToTabletsAndRetry() { 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 0000000000000..816b79bd0134c --- /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 b9082a54c75e0..b3b6653371c7c 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.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 void testFrameLimitDiscoveryFailureDoesNotSendPiece() throws Exception { 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 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 082fa90a23e84..749e06319082e 100644 --- a/pom.xml +++ b/pom.xml @@ -1518,6 +1518,7 @@ -->