[coordinator] Log the reason for failures in bucket or replica state changes (apache#2145)

Author: platinumhamburg

fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/ReplicaStateMachine.java

@@ -221,8 +221,8 @@ private void doHandleStateChanges(
                                 try {
                                     partitionName = getPartitionName(tableBucket);
                                 } catch (PartitionNotExistException e) {
-                                    LOG.error(e.getMessage());
-                                    logFailedSateChange(replica, currentState, targetState);
+                                    logFailedSateChange(
+                                            replica, currentState, targetState, e.getMessage());
                                     return;
                                 }
 
@@ -282,11 +282,11 @@ private void doHandleStateChanges(
                         try {
                             partitionName = getPartitionName(tableBucket);
                         } catch (PartitionNotExistException e) {
-                            LOG.error(e.getMessage());
                             logFailedSateChange(
                                     tableBucketReplica,
                                     coordinatorContext.getReplicaState(tableBucketReplica),
-                                    targetState);
+                                    targetState,
+                                    e.getMessage());
                             continue;
                         }
                         // send leader request to the replica server
@@ -343,7 +343,11 @@ protected Collection<TableBucketReplica> checkValidReplicaStateChange(
                                 return true;
                             } else {
                                 logInvalidTransition(replica, curState, targetState);
-                                logFailedSateChange(replica, curState, targetState);
+                                logFailedSateChange(
+                                        replica,
+                                        curState,
+                                        targetState,
+                                        "Invalid Replica State Transition.");
                                 return false;
                             }
                         })
@@ -376,12 +380,16 @@ private void logInvalidTransition(
     }
 
     private void logFailedSateChange(
-            TableBucketReplica replica, ReplicaState currState, ReplicaState targetState) {
+            TableBucketReplica replica,
+            ReplicaState currState,
+            ReplicaState targetState,
+            String reason) {
         LOG.error(
-                "Fail to change state for table bucket replica {} from {} to {}.",
+                "Fail to change state for table bucket replica {} from {} to {}, reason: {}.",
                 stringifyReplica(replica),
                 currState,
-                targetState);
+                targetState,
+                reason);
     }
 
     private void logSuccessfulStateChange(

fluss-server/src/main/java/org/apache/fluss/server/coordinator/statemachine/TableBucketStateMachine.java

@@ -209,10 +209,13 @@ private void doHandleStateChange(
                     partitionName =
                             coordinatorContext.getPartitionName(tableBucket.getPartitionId());
                     if (partitionName == null) {
-                        LOG.error(
-                                "Can't find partition name for partition: {}.",
-                                tableBucket.getBucket());
-                        logFailedStateChange(tableBucket, currentState, targetState);
+                        logFailedStateChange(
+                                tableBucket,
+                                currentState,
+                                targetState,
+                                String.format(
+                                        "Can't find partition name for partition: %s.",
+                                        tableBucket.getBucket()));
                         return;
                     }
                 }
@@ -222,7 +225,8 @@ private void doHandleStateChange(
                     Optional<ElectionResult> optionalElectionResult =
                             initLeaderForTableBuckets(tableBucket, assignedServers);
                     if (!optionalElectionResult.isPresent()) {
-                        logFailedStateChange(tableBucket, currentState, targetState);
+                        logFailedStateChange(
+                                tableBucket, currentState, targetState, "Elect Result is empty.");
                     } else {
                         // transmit state
                         doStateChange(tableBucket, targetState);
@@ -244,7 +248,8 @@ private void doHandleStateChange(
                             electNewLeaderForTableBuckets(
                                     tableBucket, replicaLeaderElectionStrategy);
                     if (!optionalElectionResult.isPresent()) {
-                        logFailedStateChange(tableBucket, currentState, targetState);
+                        logFailedStateChange(
+                                tableBucket, currentState, targetState, "Elect result is empty.");
                     } else {
                         // transmit state
                         doStateChange(tableBucket, targetState);
@@ -337,10 +342,13 @@ public void batchHandleOnlineChangeAndInitLeader(Set<TableBucket> tableBuckets)
             if (tableBucket.getPartitionId() != null) {
                 partitionName = coordinatorContext.getPartitionName(tableBucket.getPartitionId());
                 if (partitionName == null) {
-                    LOG.error(
-                            "Can't find partition name for partition: {}.",
-                            tableBucket.getBucket());
-                    logFailedStateChange(tableBucket, currentState, BucketState.OnlineBucket);
+                    logFailedStateChange(
+                            tableBucket,
+                            currentState,
+                            BucketState.OnlineBucket,
+                            String.format(
+                                    "Can't find partition name for partition: %s.",
+                                    tableBucket.getBucket()));
                     continue;
                 }
             }
@@ -350,7 +358,11 @@ public void batchHandleOnlineChangeAndInitLeader(Set<TableBucket> tableBuckets)
             Optional<ElectionResult> optionalElectionResult =
                     doInitElectionForBucket(tableBucket, assignedServers);
             if (!optionalElectionResult.isPresent()) {
-                logFailedStateChange(tableBucket, currentState, BucketState.OnlineBucket);
+                logFailedStateChange(
+                        tableBucket,
+                        currentState,
+                        BucketState.OnlineBucket,
+                        "Elect result is empty.");
                 continue;
             }
             ElectionResult electionResult = optionalElectionResult.get();
@@ -506,7 +518,8 @@ private boolean checkValidTableBucketStateChange(
             return true;
         } else {
             logInvalidTransition(tableBucket, curState, targetState);
-            logFailedStateChange(tableBucket, curState, targetState);
+            logFailedStateChange(
+                    tableBucket, curState, targetState, "Invalid TableBucket State Transition.");
             return false;
         }
     }
@@ -537,12 +550,16 @@ private void logInvalidTransition(
     }
 
     private void logFailedStateChange(
-            TableBucket tableBucket, BucketState currState, BucketState targetState) {
+            TableBucket tableBucket,
+            BucketState currState,
+            BucketState targetState,
+            String reason) {
         LOG.error(
-                "Fail to change state for table bucket {} from {} to {}.",
+                "Fail to change state for table bucket {} from {} to {}, reason: {}",
                 stringifyBucket(tableBucket),
                 currState,
-                targetState);
+                targetState,
+                reason);
     }
 
     private void logSuccessfulStateChange(
@@ -610,8 +627,12 @@ private Optional<ElectionResult> electLeader(
 
         if (!resultOpt.isPresent()) {
             LOG.error(
-                    "The leader election for table bucket {} is empty.",
-                    stringifyBucket(tableBucket));
+                    "The leader election for table bucket {} is empty, assignment: {}, live replicas: {}, leaderAndIsr: {}, strategy: {}",
+                    stringifyBucket(tableBucket),
+                    assignment,
+                    liveReplicas,
+                    leaderAndIsr,
+                    electionStrategy);
             return Optional.empty();
         }
         return resultOpt;