fix

Author: LsomeYeah

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/DynamicBucketRowWriteOperator.java

@@ -60,7 +60,9 @@ public void processElement(StreamRecord<Tuple2<InternalRow, Integer>> element)
     @Override
     protected List<Committable> prepareCommit(boolean waitCompaction, long checkpointId)
             throws IOException {
-        return super.prepareCommit(waitCompaction, checkpointId);
+        List<Committable> committables = super.prepareCommit(waitCompaction, checkpointId);
+        tryRefreshWrite();
+        return committables;
     }
 
     /** {@link StreamOperatorFactory} of {@link DynamicBucketRowWriteOperator}. */

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/RowDataStoreWriteOperator.java

@@ -220,7 +220,7 @@ protected List<Committable> prepareCommit(boolean waitCompaction, long checkpoin
                                                     Committable.Kind.LOG_OFFSET,
                                                     new LogOffsetCommittable(k, v))));
         }
-
+        tryRefreshWrite();
         return committables;
     }
 

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/TableWriteOperator.java

@@ -159,17 +159,15 @@ public void close() throws Exception {
     @Override
     protected List<Committable> prepareCommit(boolean waitCompaction, long checkpointId)
             throws IOException {
-        List<Committable> committables = write.prepareCommit(waitCompaction, checkpointId);
-        tryRefreshWrite();
-        return committables;
+        return write.prepareCommit(waitCompaction, checkpointId);
     }
 
     @VisibleForTesting
     public StoreSinkWrite getWrite() {
         return write;
     }
 
-    private void tryRefreshWrite() {
+    protected void tryRefreshWrite() {
         if (writeRefresher != null) {
             writeRefresher.tryRefresh();
             table = writeRefresher.updatedTable();