From 35f81a5b1569ae997c73099499dd8cd614e6c9fe Mon Sep 17 00:00:00 2001 From: "yangshixin.2026" Date: Mon, 21 Sep 2026 16:23:59 +0800 Subject: [PATCH] [fix](fe) Preserve Broker Load diagnostics ### What problem does this PR solve? Issue Number: None Related PR: #67002 Problem Summary: A Broker Load fragment may report DATA_QUALITY_ERROR together with its error URL and first error message. Coordinator handles the failure and releases the completion latch before storing those diagnostics. The loading task can then snapshot empty values into the final job state. SHOW LOAD consequently loses the URL or FirstErrorMsg even though the backend supplied them. Publish both fields before cancellation wakes the loading task in the legacy Coordinator and Nereids LoadProcessor. Use the pre-status hook for Nereids diagnostics while retaining final report aggregation in its existing position. The related PR addresses the URL race; this change also preserves FirstErrorMsg and tests the production cancellation path. The unit tests inspect diagnostics at the cancellation/latch boundary, so they exercise the race deterministically without sleeps. They also cover normal completion and reports that omit diagnostic fields. Scope and compatibility: - Change diagnostic publication order only; retain transaction, counter, retry, and persistence behavior. - Add no SQL syntax, configuration, RPC field, or storage format. - This fixes diagnostics carried by the report that triggers failure; it does not add collection of reports arriving after task teardown. ### Release note Fix missing error URLs and first error messages in SHOW LOAD when a Broker Load data quality error report includes those diagnostics. ### Check List (For Author) - Test - [ ] Regression test - [x] Unit Test - [ ] Manual test - [ ] No need to test or manual test - Behavior changed: - [ ] No. - [x] Yes. Preserve reported diagnostics when load tasks wake on failure. - Does this need documentation? - [x] No. Restore existing SHOW LOAD diagnostic fields without introducing a new user-facing interface. - [ ] Yes. ### Validation Local validation passed with JDK 17, Maven 3.9.9, and Thrift 0.24.0: - CoordinatorLoadDiagnosticsTest: 2 tests - LoadProcessorTest: 3 tests - AbstractJobProcessorTest: 2 tests - QeProcessorImplReportAckTest: 7 tests Total: 14 tests, 0 failures, 0 errors, and 0 skipped; 5 tests are new. FE Checkstyle reported 0 violations. git diff --check passed. No cluster regression test was run. ```bash bash ./run-fe-ut.sh --run \ 'org.apache.doris.qe.CoordinatorLoadDiagnosticsTest,'\ 'org.apache.doris.qe.runtime.LoadProcessorTest,'\ 'org.apache.doris.qe.AbstractJobProcessorTest,'\ 'org.apache.doris.qe.QeProcessorImplReportAckTest' ``` --- .../apache/doris/qe/AbstractJobProcessor.java | 4 + .../java/org/apache/doris/qe/Coordinator.java | 17 +- .../doris/qe/runtime/LoadProcessor.java | 20 +- .../qe/CoordinatorLoadDiagnosticsTest.java | 116 ++++++++++++ .../doris/qe/runtime/LoadProcessorTest.java | 171 ++++++++++++++++++ 5 files changed, 314 insertions(+), 14 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java index 647391dffbc422..235ba022263062 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java @@ -61,6 +61,9 @@ public AbstractJobProcessor(CoordinatorContext coordinatorContext) { protected abstract void doProcessReportExecStatus( TReportExecStatusParams params, SingleFragmentPipelineTask fragmentTask); + // Publish diagnostics that must be visible before an error status releases waiters. + protected void publishReportDiagnosticsBeforeStatus(TReportExecStatusParams params) {} + @Override public final void setPipelineExecutionTask(PipelineExecutionTask pipelineExecutionTask) { Preconditions.checkArgument(pipelineExecutionTask != null, "sqlPipelineTask can not be null"); @@ -134,6 +137,7 @@ public final boolean updateFragmentExecStatus(TReportExecStatusParams params) { DebugUtil.printId(queryId), params.getFragmentId(), DebugUtil.printId(params.getFragmentInstanceId()), params.getBackendId(), status.toString()); + publishReportDiagnosticsBeforeStatus(params); coordinatorContext.updateStatusIfOk(status); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index b3005119418039..5b3df949162f8e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -2788,6 +2788,15 @@ public boolean updateFragmentExecStatus(TReportExecStatusParams params) { boolean accepted = false; try { Status status = new Status(params.status); + // Publish load diagnostics before an error status can release the completion latch. + if (params.isSetTrackingUrl()) { + LOG.info("query_id={} tracking_url: {}", DebugUtil.printId(queryId), params.getTrackingUrl()); + trackingUrl = params.getTrackingUrl(); + } + if (params.isSetFirstErrorMsg()) { + LOG.info("query_id={} first_error_msg: {}", DebugUtil.printId(queryId), params.getFirstErrorMsg()); + firstErrorMsg = params.getFirstErrorMsg(); + } // for now, abort the query if we see any error except if the error is cancelled // and returned_all_results_ is true. // (UpdateStatus() initiates cancellation, if it hasn't already been initiated) @@ -2814,14 +2823,6 @@ public boolean updateFragmentExecStatus(TReportExecStatusParams params) { if (params.isSetLoadCounters() && loadCounters != null) { updateLoadCounters(params.getLoadCounters()); } - if (params.isSetTrackingUrl()) { - LOG.info("query_id={} tracking_url: {}", DebugUtil.printId(queryId), params.getTrackingUrl()); - trackingUrl = params.getTrackingUrl(); - } - if (params.isSetFirstErrorMsg()) { - LOG.info("query_id={} first_error_msg: {}", DebugUtil.printId(queryId), params.getFirstErrorMsg()); - firstErrorMsg = params.getFirstErrorMsg(); - } // Keep this report's identity local so another report cannot redirect its commit data. long reportTxnId = params.isSetTxnId() ? params.getTxnId() : txnId; if (params.isSetTxnId()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java index d4878ff99a3d6c..01f590ce431a42 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java @@ -159,6 +159,19 @@ public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return latch.get().await(timeout, unit); } + @Override + protected void publishReportDiagnosticsBeforeStatus(TReportExecStatusParams params) { + updateLoadDiagnostics(params); + } + + private void updateLoadDiagnostics(TReportExecStatusParams params) { + if (params.isSetTrackingUrl()) { + loadContext.updateTrackingUrl(params.getTrackingUrl()); + } + if (params.isSetFirstErrorMsg()) { + loadContext.updateFirstErrorMsg(params.getFirstErrorMsg()); + } + } @Override protected void doProcessReportExecStatus(TReportExecStatusParams params, SingleFragmentPipelineTask fragmentTask) { @@ -224,12 +237,7 @@ private void acceptFinalReport(TReportExecStatusParams params) { if (params.isSetLoadCounters()) { loadContext.updateLoadCounters(params.getLoadCounters()); } - if (params.isSetTrackingUrl()) { - loadContext.updateTrackingUrl(params.getTrackingUrl()); - } - if (params.isSetFirstErrorMsg()) { - loadContext.updateFirstErrorMsg(params.getFirstErrorMsg()); - } + updateLoadDiagnostics(params); if (params.isSetTxnId()) { loadContext.updateTransactionId(params.getTxnId()); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java new file mode 100644 index 00000000000000..bbf924f506f5b9 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java @@ -0,0 +1,116 @@ +// 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.doris.qe; + +import org.apache.doris.analysis.DescriptorTable; +import org.apache.doris.common.MarkedCountDownLatch; +import org.apache.doris.common.Pair; +import org.apache.doris.common.Status; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanFragmentId; +import org.apache.doris.thrift.TReportExecStatusParams; +import org.apache.doris.thrift.TStatus; +import org.apache.doris.thrift.TStatusCode; +import org.apache.doris.thrift.TUniqueId; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +class CoordinatorLoadDiagnosticsTest { + private static final int FRAGMENT_ID = 7; + private static final long BACKEND_ID = 9; + private static final String TRACKING_URL = "http://127.0.0.1/error-log"; + private static final String FIRST_ERROR_MSG = "invalid integer. Src line: invalid"; + + @Test + void publishesDiagnosticsBeforeFailureReleasesWaiters() throws Exception { + AtomicBoolean cancelled = new AtomicBoolean(); + Coordinator coordinator = new Coordinator(-1L, new TUniqueId(12345, 6), new DescriptorTable(), + Collections.singletonList(fragment()), Collections.emptyList(), "UTC", false, false) { + @Override + protected void cancelInternal(Status cancelReason) { + super.cancelInternal(cancelReason); + // Inspect the result at the exact point where a load task can return from join(), + // before the report handler resumes. No scheduling delay is needed to expose the race. + Assertions.assertTrue(join(1)); + Assertions.assertEquals(TStatusCode.DATA_QUALITY_ERROR, getExecStatus().getErrorCode()); + Assertions.assertEquals(TRACKING_URL, getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR_MSG, getFirstErrorMsg()); + cancelled.set(true); + } + }; + prepareReport(coordinator); + + Assertions.assertTrue(coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR))); + + Assertions.assertTrue(cancelled.get()); + } + + @Test + void retainsDiagnosticsOnNormalCompletion() throws Exception { + Coordinator coordinator = new Coordinator(-1L, new TUniqueId(12345, 7), new DescriptorTable(), + Collections.singletonList(fragment()), Collections.emptyList(), "UTC", false, false); + prepareReport(coordinator); + + Assertions.assertTrue(coordinator.updateFragmentExecStatus(report(TStatusCode.OK))); + + Assertions.assertTrue(coordinator.join(1)); + Assertions.assertTrue(coordinator.getExecStatus().ok()); + Assertions.assertEquals(TRACKING_URL, coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR_MSG, coordinator.getFirstErrorMsg()); + } + + private static PlanFragment fragment() { + PlanFragment fragment = Mockito.mock(PlanFragment.class); + Mockito.when(fragment.getFragmentId()).thenReturn(new PlanFragmentId(FRAGMENT_ID)); + return fragment; + } + + private static TReportExecStatusParams report(TStatusCode statusCode) { + return new TReportExecStatusParams() + .setFragmentId(FRAGMENT_ID) + .setBackendId(BACKEND_ID) + .setDone(true) + .setStatus(new TStatus(statusCode)) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR_MSG); + } + + @SuppressWarnings("unchecked") + private static void prepareReport(Coordinator coordinator) throws Exception { + Coordinator.PipelineExecContext context = Mockito.mock(Coordinator.PipelineExecContext.class); + Mockito.when(context.updatePipelineStatus(Mockito.any())).thenReturn(true); + Field contextsField = Coordinator.class.getDeclaredField("pipelineExecContexts"); + contextsField.setAccessible(true); + Map, Coordinator.PipelineExecContext> contexts = + (Map, Coordinator.PipelineExecContext>) contextsField.get(coordinator); + contexts.put(Pair.of(FRAGMENT_ID, BACKEND_ID), context); + + MarkedCountDownLatch latch = new MarkedCountDownLatch<>(1); + latch.addMark(FRAGMENT_ID, BACKEND_ID); + Field latchField = Coordinator.class.getDeclaredField("fragmentsDoneLatch"); + latchField.setAccessible(true); + latchField.set(coordinator, latch); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java new file mode 100644 index 00000000000000..3839270ee5f431 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java @@ -0,0 +1,171 @@ +// 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.doris.qe.runtime; + +import org.apache.doris.analysis.DescriptorTable; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Status; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.load.loadv2.LoadManager; +import org.apache.doris.load.loadv2.ProgressManager; +import org.apache.doris.nereids.trees.plans.distribute.PipelineDistributedPlan; +import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker; +import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJob; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanFragmentId; +import org.apache.doris.qe.CoordinatorContext; +import org.apache.doris.qe.NereidsCoordinator; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.system.Backend; +import org.apache.doris.thrift.TReportExecStatusParams; +import org.apache.doris.thrift.TStatus; +import org.apache.doris.thrift.TStatusCode; +import org.apache.doris.thrift.TUniqueId; + +import com.google.protobuf.ByteString; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +class LoadProcessorTest { + private static final long BACKEND_ID = 9; + private static final int FRAGMENT_ID = 7; + private static final String TRACKING_URL = "http://127.0.0.1/error-log"; + private static final String FIRST_ERROR = "Cannot convert value to integer"; + + @Test + void publishesDiagnosticsBeforeCancellationReleasesLoadWaiters() throws Exception { + Fixture fixture = createFixture(); + Mockito.doAnswer(invocation -> { + // LoadProcessor.cancel invokes remote cancellation before releasing its real latch. + Assertions.assertFalse(fixture.processor.isDone()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + Assertions.assertEquals(TStatusCode.DATA_QUALITY_ERROR, + fixture.coordinator.getExecStatus().getErrorCode()); + return null; + }).when(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + + Assertions.assertFalse(fixture.processor.await(0, TimeUnit.MILLISECONDS)); + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR)); + + Mockito.verify(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + Assertions.assertTrue(fixture.processor.await(0, TimeUnit.MILLISECONDS)); + Assertions.assertTrue(fixture.coordinator.join(1)); + } + + @Test + void successfulFinalReportPreservesDiagnostics() throws Exception { + Fixture fixture = createFixture(); + + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.OK) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR)); + + Assertions.assertTrue(fixture.coordinator.join(1)); + Assertions.assertTrue(fixture.coordinator.getExecStatus().ok()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + Mockito.verify(fixture.fragmentsTask, Mockito.never()).cancelExecute(Mockito.any(Status.class)); + } + + @Test + void failedReportWithoutDiagnosticsPreservesEarlierDiagnostics() throws Exception { + Fixture fixture = createFixture(); + fixture.processor.loadContext.updateTrackingUrl(TRACKING_URL); + fixture.processor.loadContext.updateFirstErrorMsg(FIRST_ERROR); + Mockito.doAnswer(invocation -> { + Assertions.assertFalse(fixture.processor.isDone()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + return null; + }).when(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR)); + + Mockito.verify(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + Assertions.assertTrue(fixture.coordinator.join(1)); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + } + + private static TReportExecStatusParams report(TStatusCode status) { + return new TReportExecStatusParams() + .setBackendId(BACKEND_ID) + .setFragmentId(FRAGMENT_ID) + .setDone(true) + .setStatus(new TStatus(status)); + } + + private static Fixture createFixture() { + PlanFragment fragment = Mockito.mock(PlanFragment.class); + Mockito.when(fragment.getFragmentId()).thenReturn(new PlanFragmentId(FRAGMENT_ID)); + UnassignedJob fragmentJob = Mockito.mock(UnassignedJob.class); + Mockito.when(fragmentJob.getFragment()).thenReturn(fragment); + PipelineDistributedPlan distributedPlan = Mockito.mock(PipelineDistributedPlan.class); + Mockito.when(distributedPlan.getFragmentJob()).thenReturn(fragmentJob); + Mockito.when(distributedPlan.getInstanceJobs()).thenReturn(Collections.emptyList()); + + Env env = Mockito.mock(Env.class); + Mockito.when(env.getLoadManager()).thenReturn(Mockito.mock(LoadManager.class)); + Mockito.when(env.getProgressManager()).thenReturn(Mockito.mock(ProgressManager.class)); + NereidsCoordinator coordinator = Mockito.mock(NereidsCoordinator.class, Mockito.CALLS_REAL_METHODS); + CoordinatorContext context; + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + context = CoordinatorContext.buildForLoad(coordinator, -1, new TUniqueId(1, 2), + Collections.singletonList(fragment), Collections.singletonList(distributedPlan), + Collections.emptyList(), new DescriptorTable(), "UTC", true, false); + } + Deencapsulation.setField(coordinator, "coordinatorContext", context); + LoadProcessor processor = context.asLoadProcessor(); + + Backend backend = Mockito.mock(Backend.class); + Mockito.when(backend.getId()).thenReturn(BACKEND_ID); + BackendWorker worker = Mockito.mock(BackendWorker.class); + Mockito.when(worker.id()).thenReturn(BACKEND_ID); + BackendServiceProxy backendProxy = Mockito.mock(BackendServiceProxy.class); + SingleFragmentPipelineTask fragmentTask = new SingleFragmentPipelineTask( + backend, FRAGMENT_ID, Collections.singleton(new TUniqueId(3, 4))); + MultiFragmentsPipelineTask fragmentsTask = Mockito.spy(new MultiFragmentsPipelineTask( + context, backend, backendProxy, ByteString.EMPTY, + Collections.singletonMap(FRAGMENT_ID, fragmentTask))); + Mockito.doNothing().when(fragmentsTask).cancelExecute(Mockito.any(Status.class)); + processor.setPipelineExecutionTask(new PipelineExecutionTask(context, backendProxy, + Collections.singletonMap(worker, fragmentsTask))); + return new Fixture(coordinator, processor, fragmentsTask); + } + + private static class Fixture { + final NereidsCoordinator coordinator; + final LoadProcessor processor; + final MultiFragmentsPipelineTask fragmentsTask; + + Fixture(NereidsCoordinator coordinator, LoadProcessor processor, MultiFragmentsPipelineTask fragmentsTask) { + this.coordinator = coordinator; + this.processor = processor; + this.fragmentsTask = fragmentsTask; + } + } +}