[fluss-server] Restrict replicasOnOffline to fatal errors only

Only fatal errors (STORAGE_EXCEPTION, LOG_STORAGE_EXCEPTION,
KV_STORAGE_EXCEPTION, UNKNOWN_SERVER_ERROR) mark replicas offline.
All other NotifyLeaderAndIsr errors are transient and do not affect
leader election eligibility.

Also fixes ReplicaManager to return LeaderNotAvailableException
instead of StorageException when leaderId is null or negative.

Author: platinumhamburg

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -59,6 +59,7 @@
 import org.apache.fluss.rpc.messages.RebalanceResponse;
 import org.apache.fluss.rpc.messages.RemoveServerTagResponse;
 import org.apache.fluss.rpc.protocol.ApiError;
+import org.apache.fluss.rpc.protocol.Errors;
 import org.apache.fluss.server.coordinator.event.AccessContextEvent;
 import org.apache.fluss.server.coordinator.event.AddServerTagEvent;
 import org.apache.fluss.server.coordinator.event.AdjustIsrReceivedEvent;
@@ -342,8 +343,14 @@ private void initCoordinatorContext() throws Exception {
         long start4loadTabletServer = System.currentTimeMillis();
         Map<Integer, TabletServerRegistration> tabletServerRegistrations =
                 zooKeeperClient.getTabletServers(currentServers);
+        List<Integer> skippedNullRegistration = new ArrayList<>();
+        List<Integer> skippedNoEndpoint = new ArrayList<>();
         for (int server : currentServers) {
             TabletServerRegistration registration = tabletServerRegistrations.get(server);
+            if (registration == null) {
+                skippedNullRegistration.add(server);
+                continue;
+            }
             ServerInfo serverInfo =
                     new ServerInfo(
                             server,
@@ -357,6 +364,7 @@ private void initCoordinatorContext() throws Exception {
                         "Can not find endpoint for listener name {} for tablet server {}",
                         internalListenerName,
                         serverInfo);
+                skippedNoEndpoint.add(server);
                 continue;
             }
             tabletServerInfos.add(serverInfo);
@@ -370,8 +378,30 @@ private void initCoordinatorContext() throws Exception {
 
         coordinatorContext.setLiveTabletServers(tabletServerInfos);
         LOG.info(
-                "Load tablet servers success in {}ms when initializing coordinator context.",
-                System.currentTimeMillis() - start4loadTabletServer);
+                "Load tablet servers success in {}ms when initializing coordinator context. "
+                        + "ZK returned {} servers, loaded {} into liveSet, "
+                        + "skipped {} (null registration), skipped {} (no endpoint). "
+                        + "Live server IDs: {}",
+                System.currentTimeMillis() - start4loadTabletServer,
+                currentServers.length,
+                tabletServerInfos.size(),
+                skippedNullRegistration.size(),
+                skippedNoEndpoint.size(),
+                tabletServerInfos.stream()
+                        .map(s -> String.valueOf(s.id()))
+                        .collect(Collectors.joining(",")));
+        if (!skippedNullRegistration.isEmpty()) {
+            LOG.warn(
+                    "Skipped {} servers with null ZK registration: {}",
+                    skippedNullRegistration.size(),
+                    skippedNullRegistration);
+        }
+        if (!skippedNoEndpoint.isEmpty()) {
+            LOG.warn(
+                    "Skipped {} servers with no internal endpoint: {}",
+                    skippedNoEndpoint.size(),
+                    skippedNoEndpoint);
+        }
 
         // init tablet server channels
         coordinatorChannelManager.startup(internalServerNodes);
@@ -938,11 +968,25 @@ private void processNotifyLeaderAndIsrResponseReceivedEvent(
                 notifyLeaderAndIsrResponseReceivedEvent.getNotifyLeaderAndIsrResultForBuckets();
         for (NotifyLeaderAndIsrResultForBucket notifyLeaderAndIsrResultForBucket :
                 notifyLeaderAndIsrResultForBuckets) {
-            // if the error code is not none, we will consider it as offline
             if (notifyLeaderAndIsrResultForBucket.failed()) {
-                offlineReplicas.add(
-                        new TableBucketReplica(
-                                notifyLeaderAndIsrResultForBucket.getTableBucket(), serverId));
+                Errors error = notifyLeaderAndIsrResultForBucket.getError().error();
+                TableBucket tableBucket = notifyLeaderAndIsrResultForBucket.getTableBucket();
+                if (isFatalReplicaError(error)) {
+                    LOG.warn(
+                            "Fatal NotifyLeaderAndIsr error for bucket {} on server {}: {}. "
+                                    + "Marking replica offline.",
+                            tableBucket,
+                            serverId,
+                            notifyLeaderAndIsrResultForBucket.getError());
+                    offlineReplicas.add(new TableBucketReplica(tableBucket, serverId));
+                } else {
+                    LOG.warn(
+                            "Transient NotifyLeaderAndIsr error for bucket {} on server {}: {}. "
+                                    + "Replica remains online.",
+                            tableBucket,
+                            serverId,
+                            notifyLeaderAndIsrResultForBucket.getError());
+                }
             }
         }
         if (!offlineReplicas.isEmpty()) {
@@ -951,6 +995,18 @@ private void processNotifyLeaderAndIsrResponseReceivedEvent(
         }
     }
 
+    /**
+     * Returns true if the error indicates a fatal replica failure (storage corruption, unknown
+     * internal error) that warrants excluding this replica from future leader elections. All other
+     * errors are considered transient and should NOT mark the replica offline.
+     */
+    private static boolean isFatalReplicaError(Errors error) {
+        return error == Errors.STORAGE_EXCEPTION
+                || error == Errors.LOG_STORAGE_EXCEPTION
+                || error == Errors.KV_STORAGE_EXCEPTION
+                || error == Errors.UNKNOWN_SERVER_ERROR;
+    }
+
     private void onReplicaBecomeOffline(Set<TableBucketReplica> offlineReplicas) {
         LOG.info("The replica {} become offline.", offlineReplicas);
         for (TableBucketReplica offlineReplica : offlineReplicas) {

fluss-server/src/main/java/org/apache/fluss/server/replica/ReplicaManager.java

@@ -27,6 +27,7 @@
 import org.apache.fluss.exception.InvalidColumnProjectionException;
 import org.apache.fluss.exception.InvalidCoordinatorException;
 import org.apache.fluss.exception.InvalidRequiredAcksException;
+import org.apache.fluss.exception.LeaderNotAvailableException;
 import org.apache.fluss.exception.LogOffsetOutOfRangeException;
 import org.apache.fluss.exception.LogStorageException;
 import org.apache.fluss.exception.NotLeaderOrFollowerException;
@@ -1188,17 +1189,16 @@ private void addFetcherForReplicas(
             Integer leaderId = replica.getLeaderId();
             TableBucket tb = replica.getTableBucket();
             LogTablet logTablet = replica.getLogTablet();
-            if (leaderId == null) {
+            if (leaderId == null || leaderId < 0) {
                 result.put(
                         tb,
                         new NotifyLeaderAndIsrResultForBucket(
                                 tb,
                                 ApiError.fromThrowable(
-                                        new StorageException(
+                                        new LeaderNotAvailableException(
                                                 String.format(
-                                                        "Could not find leader for follower replica %s while make "
-                                                                + "follower for %s.",
-                                                        serverId, tb)))));
+                                                        "No leader available for follower replica %s on server %s.",
+                                                        tb, serverId)))));
             } else {
                 bucketAndStatus.put(
                         tb,