[FLINK-35893][cdc-connect][paimon] use filterAndCommit API for retried Committables.

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java

@@ -24,8 +24,9 @@
 import org.apache.paimon.flink.sink.StoreMultiCommitter;
 import org.apache.paimon.manifest.WrappedManifestCommittable;
 import org.apache.paimon.options.Options;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -34,6 +35,8 @@
 /** A {@link Committer} to commit write results for multiple tables. */
 public class PaimonCommitter implements Committer<MultiTableCommittable> {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(PaimonCommitter.class);
+
     private final StoreMultiCommitter storeMultiCommitter;
 
     public PaimonCommitter(Options catalogOptions, String commitUser) {
@@ -46,19 +49,39 @@ public PaimonCommitter(Options catalogOptions, String commitUser) {
     }
 
     @Override
-    public void commit(Collection<CommitRequest<MultiTableCommittable>> commitRequests)
-            throws IOException, InterruptedException {
+    public void commit(Collection<CommitRequest<MultiTableCommittable>> commitRequests) {
         if (commitRequests.isEmpty()) {
             return;
         }
+
         List<MultiTableCommittable> committables =
                 commitRequests.stream()
                         .map(CommitRequest::getCommittable)
                         .collect(Collectors.toList());
+        // All CommitRequest shared the same checkpointId.
         long checkpointId = committables.get(0).checkpointId();
+        int retriedNumber = commitRequests.stream().findFirst().get().getNumberOfRetries();
         WrappedManifestCommittable wrappedManifestCommittable =
                 storeMultiCommitter.combine(checkpointId, 1L, committables);
-        storeMultiCommitter.commit(Collections.singletonList(wrappedManifestCommittable));
+        try {
+            if (retriedNumber > 0) {
+                storeMultiCommitter.filterAndCommit(
+                        Collections.singletonList(wrappedManifestCommittable));
+            } else {
+                storeMultiCommitter.commit(Collections.singletonList(wrappedManifestCommittable));
+            }
+            commitRequests.forEach(CommitRequest::signalAlreadyCommitted);
+            LOGGER.info(
+                    String.format(
+                            "Commit succeeded for %s with %s committable",
+                            checkpointId, committables.size()));
+        } catch (Exception e) {
+            commitRequests.forEach(CommitRequest::retryLater);
+            LOGGER.warn(
+                    String.format(
+                            "Commit failed for %s with %s committable",
+                            checkpointId, committables.size()));
+        }
     }
 
     @Override