[fix](job) fix NPE in routine load Kafka meta request (#63180)

### What problem does this PR solve?

Problem Summary:

In a single-BE deployment, Kafka routine load fetches topic metadata
through the only BE. If that BE cannot connect to Kafka, the metadata
request fails and the BE is skipped in the current retry loop. Then FE
may have no normal candidate backend left and falls back to backend ids
in the routine load blacklist.

The blacklist can contain stale backend ids that no longer exist in
`SystemInfoService`. In that case, `KafkaUtil` may get a null `Backend`
and throw a NullPointerException when calling `be.getHost()`. This hides
the real Kafka metadata error, such as broker connection failure.

This PR filters stale backend ids when reading the routine load
blacklist and adds a final null check before creating the BE address.
The original Kafka metadata error is preserved instead of being replaced
by the secondary NPE.

A regression case is added with an invalid `kafka_broker_list` to verify
that routine load reports the expected Kafka metadata error path.

Author: sollhui

fe/fe-core/src/main/java/org/apache/doris/datasource/kafka/KafkaUtil.java

@@ -252,8 +252,16 @@ private static InternalService.PProxyResult getInfoRequest(InternalService.PProx
                 // 2. If that sole backend is decommissioned, the aliveBackends list becomes empty.
                 // Hence, in such cases, it's essential to rely on the blacklist to obtain meta information.
                 if (backendIds.isEmpty()) {
-                    for (Long beId : Env.getCurrentEnv().getRoutineLoadManager().getBlacklist().keySet()) {
-                        backendIds.add(beId);
+                    Map<Long, Long> blacklist = Env.getCurrentEnv().getRoutineLoadManager().getBlacklist();
+                    for (Long beId : blacklist.keySet()) {
+                        Backend backend = Env.getCurrentSystemInfo().getBackend(beId);
+                        if (backend != null) {
+                            backendIds.add(beId);
+                        } else {
+                            blacklist.remove(beId);
+                            LOG.warn("remove stale backend {} from routine load blacklist when getting kafka meta",
+                                    beId);
+                        }
                     }
                 }
                 if (backendIds.isEmpty()) {
@@ -264,7 +272,16 @@ private static InternalService.PProxyResult getInfoRequest(InternalService.PProx
                     throw new LoadException("failed to get info: " + errorMsg + ",");
                 }
                 Collections.shuffle(backendIds);
-                Backend be = Env.getCurrentSystemInfo().getBackend(backendIds.get(0));
+                long selectedBeId = backendIds.get(0);
+                Backend be = Env.getCurrentSystemInfo().getBackend(selectedBeId);
+                if (be == null) {
+                    if (errorMsg == null) {
+                        errorMsg = "backend " + selectedBeId + " does not exist";
+                    }
+                    LOG.warn("skip stale backend {} when getting kafka meta", selectedBeId);
+                    retryTimes++;
+                    continue;
+                }
                 address = new TNetworkAddress(be.getHost(), be.getBrpcPort());
                 long beId = be.getId();
 

regression-test/suites/load_p0/routine_load/test_black_list.groovy

@@ -150,5 +150,61 @@ suite("test_black_list","nonConcurrent,p0") {
             GetDebugPoint().disableDebugPointForAllBEs(inject)
             sql "stop routine load for ${job}"
         }
+
+        def invalidBrokerTableName = "test_black_list_invalid_broker"
+        def invalidBrokerJob = "test_black_list_invalid_broker_job"
+        sql """ DROP TABLE IF EXISTS ${invalidBrokerTableName} """
+        sql """
+            CREATE TABLE IF NOT EXISTS ${invalidBrokerTableName} (
+                `k1` int(20) NULL,
+                `k2` string NULL,
+                `v1` date  NULL,
+                `v2` string  NULL,
+                `v3` datetime  NULL,
+                `v4` string  NULL
+            ) ENGINE=OLAP
+            DUPLICATE KEY(`k1`)
+            COMMENT 'OLAP'
+            DISTRIBUTED BY HASH(`k1`) BUCKETS 3
+            PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+        """
+
+        try {
+            sql """
+                CREATE ROUTINE LOAD ${invalidBrokerJob} ON ${invalidBrokerTableName}
+                COLUMNS TERMINATED BY ","
+                FROM KAFKA
+                (
+                    "kafka_broker_list" = "127.0.0.1:1",
+                    "kafka_topic" = "${kafkaCsvTpoics[0]}",
+                    "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+                );
+            """
+
+            def count = 0
+            while (true) {
+                sleep(1000)
+                def state = sql "show routine load for ${invalidBrokerJob}"
+                def stateChangedReason = state[0][17].toString()
+                def otherMsg = state[0][19].toString()
+                def errorMsg = "${stateChangedReason} ${otherMsg}"
+                log.info("invalid broker routine load state: ${state[0][8].toString()}".toString())
+                log.info("invalid broker reason of state changed: ${stateChangedReason}".toString())
+                log.info("invalid broker other msg: ${otherMsg}".toString())
+                if (errorMsg.contains("Failed to get all partitions of kafka topic")) {
+                    assertTrue(errorMsg.contains("failed to get info"))
+                    assertTrue(errorMsg.contains("failed to get partition meta: Local: Broker transport failure"))
+                    break
+                }
+                if (count >= 90) {
+                    log.error("routine load invalid broker test fail")
+                    assertEquals(1, 2)
+                    break
+                }
+                count++
+            }
+        } finally {
+            try_sql "stop routine load for ${invalidBrokerJob}"
+        }
     }
 }