[server] feat: committed isr boundary

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

@@ -1925,7 +1925,8 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
         // TODO verify leader epoch.
 
         List<AdjustIsrResultForBucket> result = new ArrayList<>();
-        Map<TableBucket, LeaderAndIsr> newLeaderAndIsrList = new HashMap<>();
+        Map<TableBucket, LeaderAndIsr> proposedChanges = new HashMap<>();
+        Map<TableBucket, LeaderAndIsr> committedChanges = new HashMap<>();
         for (Map.Entry<TableBucket, LeaderAndIsr> entry : leaderAndIsrList.entrySet()) {
             TableBucket tableBucket = entry.getKey();
             LeaderAndIsr tryAdjustLeaderAndIsr = entry.getValue();
@@ -1959,41 +1960,42 @@ private List<AdjustIsrResultForBucket> tryProcessAdjustIsr(
                             tryAdjustLeaderAndIsr.standbyReplicas(),
                             coordinatorContext.getCoordinatorEpoch(),
                             currentLeaderAndIsr.bucketEpoch() + 1);
-            newLeaderAndIsrList.put(tableBucket, newLeaderAndIsr);
+            proposedChanges.put(tableBucket, newLeaderAndIsr);
         }
 
         try {
             zooKeeperClient.batchUpdateLeaderAndIsr(
-                    newLeaderAndIsrList, coordinatorContext.getCoordinatorZkVersion());
-            newLeaderAndIsrList.forEach(
+                    proposedChanges, coordinatorContext.getCoordinatorZkVersion());
+            committedChanges.putAll(proposedChanges);
+            committedChanges.forEach(
                     (tableBucket, newLeaderAndIsr) ->
                             result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr)));
         } catch (Exception batchException) {
             LOG.error("Error when batch update leader and isr. Try one by one.", batchException);
 
-            for (Map.Entry<TableBucket, LeaderAndIsr> entry : newLeaderAndIsrList.entrySet()) {
+            for (Map.Entry<TableBucket, LeaderAndIsr> entry : proposedChanges.entrySet()) {
                 TableBucket tableBucket = entry.getKey();
                 LeaderAndIsr newLeaderAndIsr = entry.getValue();
                 try {
                     zooKeeperClient.updateLeaderAndIsr(
                             tableBucket,
                             newLeaderAndIsr,
                             coordinatorContext.getCoordinatorZkVersion());
+                    committedChanges.put(tableBucket, newLeaderAndIsr);
+                    result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
                 } catch (Exception e) {
                     LOG.error("Error when register leader and isr.", e);
                     result.add(
                             new AdjustIsrResultForBucket(tableBucket, ApiError.fromThrowable(e)));
                 }
-                // Successful return.
-                result.add(new AdjustIsrResultForBucket(tableBucket, newLeaderAndIsr));
             }
         }
 
         // update coordinator leader and isr cache.
-        newLeaderAndIsrList.forEach(coordinatorContext::putBucketLeaderAndIsr);
+        committedChanges.forEach(coordinatorContext::putBucketLeaderAndIsr);
 
         // First, try to judge whether the bucket is in rebalance task when isr change.
-        newLeaderAndIsrList.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange);
+        committedChanges.keySet().forEach(this::tryToCompleteRebalanceTaskOnLeaderAndIsrChange);
 
         // TODO update metadata for all alive tablet servers.