[hotfix] Improve bucket ready check logic (#565)

Author: luoyuxia

fluss-client/src/main/java/com/alibaba/fluss/client/write/RecordAccumulator.java

@@ -393,11 +393,8 @@ private long bucketReady(
 
         boolean exhausted = writerBufferPool.queued() > 0;
         for (Map.Entry<Integer, Deque<WriteBatch>> entry : batches.entrySet()) {
-            int bucketId = entry.getKey();
             Deque<WriteBatch> deque = entry.getValue();
 
-            TableBucket tableBucket = cluster.getTableBucket(physicalTablePath, bucketId);
-            ServerNode leader = cluster.leaderFor(tableBucket);
             final long waitedTimeMs;
             final int dequeSize;
             final boolean full;
@@ -419,6 +416,9 @@ private long bucketReady(
                 full = dequeSize > 1 || batch.isClosed();
             }
 
+            int bucketId = entry.getKey();
+            TableBucket tableBucket = cluster.getTableBucket(physicalTablePath, bucketId);
+            ServerNode leader = cluster.leaderFor(tableBucket);
             if (leader == null) {
                 // This is a bucket for which leader is not known, but messages are
                 // available to send. Note that entries are currently not removed from