This is an automated email from the ASF dual-hosted git repository.
fpaul pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 126ae4df9ef [FLINK-29512][streaming] Align the checkpointId of
recovered CommittableWithLinages with the CommittableSummary
126ae4df9ef is described below
commit 126ae4df9ef8bab98a53433ef39c698cf8f04c60
Author: Fabian Paul <[email protected]>
AuthorDate: Tue Oct 11 14:52:44 2022 +0200
[FLINK-29512][streaming] Align the checkpointId of recovered
CommittableWithLinages with the CommittableSummary
Before this change during recovery for the CommitableCollector we
initialized the SubtaskCommittableManager always with the initial
checkpoint id (1) but the holding CheckpointCommittable with the
checkpoint id in state. This could lead to that the emitted
CommittableWithLinages update SubtaskCommittableManagers that they do
not belong to cause "Unknown subtask for <id>" failures.
---
.../CommittableCollectorSerializer.java | 30 ++++++++++++++++++---
.../operators/sink/CommitterOperatorTest.java | 14 +++++++---
.../CommittableCollectorSerializerTest.java | 31 ++++++++++++++++++++--
3 files changed, 66 insertions(+), 9 deletions(-)
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
index 6f8495b6637..761de50e922 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
@@ -19,7 +19,6 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
@@ -27,6 +26,8 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.DataOutputView;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -146,7 +147,7 @@ public final class CommittableCollectorSerializer<CommT>
long checkpointId = in.readLong();
List<SubtaskCommittableManager<CommT>> subtasks =
SimpleVersionedSerialization.readVersionAndDeserializeList(
- new SubtaskSimpleVersionedSerializer(), in);
+ new
SubtaskSimpleVersionedSerializer(checkpointId), in);
return new CheckpointCommittableManagerImpl<>(
subtasks.stream()
.collect(
@@ -161,6 +162,27 @@ public final class CommittableCollectorSerializer<CommT>
private class SubtaskSimpleVersionedSerializer
implements
SimpleVersionedSerializer<SubtaskCommittableManager<CommT>> {
+ @Nullable private final Long checkpointId;
+
+ /**
+ * This ctor must be used to create a deserializer where the
checkpointId is used to set the
+ * checkpointId of the deserialized SubtaskCommittableManager.
+ *
+ * @param checkpointId used to recover the SubtaskCommittableManager
+ */
+ public SubtaskSimpleVersionedSerializer(long checkpointId) {
+ this.checkpointId = checkpointId;
+ }
+
+ /**
+ * When using this ctor, you cannot use the serializer for
deserialization because it misses
+ * the checkpointId. For deserialization please use {@link
+ * #SubtaskSimpleVersionedSerializer(long)}.
+ */
+ public SubtaskSimpleVersionedSerializer() {
+ this.checkpointId = null;
+ }
+
@Override
public int getVersion() {
return 0;
@@ -192,7 +214,9 @@ public final class CommittableCollectorSerializer<CommT>
in.readInt(),
in.readInt(),
subtaskId,
- Sink.InitContext.INITIAL_CHECKPOINT_ID);
+ checkNotNull(
+ checkpointId,
+ "CheckpointId must be set to align the
SubtaskCommittableManager with holding CheckpointCommittableManager."));
}
private class RequestSimpleVersionedSerializer
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTest.java
index d7b0b550206..9af5b177b4a 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTest.java
@@ -235,14 +235,18 @@ class CommitterOperatorTest {
originalSubtaskId);
testHarness.open();
+ // We cannot test a different checkpoint thant 0 because when using
the OperatorTestHarness
+ // for recovery the lastCompleted checkpoint is always reset to 0.
+ long checkpointId = 0L;
+
final CommittableSummary<String> committableSummary =
- new CommittableSummary<>(originalSubtaskId, 1, 0L, 1, 1, 0);
+ new CommittableSummary<>(originalSubtaskId, 1, checkpointId,
1, 1, 0);
testHarness.processElement(new StreamRecord<>(committableSummary));
final CommittableWithLineage<String> first =
- new CommittableWithLineage<>("1", 0L, originalSubtaskId);
+ new CommittableWithLineage<>("1", checkpointId,
originalSubtaskId);
testHarness.processElement(new StreamRecord<>(first));
- final OperatorSubtaskState snapshot = testHarness.snapshot(0L, 2L);
+ final OperatorSubtaskState snapshot =
testHarness.snapshot(checkpointId, 2L);
// Trigger first checkpoint but committer needs retry
testHarness.notifyOfCompletedCheckpoint(0);
@@ -268,14 +272,16 @@ class CommitterOperatorTest {
assertThat(output).hasSize(2);
assertThat(committer.getSuccessfulCommits()).isEqualTo(1);
SinkV2Assertions.assertThat(toCommittableSummary(output.get(0)))
+ .hasCheckpointId(checkpointId)
.hasFailedCommittables(committableSummary.getNumberOfFailedCommittables())
.hasOverallCommittables(committableSummary.getNumberOfCommittables())
.hasPendingCommittables(0);
+ // Expect the same checkpointId that the original snapshot was made
with.
SinkV2Assertions.assertThat(toCommittableWithLinage(output.get(1)))
.isEqualTo(
new CommittableWithLineage<>(
- first.getCommittable(), 1L,
subtaskIdAfterRecovery));
+ first.getCommittable(), checkpointId,
subtaskIdAfterRecovery));
restored.close();
}
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
index af9e4070d61..b69b7a10695 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
+import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataOutputSerializer;
@@ -42,9 +43,11 @@ class CommittableCollectorSerializerTest {
private static final SimpleVersionedSerializer<Integer>
COMMITTABLE_SERIALIZER =
new IntegerSerializer();
-
+ private static final int SUBTASK_ID = 1;
+ private static final int NUMBER_OF_SUBTASKS = 1;
private static final CommittableCollectorSerializer<Integer> SERIALIZER =
- new CommittableCollectorSerializer<>(COMMITTABLE_SERIALIZER, 1, 1);
+ new CommittableCollectorSerializer<>(
+ COMMITTABLE_SERIALIZER, SUBTASK_ID, NUMBER_OF_SUBTASKS);
@Test
void testCommittableCollectorV1SerDe() throws IOException {
@@ -107,6 +110,30 @@ class CommittableCollectorSerializerTest {
"Deserialized CommittableCollector", subtaskId,
numberOfSubtasks, copy);
}
+ @Test
+ void
testAlignSubtaskCommittableManagerCheckpointWithCheckpointCommittableManagerCheckpointId()
+ throws IOException {
+ // Create CommittableCollector holding a higher checkpointId than
+ // Sink.InitContext#INITIAL_CHECKPOINT_ID
+ long checkpointId = Sink.InitContext.INITIAL_CHECKPOINT_ID + 1;
+ final CommittableCollector<Integer> committableCollector =
+ new CommittableCollector<>(SUBTASK_ID, NUMBER_OF_SUBTASKS);
+ committableCollector.addMessage(
+ new CommittableSummary<>(SUBTASK_ID, NUMBER_OF_SUBTASKS,
checkpointId, 1, 1, 0));
+ committableCollector.addMessage(new CommittableWithLineage<>(1,
checkpointId, SUBTASK_ID));
+
+ final CommittableCollector<Integer> copy =
+ SERIALIZER.deserialize(2,
SERIALIZER.serialize(committableCollector));
+
+ final Collection<CheckpointCommittableManagerImpl<Integer>>
checkpointCommittables =
+ copy.getCheckpointCommittables();
+ assertThat(checkpointCommittables).hasSize(1);
+ final CheckpointCommittableManagerImpl<Integer> committableManager =
+ checkpointCommittables.iterator().next();
+
assertThat(committableManager.getSubtaskCommittableManager(SUBTASK_ID).getCheckpointId())
+ .isEqualTo(committableManager.getCheckpointId());
+ }
+
private void assertCommittableCollector(
String assertMessageHeading,
int subtaskId,
