branch-4.1: [fix](fe) Report missing backend during stream load planning #69989 (#70069)

Cherry-picked from #69989

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.

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;