This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new deb079478d0 [fix](fe) Report missing backend during stream load
planning (#69989)
deb079478d0 is described below
commit deb079478d0114d79b04166b22861918fcdd7a51
Author: wudi <[email protected]>
AuthorDate: Tue Nov 3 14:23:43 2026 +0800
[fix](fe) Report missing backend during stream load planning (#69989)
### What problem does this PR solve?
Issue Number: None
Problem Summary:
A cloud Stream Load request can still arrive from a BE after scaling has
removed that BE from FE metadata. Resolving the compute group then
obtains a null backend and throws a NullPointerException, which reaches
the caller as a generic INTERNAL_ERROR.
Return a descriptive UserException containing the missing backend ID
instead. The existing RPC exception handler converts it to
ANALYSIS_ERROR. The request fails before plan generation, allowing the
existing CDC task failure and automatic rescheduling flow to handle
recovery.
Add an RPC-level test for a missing backend with token authentication
and a compute-group header. It checks the returned status and backend
ID, absence of an execution plan, and ConnectContext cleanup.
---
.../org/apache/doris/load/StreamLoadHandler.java | 4 ++-
.../apache/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 06fde6758dc..84acd0f2b87 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
@@ -146,7 +146,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 6ec4ea387c0..00e4a149b84 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
@@ -35,6 +35,7 @@ import org.apache.doris.mysql.privilege.Auth;
import org.apache.doris.nereids.load.NereidsStreamLoadPlanner;
import org.apache.doris.nereids.load.NereidsStreamLoadTask;
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.TDataSink;
@@ -47,6 +48,7 @@ import org.apache.doris.thrift.TOlapTableSink;
import org.apache.doris.thrift.TPipelineFragmentParams;
import org.apache.doris.thrift.TPlanFragment;
import org.apache.doris.thrift.TQueryOptions;
+import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
import org.apache.doris.thrift.TTabletLocation;
@@ -59,6 +61,7 @@ import org.mockito.MockedStatic;
import org.mockito.Mockito;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
@@ -272,6 +275,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);
+
+ Assertions.assertEquals(TStatusCode.ANALYSIS_ERROR,
result.getStatus().getStatusCode());
+
Assertions.assertTrue(result.getStatus().getErrorMsgs().get(0).contains("10001"));
+ Assertions.assertFalse(result.isSetPipelineParams());
+ Assertions.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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
