simulate compaction and more complex situations

apache · Oct 21, 2024 · cd0587f · cd0587f
1 parent 4b13c49
commit cd0587f
Show file tree

Hide file tree

Showing 2 changed files with 100 additions and 14 deletions.
diff --git a/...-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java b/...-paimon/src/main/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonCommitter.java
@@ -27,6 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -49,7 +50,8 @@ public PaimonCommitter(Options catalogOptions, String commitUser) {
     }
 
     @Override
-    public void commit(Collection<CommitRequest<MultiTableCommittable>> commitRequests) {
+    public void commit(Collection<CommitRequest<MultiTableCommittable>> commitRequests)
+            throws IOException {
         if (commitRequests.isEmpty()) {
             return;
         }
@@ -60,27 +62,23 @@ public void commit(Collection<CommitRequest<MultiTableCommittable>> commitReques
                         .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);
         try {
-            if (retriedNumber > 0) {
-                storeMultiCommitter.filterAndCommit(
-                        Collections.singletonList(wrappedManifestCommittable));
-            } else {
-                storeMultiCommitter.commit(Collections.singletonList(wrappedManifestCommittable));
-            }
+            storeMultiCommitter.filterAndCommit(
+                    Collections.singletonList(wrappedManifestCommittable));
             commitRequests.forEach(CommitRequest::signalAlreadyCommitted);
             LOGGER.info(
-                    String.format(
-                            "Commit succeeded for %s with %s committable",
-                            checkpointId, committables.size()));
+                    "Commit succeeded for {} with {} 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()));
+                    "Commit failed for {} with {} committable",
+                    checkpointId,
+                    committables.size(),
+                    e);
         }
     }
 

diff --git a/...paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java b/...paimon/src/test/java/org/apache/flink/cdc/connectors/paimon/sink/v2/PaimonSinkITCase.java
@@ -454,6 +454,94 @@ public void testSinkWithMultiTables(String metastore)
                 Collections.singletonList(Row.ofKind(RowKind.INSERT, "1", "1")), result);
     }
 
+    @ParameterizedTest
+    @ValueSource(strings = {"filesystem", "hive"})
+    public void testDuplicateCommitAfterRestore(String metastore)
+            throws IOException, InterruptedException, Catalog.DatabaseNotEmptyException,
+                    Catalog.DatabaseNotExistException, SchemaEvolveException {
+        initialize(metastore);
+        PaimonSink<Event> paimonSink =
+                new PaimonSink<>(
+                        catalogOptions, new PaimonRecordEventSerializer(ZoneId.systemDefault()));
+        PaimonWriter<Event> writer = paimonSink.createWriter(new MockInitContext());
+        Committer<MultiTableCommittable> committer = paimonSink.createCommitter();
+
+        // insert
+        for (Event event : createTestEvents()) {
+            writer.write(event, null);
+        }
+        writer.flush(false);
+        Collection<Committer.CommitRequest<MultiTableCommittable>> commitRequests =
+                writer.prepareCommit().stream()
+                        .map(MockCommitRequestImpl::new)
+                        .collect(Collectors.toList());
+        committer.commit(commitRequests);
+
+        // We add a loop for restore 3 times
+        for (int i = 0; i < 3; i++) {
+            // We've two steps in checkpoint: 1. snapshotState(ckp); 2.
+            // notifyCheckpointComplete(ckp).
+            // It's possible that flink job will restore from a checkpoint with only step#1 finished
+            // and
+            // step#2 not.
+            // CommitterOperator will try to re-commit recovered transactions.
+            committer.commit(commitRequests);
+            List<DataChangeEvent> events =
+                    Arrays.asList(
+                            DataChangeEvent.insertEvent(
+                                    table1,
+                                    generator.generate(
+                                            new Object[] {
+                                                BinaryStringData.fromString(
+                                                        Integer.toString(i + 2)),
+                                                BinaryStringData.fromString(Integer.toString(i + 2))
+                                            })));
+            Assertions.assertDoesNotThrow(
+                    () -> {
+                        for (Event event : events) {
+                            writer.write(event, null);
+                        }
+                    });
+            writer.flush(false);
+            //Checkpoint id start from 1
+            long checkpointId = i + 2;
+            committer.commit(
+                    writer.prepareCommit().stream()
+                            .map(
+                                    committable -> {
+                                        // update the right checkpointId for MultiTableCommittable
+                                        return new MultiTableCommittable(
+                                                committable.getDatabase(),
+                                                committable.getTable(),
+                                                checkpointId,
+                                                committable.kind(),
+                                                committable.wrappedCommittable());
+                                    })
+                            .map(MockCommitRequestImpl::new)
+                            .collect(Collectors.toList()));
+        }
+
+        List<Row> result = new ArrayList<>();
+        tEnv.sqlQuery("select * from paimon_catalog.test.`table1$snapshots`")
+                .execute()
+                .collect()
+                .forEachRemaining(result::add);
+        Assertions.assertEquals(result.size(), 4);
+        result.clear();
+
+        tEnv.sqlQuery("select * from paimon_catalog.test.`table1`")
+                .execute()
+                .collect()
+                .forEachRemaining(result::add);
+        Assertions.assertEquals(
+                Arrays.asList(
+                        Row.ofKind(RowKind.INSERT, "1", "1"),
+                        Row.ofKind(RowKind.INSERT, "2", "2"),
+                        Row.ofKind(RowKind.INSERT, "3", "3"),
+                        Row.ofKind(RowKind.INSERT, "4", "4")),
+                result);
+    }
+
     private static class MockCommitRequestImpl<CommT> extends CommitRequestImpl<CommT> {
 
         protected MockCommitRequestImpl(CommT committable) {