From e85d144c706e8512f70baafbc62669a1d4adf1a0 Mon Sep 17 00:00:00 2001 From: wudi Date: Wed, 4 Nov 2026 11:35:40 +0800 Subject: [PATCH] branch-4.1: [fix](fe) Report missing backend during stream load planning #69989 (#70069) Cherry-picked from #69989 --- .../apache/doris/load/StreamLoadHandler.java | 4 ++- .../doris/load/StreamLoadHandlerTest.java | 33 +++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java index 99f516f022e..63e2e667bc4 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/StreamLoadHandler.java @@ -142,7 +142,9 @@ public class StreamLoadHandler { if (request.isSetBackendId()) { long backendId = request.getBackendId(); Backend backend = Env.getCurrentSystemInfo().getBackend(backendId); - Preconditions.checkNotNull(backend); + if (backend == null) { + throw new UserException("Backend " + backendId + " does not exist"); + } String computeGroup = backend.getCloudClusterName(); // Token/auth-code and user-less internal loads keep their existing trusted path. Regular // stream loads must still validate compute group privilege, existence, and status. diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java index 8f266e81ab8..992b14908b1 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/StreamLoadHandlerTest.java @@ -27,8 +27,10 @@ import org.apache.doris.common.jmockit.Deencapsulation; import org.apache.doris.datasource.InternalCatalog; import org.apache.doris.mysql.privilege.Auth; import org.apache.doris.qe.ConnectContext; +import org.apache.doris.service.FrontendServiceImpl; import org.apache.doris.system.Backend; import org.apache.doris.system.SystemInfoService; +import org.apache.doris.thrift.TStatusCode; import org.apache.doris.thrift.TStreamLoadPutRequest; import org.apache.doris.thrift.TStreamLoadPutResult; @@ -38,6 +40,7 @@ import org.mockito.MockedStatic; import org.mockito.Mockito; import java.util.Arrays; +import java.util.Collections; import java.util.List; public class StreamLoadHandlerTest { @@ -91,6 +94,36 @@ public class StreamLoadHandlerTest { } } + @Test + public void testStreamLoadPutRejectsMissingBackend() throws Exception { + SystemInfoService originalSystemInfoService = Env.getCurrentSystemInfo(); + String originalCloudUniqueId = Config.cloud_unique_id; + CloudSystemInfoService systemInfoService = + new TestCloudSystemInfoService(Collections.emptyList()); + TStreamLoadPutRequest request = new TStreamLoadPutRequest(); + request.setUser(""); + request.setToken("test_token"); + request.setBackendId(10001L); + request.setCloudCluster("header_compute_group"); + + try { + Config.cloud_unique_id = "test_cloud_unique_id"; + Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo", systemInfoService); + ConnectContext.remove(); + + TStreamLoadPutResult result = new FrontendServiceImpl(null).streamLoadPut(request); + + Assert.assertEquals(TStatusCode.ANALYSIS_ERROR, result.getStatus().getStatusCode()); + Assert.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("10001")); + Assert.assertFalse(result.isSetPipelineParams()); + Assert.assertNull(ConnectContext.get()); + } finally { + ConnectContext.remove(); + Config.cloud_unique_id = originalCloudUniqueId; + Deencapsulation.setField(Env.getCurrentEnv(), "systemInfo", originalSystemInfoService); + } + } + @Test public void testGroupCommitValidatesBackendComputeGroupPrivilege() throws Exception { String originalCloudUniqueId = Config.cloud_unique_id;