This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 906b918ea [flink] Resolve bounded stream input finished in multiple
checkpoint interval then cause commit failed (#2462)
906b918ea is described below
commit 906b918ea2d8fd92a50c10fd66c01a9fa6a6bbf5
Author: YeJunHao <[email protected]>
AuthorDate: Wed Dec 13 18:02:03 2023 +0800
[flink] Resolve bounded stream input finished in multiple checkpoint
interval then cause commit failed (#2462)
---
.../org/apache/paimon/flink/sink/Committer.java | 3 +
.../paimon/flink/sink/CommitterOperator.java | 21 +++-
.../apache/paimon/flink/sink/StoreCommitter.java | 9 ++
.../paimon/flink/sink/StoreMultiCommitter.java | 10 +-
.../paimon/flink/sink/CommitterOperatorTest.java | 135 +++++++++++++++++++++
5 files changed, 174 insertions(+), 4 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
index a58d97e4c..328c95e40 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/Committer.java
@@ -41,6 +41,9 @@ public interface Committer<CommitT, GlobalCommitT> extends
AutoCloseable {
GlobalCommitT combine(long checkpointId, long watermark, List<CommitT>
committables)
throws IOException;
+ GlobalCommitT combine(
+ long checkpointId, long watermark, GlobalCommitT t, List<CommitT>
committables);
+
/** Commits the given {@link GlobalCommitT}. */
void commit(List<GlobalCommitT> globalCommittables) throws IOException,
InterruptedException;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
index 16563cb77..4d8b3a4f5 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/CommitterOperator.java
@@ -193,16 +193,31 @@ public class CommitterOperator<CommitT, GlobalCommitT>
extends AbstractStreamOpe
for (Map.Entry<Long, List<CommitT>> entry : grouped.entrySet()) {
Long cp = entry.getKey();
List<CommitT> committables = entry.getValue();
- if (committablesPerCheckpoint.containsKey(cp)) {
+ // To prevent the asynchronous completion of tasks with multiple
concurrent bounded
+ // stream inputs, which leads to some tasks passing a Committable
with cp =
+ // Long.MAX_VALUE during the endInput method call of the current
checkpoint, while other
+ // tasks pass a Committable with Long.MAX_VALUE during other
checkpoints hence causing
+ // an error here, we have a special handling for Committables with
Long.MAX_VALUE:
+ // instead of throwing an error, we merge them.
+ if (cp != null && cp == Long.MAX_VALUE &&
committablesPerCheckpoint.containsKey(cp)) {
+ // Merge the Long.MAX_VALUE committables here.
+ GlobalCommitT commitT =
+ committer.combine(
+ cp,
+ currentWatermark,
+ committablesPerCheckpoint.get(cp),
+ committables);
+ committablesPerCheckpoint.put(cp, commitT);
+ } else if (committablesPerCheckpoint.containsKey(cp)) {
throw new RuntimeException(
String.format(
"Repeatedly commit the same checkpoint files.
\n"
+ "The previous files is %s, \n"
+ "and the subsequent files is %s",
committablesPerCheckpoint.get(cp),
committables));
+ } else {
+ committablesPerCheckpoint.put(cp, toCommittables(cp,
committables));
}
-
- committablesPerCheckpoint.put(cp, toCommittables(cp,
committables));
}
this.inputs.clear();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
index 7857242bd..620bcbbc4 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
@@ -69,6 +69,15 @@ public class StoreCommitter implements
Committer<Committable, ManifestCommittabl
public ManifestCommittable combine(
long checkpointId, long watermark, List<Committable> committables)
{
ManifestCommittable manifestCommittable = new
ManifestCommittable(checkpointId, watermark);
+ return combine(checkpointId, watermark, manifestCommittable,
committables);
+ }
+
+ @Override
+ public ManifestCommittable combine(
+ long checkpointId,
+ long watermark,
+ ManifestCommittable manifestCommittable,
+ List<Committable> committables) {
for (Committable committable : committables) {
switch (committable.kind()) {
case FILE:
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
index 089311208..a2d3f16ba 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
@@ -89,8 +89,16 @@ public class StoreMultiCommitter
long checkpointId, long watermark, List<MultiTableCommittable>
committables) {
WrappedManifestCommittable wrappedManifestCommittable =
new WrappedManifestCommittable(checkpointId, watermark);
- for (MultiTableCommittable committable : committables) {
+ return combine(checkpointId, watermark, wrappedManifestCommittable,
committables);
+ }
+ @Override
+ public WrappedManifestCommittable combine(
+ long checkpointId,
+ long watermark,
+ WrappedManifestCommittable wrappedManifestCommittable,
+ List<MultiTableCommittable> committables) {
+ for (MultiTableCommittable committable : committables) {
ManifestCommittable manifestCommittable =
wrappedManifestCommittable.computeCommittableIfAbsent(
Identifier.create(committable.getDatabase(),
committable.getTable()),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
index a30be40ef..6c4b46ce3 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/CommitterOperatorTest.java
@@ -25,10 +25,13 @@ import org.apache.paimon.data.GenericRow;
import org.apache.paimon.flink.VersionedSerializerWrapper;
import org.apache.paimon.flink.utils.MetricUtils;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.CompactIncrement;
+import org.apache.paimon.io.NewFilesIncrement;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.manifest.ManifestCommittableSerializer;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.StreamWriteBuilder;
@@ -278,6 +281,138 @@ public class CommitterOperatorTest extends
CommitterOperatorTestBase {
Assertions.assertThat(actual).hasSameElementsAs(Lists.newArrayList(commitUser));
}
+ @Test
+ public void testCommitInputEnd() throws Exception {
+ FileStoreTable table = createFileStoreTable();
+ String commitUser = UUID.randomUUID().toString();
+ OneInputStreamOperator<Committable, Committable> operator =
+ createCommitterOperator(table, commitUser, new
NoopCommittableStateManager());
+ OneInputStreamOperatorTestHarness<Committable, Committable>
testHarness =
+ createTestHarness(operator);
+ testHarness.open();
+ Assertions.assertThatCode(
+ () -> {
+ long time = System.currentTimeMillis();
+ long cp = 0L;
+ testHarness.processElement(
+ new Committable(
+ Long.MAX_VALUE,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.snapshot(cp++, time++);
+ testHarness.processElement(
+ new Committable(
+ Long.MAX_VALUE,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.processElement(
+ new Committable(
+ Long.MAX_VALUE,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.snapshot(cp++, time++);
+ testHarness.snapshot(cp, time);
+ })
+ .doesNotThrowAnyException();
+
+ if (operator instanceof CommitterOperator) {
+ Assertions.assertThat(
+ ((ManifestCommittable)
+ ((CommitterOperator) operator)
+
.committablesPerCheckpoint.get(Long.MAX_VALUE))
+ .fileCommittables()
+ .size())
+ .isEqualTo(3);
+ }
+
+ Assertions.assertThatCode(
+ () -> {
+ long time = System.currentTimeMillis();
+ long cp = 0L;
+ testHarness.processElement(
+ new Committable(
+ 0L,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.snapshot(cp++, time++);
+ testHarness.processElement(
+ new Committable(
+ 0L,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.processElement(
+ new Committable(
+ Long.MAX_VALUE,
+ Committable.Kind.FILE,
+ new CommitMessageImpl(
+ BinaryRow.EMPTY_ROW,
+ 0,
+ new NewFilesIncrement(
+
Collections.emptyList(),
+
Collections.emptyList()),
+ new CompactIncrement(
+
Collections.emptyList(),
+
Collections.emptyList(),
+
Collections.emptyList()))),
+ cp);
+ testHarness.snapshot(cp++, time++);
+ testHarness.snapshot(cp, time);
+ })
+ .hasRootCauseInstanceOf(RuntimeException.class)
+ .cause()
+ .hasMessageContaining("Repeatedly commit the same checkpoint
files.");
+ }
+
private static OperatorSubtaskState writeAndSnapshot(
FileStoreTable table,
String commitUser,
