From cf8c0e28a2f49ada0e97148ad64ab78664739e37 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 15:53:31 +0800 Subject: [PATCH 1/2] Preserve LOAD failure details in logs and client responses --- .../iotdb/db/i18n/DataNodeQueryMessages.java | 17 ++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 17 ++ .../plan/analyze/load/LoadTsFileAnalyzer.java | 56 +++-- .../scheduler/load/LoadFallbackHandler.java | 56 +++-- .../load/LoadTsFileDispatcherImpl.java | 25 ++- .../LoadTsFileAnalyzerDiagnosticsTest.java | 206 ++++++++++++++++++ .../load/LoadFallbackHandlerTest.java | 154 +++++++++++++ .../load/LoadTsFileDispatcherImplTest.java | 84 ++++++- 8 files changed, 561 insertions(+), 54 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java 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 c89034e8831df..d6fcfc64eb196 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 @@ -20,6 +20,23 @@ package org.apache.iotdb.db.i18n; public final class 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_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A = + "Failed to read TsFile model information before tablet conversion for statement %s: %s"; + 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 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_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A = "Schema region is unavailable. Please retry later."; 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 a3307bd07cb42..e06c0ded22ce4 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 @@ -20,6 +20,23 @@ package org.apache.iotdb.db.i18n; public final class 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_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A = + "执行语句 %s 时，在转换为 tablet 前读取 TsFile 模型信息失败：%s"; + 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 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_SCHEMA_REGION_IS_UNAVAILABLE_PLEASE_RETRY_LATER_D642510A = "SchemaRegion 不可用，请稍后重试。"; 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 52ab6f5ebd204..268ea2a222e9e 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 @@ -804,13 +804,15 @@ private void executeTabletConversionOnException( try { getFileModelInfoBeforeTabletConversion(); } catch (Exception e1) { - LOGGER.warn( - DataNodeQueryMessages - .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_BECAUSE_FAILED_TO_READ_MODEL_INFO, - isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement, - e1.getMessage()); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_READ_TSFILE_MODEL_INFORMATION_BEFORE_TABLET_CONVERSION_FOR_STATEMENT_ARG_ARG_F05C852A, + isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement, + e1); + LOGGER.warn(message, e1); analysis.setFailStatus( - new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage())); + new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message)); analysis.setFinishQueryAfterAnalyze(true); setRealStatement(analysis); return; @@ -822,6 +824,7 @@ private void executeTabletConversionOnException( for (int i = 0; i < tsFiles.size(); i++) { final long startTime = System.nanoTime(); + final String filePath = tsFiles.get(i).getPath(); try { final TSStatus status = isTableModelTsFile.get(i) @@ -846,30 +849,37 @@ private void executeTabletConversionOnException( .orElse(null); if (status == null) { - LOGGER.warn( - DataNodeQueryMessages - .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_STATUS_IS_NULL, - isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD, + filePath, + DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3); + LOGGER.warn(message); analysis.setFailStatus( - new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()) - .setMessage(e.getMessage())); + new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message)); break; } else if (!loadTsFileDataTypeConverter.isSuccessful(status)) { - LOGGER.warn( - DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_STATUS_ARG, - isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement, - status); - analysis.setFailStatus(status); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_STATUS_ARG_E7C9F477, + filePath, + status); + LOGGER.warn(message); + analysis.setFailStatus(status.deepCopy().setMessage(message)); break; } } catch (final Exception e2) { - LOGGER.warn( - DataNodeQueryMessages - .LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_STATEMENT_ARG_BECAUSE_EXCEPTION_ARG, - isTableModelStatement ? loadTsFileTableStatement : loadTsFileTreeStatement, - e2.getMessage()); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_CONVERT_TSFILE_ARG_TO_TABLETS_ARG_96358FDD, + filePath, + e2); + LOGGER.warn(message, e2); analysis.setFailStatus( - new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(e.getMessage())); + new TSStatus(TSStatusCode.LOAD_FILE_ERROR.getStatusCode()).setMessage(message)); break; } finally { LOAD_TSFILE_COST_METRICS_SET.recordPhaseTimeCost( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java index 448b28a6111e0..27839d168b4ed 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandler.java @@ -34,6 +34,7 @@ import org.slf4j.LoggerFactory; import java.io.FileNotFoundException; +import java.util.ArrayList; import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -113,6 +114,7 @@ public void convertFailedTsFilesToTablets() { private void convertAndRetry() { final LoadTsFileDataTypeConverter converter = new LoadTsFileDataTypeConverter(queryContext, isGeneratedByPipe); + final List failureDetails = new ArrayList<>(); final Iterator iterator = failedTsFileNodeIndexes.iterator(); while (iterator.hasNext()) { @@ -141,21 +143,34 @@ private void convertAndRetry() { .LOAD_SUCCESSFULLY_CONVERTED_TSFILE_ARG_INTO_TABLETS_AND_INSERTED, filePath); } else { - LOGGER.warn( - DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_TSFILE_ARG_STATUS_ARG, - filePath, - 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( - DataNodeQueryMessages.LOAD_FAILED_TO_CONVERT_TO_TABLETS_FROM_TSFILE_ARG_EXCEPTION_ARG, - filePath, - 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); } } - resolveFinalState(); + resolveFinalState(failureDetails); } private TSStatus executeTableModelConversion( @@ -185,25 +200,20 @@ private TSStatus executeTreeModelConversion( return converter.convertForTreeModel(statement).orElse(null); } - private void resolveFinalState() { + private void resolveFinalState(final List failureDetails) { if (failedTsFileNodeIndexes.isEmpty()) { LOGGER.info(DataNodeQueryMessages.LOAD_ALL_FAILED_TSFILES_ARE_CONVERTED_TO_TABLETS); stateMachine.transitionToFinished(); return; } - final String remainingFailedFiles = getFailedFilePathsString(); - LOGGER.warn( - DataNodeQueryMessages - .LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3, - remainingFailedFiles); - - stateMachine.transitionToFailed( - new LoadFileException( - String.format( - DataNodeQueryMessages - .LOG_LOAD_FAILED_TO_LOAD_SOME_TSFILES_BY_CONVERTING_THEM_INTO_TABLETS_FAILED_TSFILES_ARG_7D9DB9C3, - remainingFailedFiles))); + final String message = + String.format( + DataNodeQueryMessages + .EXCEPTION_FAILED_TO_LOAD_TSFILES_AFTER_TABLET_CONVERSION_DETAILS_ARG_53647EC5, + String.join("; ", failureDetails)); + LOGGER.warn(message); + stateMachine.transitionToFailed(new LoadFileException(message)); } // ------------------------------------------------------------------------- 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 ee1a86d10447b..63a567e0d2fca 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 @@ -260,19 +260,37 @@ private void dispatchRemote( for (final TTsFilePieceReq request : requests) { final TLoadResp loadResp = client.sendTsFilePieceNode(request); if (!loadResp.isAccepted()) { + 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); final String errorCode = loadResp.isSetStatus() ? String.valueOf(loadResp.getStatus().getCode()) : UserDataTransferErrorCode.REMOTE_REJECTED.name(); recordTransferAttempt(endPoint, false, errorCode, null); transferAttemptAudited = true; - LOGGER.warn(loadResp.getMessage()); - throw new FragmentInstanceDispatchException(loadResp.getStatus()); + LOGGER.warn( + DataNodeQueryMessages + .LOG_LOAD_REJECTED_UUID_ARG_REGION_ARG_ENDPOINT_ARG_STATUS_ARG_8ED0679E, + uuid, + consensusGroupId, + endPoint, + failureStatus); + throw new FragmentInstanceDispatchException(failureStatus); } } recordTransferAttempt(endPoint, true, null, null); transferAttemptAudited = true; } + } catch (final FragmentInstanceDispatchException e) { + // A receiver rejection is already audited and carries its original status and retry hints. + throw e; } catch (final Exception e) { if (!transferAttemptAudited) { recordTransferAttempt(endPoint, false, null, e); @@ -282,8 +300,9 @@ private void dispatchRemote( final String message = String.format( DataNodeQueryMessages - .MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D, + .MESSAGE_FAILED_TO_DISPATCH_LOAD_ARG_FOR_REGION_ARG_TO_NODE_ARG_ARG_57EFEC3A, uuid, + consensusGroupId, endPoint, e); LOGGER.warn(message, e); 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..65eb3d2771084 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/analyze/load/LoadTsFileAnalyzerDiagnosticsTest.java @@ -0,0 +1,206 @@ +/* + * 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.conf.IoTDBDescriptor; +import org.apache.iotdb.db.exception.load.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.IAnalysis; +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.apache.tsfile.file.metadata.TableSchema; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.junit.After; +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.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 org.powermock.reflect.Whitebox; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.Optional; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +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; + private int originalDataNodeId; + + @Before + public void setUp() throws Exception { + originalDataNodeId = IoTDBDescriptor.getInstance().getConfig().getDataNodeId(); + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(0); + converter = mock(LoadTsFileDataTypeConverter.class); + when(converter.isSuccessful(any())).thenCallRealMethod(); + PowerMockito.whenNew(LoadTsFileDataTypeConverter.class) + .withAnyArguments() + .thenReturn(converter); + } + + @After + public void tearDown() { + IoTDBDescriptor.getInstance().getConfig().setDataNodeId(originalDataNodeId); + } + + @Test + public void testModelReadFailureReportsCurrentError() throws Exception { + final File file = temporaryFolder.newFile("unreadable.tsfile"); + PowerMockito.whenNew(TsFileSequenceReader.class) + .withArguments(file.getAbsolutePath(), true) + .thenThrow(new IOException("cannot read TsFile footer")); + + 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("cannot read TsFile footer")); + Assert.assertFalse(status.getMessage().contains("original analysis failure")); + } + + @Test + public void testConversionExceptionReportsCurrentErrorForBothModels() throws Exception { + when(converter.convertForTreeModel(any())) + .thenThrow(new IllegalStateException("tree conversion failed")); + when(converter.convertForTableModel(any())) + .thenThrow(new IllegalStateException("table conversion failed")); + + for (final boolean tableModel : new boolean[] {false, true}) { + mockModelReader(tableModel); + final File file = temporaryFolder.newFile(tableModel ? "table.tsfile" : "tree.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(tableModel ? "table conversion failed" : "tree conversion failed")); + Assert.assertFalse(status.getMessage().contains("original analysis failure")); + } + } + + @Test + public void testConversionExceptionWithoutMessageStillHasDetails() throws Exception { + mockModelReader(false); + when(converter.convertForTreeModel(any())).thenThrow(new IllegalStateException()); + + final File file = temporaryFolder.newFile("no-message.tsfile"); + final TSStatus status = convertAfterAnalysisFailure(file); + + Assert.assertTrue(status.getMessage().contains(file.getPath())); + Assert.assertTrue(status.getMessage().contains(IllegalStateException.class.getName())); + } + + @Test + public void testMissingConversionStatusHasExplicitReason() throws Exception { + mockModelReader(false); + 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)); + Assert.assertFalse(status.getMessage().contains("original analysis failure")); + } + + @Test + public void testConversionStatusKeepsCodeAndMetadataWithoutMutatingOriginal() throws Exception { + mockModelReader(false); + 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 void mockModelReader(final boolean tableModel) throws Exception { + final TsFileSequenceReader reader = mock(TsFileSequenceReader.class); + when(reader.getTableSchemaMap()) + .thenReturn( + tableModel + ? Collections.singletonMap("table1", mock(TableSchema.class)) + : Collections.emptyMap()); + PowerMockito.whenNew(TsFileSequenceReader.class).withAnyArguments().thenReturn(reader); + } + + private TSStatus convertAfterAnalysisFailure(final File file) throws Exception { + try (final LoadTsFileAnalyzer analyzer = + new LoadTsFileAnalyzer( + LoadTsFileStatement.createUnchecked(file.getPath()).setConvertOnTypeMismatch(true), + false, + new MPPQueryContext(new QueryId("load_diagnostics")))) { + final Analysis analysis = new Analysis(); + Whitebox.invokeMethod( + analyzer, + "executeTabletConversionOnException", + (IAnalysis) 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/LoadFallbackHandlerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java new file mode 100644 index 0000000000000..b765405277a4f --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/scheduler/load/LoadFallbackHandlerTest.java @@ -0,0 +1,154 @@ +/* + * 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.scheduler.load; + +import org.apache.iotdb.common.rpc.thrift.TSStatus; +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.execution.QueryState; +import org.apache.iotdb.db.queryengine.execution.QueryStateMachine; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadSingleTsFileNode; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.load.converter.LoadTsFileDataTypeConverter; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import com.google.common.util.concurrent.MoreExecutors; +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.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.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Optional; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +@RunWith(PowerMockRunner.class) +@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*"}) +@PrepareForTest(LoadFallbackHandler.class) +public class LoadFallbackHandlerTest { + + @Rule public TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private LoadTsFileDataTypeConverter converter; + + @Before + public void setUp() throws Exception { + // Progress-log formatting is unrelated to conversion results. Avoid instrumenting JDK Stream + // internals through this helper when running PowerMock on Java 17. + PowerMockito.suppress( + PowerMockito.method(LoadFallbackHandler.class, "getFailedFilePathsString")); + converter = mock(LoadTsFileDataTypeConverter.class); + when(converter.isSuccessful(any())).thenCallRealMethod(); + PowerMockito.whenNew(LoadTsFileDataTypeConverter.class) + .withAnyArguments() + .thenReturn(converter); + } + + @Test + public void testPartialFailureReportsEachFilesReason() throws Exception { + final List nodes = + Arrays.asList( + newNode("success.tsfile", false), + newNode("denied.tsfile", true), + newNode("exception.tsfile", false), + newNode("no-status.tsfile", false)); + final TSStatus denied = + new TSStatus(TSStatusCode.MULTIPLE_ERROR.getStatusCode()) + .setSubStatus( + Collections.singletonList( + new TSStatus(TSStatusCode.NO_PERMISSION.getStatusCode()) + .setMessage("missing WRITE_DATA privilege"))); + when(converter.convertForTableModel(any())).thenReturn(Optional.of(denied)); + when(converter.convertForTreeModel(any())) + .thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS)) + .thenThrow(new IllegalStateException()) + .thenReturn(Optional.empty()); + final List failedIndexes = new ArrayList<>(Arrays.asList(0, 1, 2, 3)); + final QueryStateMachine stateMachine = newStateMachine(); + + new LoadFallbackHandler(mock(MPPQueryContext.class), false, nodes, failedIndexes, stateMachine) + .convertFailedTsFilesToTablets(); + + Assert.assertEquals(QueryState.FAILED, stateMachine.getState()); + Assert.assertEquals(Arrays.asList(1, 2, 3), failedIndexes); + final TSStatus failure = stateMachine.getFailureStatus(); + Assert.assertEquals(TSStatusCode.LOAD_FILE_ERROR.getStatusCode(), failure.getCode()); + final String message = failure.getMessage(); + Assert.assertFalse(message.contains("success.tsfile")); + for (int i = 1; i < nodes.size(); i++) { + Assert.assertTrue(message.contains(nodes.get(i).getTsFileResource().getTsFilePath())); + } + Assert.assertTrue(message.contains("missing WRITE_DATA privilege")); + Assert.assertTrue(message.contains(String.valueOf(TSStatusCode.NO_PERMISSION.getStatusCode()))); + Assert.assertTrue(message.contains(IllegalStateException.class.getName())); + Assert.assertTrue( + message.contains( + DataNodeQueryMessages.MESSAGE_TABLET_CONVERSION_RETURNED_NO_STATUS_88BAB5D3)); + } + + @Test + public void testSuccessfulConversionsStillFinishLoad() throws Exception { + final List nodes = + Arrays.asList(newNode("tree.tsfile", false), newNode("table.tsfile", true)); + when(converter.convertForTreeModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS)); + when(converter.convertForTableModel(any())).thenReturn(Optional.of(RpcUtils.SUCCESS_STATUS)); + final List failedIndexes = new ArrayList<>(Arrays.asList(0, 1)); + final QueryStateMachine stateMachine = newStateMachine(); + + new LoadFallbackHandler(mock(MPPQueryContext.class), false, nodes, failedIndexes, stateMachine) + .convertFailedTsFilesToTablets(); + + Assert.assertEquals(QueryState.FINISHED, stateMachine.getState()); + Assert.assertTrue(failedIndexes.isEmpty()); + } + + private LoadSingleTsFileNode newNode(final String name, final boolean tableModel) + throws Exception { + final File file = temporaryFolder.newFile(name); + final LoadSingleTsFileNode node = mock(LoadSingleTsFileNode.class); + when(node.getTsFileResource()).thenReturn(new TsFileResource(file)); + when(node.isTableModel()).thenReturn(tableModel); + when(node.getDatabase()).thenReturn(tableModel ? "test" : null); + return node; + } + + private QueryStateMachine newStateMachine() { + final QueryStateMachine stateMachine = + new QueryStateMachine( + new QueryId("load_fallback"), MoreExecutors.newDirectExecutorService()); + stateMachine.transitionToRunning(); + return stateMachine; + } +} 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 18c09cf80e26d..e553391307db6 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,6 +24,7 @@ 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; @@ -31,16 +32,19 @@ import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; 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.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 { From 41adfeff7bcb8f48c454e2dbfced863ff6771474 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:06:44 +0800 Subject: [PATCH 2/2] Declare the PowerMock reflection test dependency --- iotdb-core/datanode/pom.xml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/iotdb-core/datanode/pom.xml b/iotdb-core/datanode/pom.xml index 48bc0490611dd..16368a3a6ae4e 100644 --- a/iotdb-core/datanode/pom.xml +++ b/iotdb-core/datanode/pom.xml @@ -289,6 +289,11 @@ powermock-api-mockito2 test + + org.powermock + powermock-reflect + test + org.awaitility awaitility