You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2024/09/17 21:20:25 UTC
(flink) branch master updated (96b84258b6e -> 6d60f41a9b1)
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 96b84258b6e [FLINK-36208][core] Use ThreadLocalRandom in AbstractID
new c56def03b1d [FLINK-25920] Straighten EOI handling in CommittableCollector
new 8ce679ae847 [FLINK-25920] Turn CommittableManager#merge functional
new ad01d71f871 [FLINK-25920] Improve sink test assertions
new 1d32f1b539c [FLINK-25920] Only process complete batches of committables
new 42174085765 [FLINK-25920] Refactor & Revise SinkWriterOperatorTestBase
new 37e6724813b [FLINK-25920] Handle duplicate EOI in Sink
new 2cdd3f05ba8 [FLINK-25920] Improve logging in committable handling of the sink
new 6d60f41a9b1 [FLINK-25920] Fix AbstractStreamingWriter sending after EOI
The 8 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 184 ++++++++---------
.../sink/compactor/operator/CompactorOperator.java | 10 +-
.../operator/CompactorOperatorStateHandler.java | 10 +-
.../file/table/stream/AbstractStreamingWriter.java | 56 ++++-
.../core/io/SimpleVersionedSerializerAdapter.java | 58 ++++++
.../api/connector/sink2/CommittableMessage.java | 22 +-
.../sink2/CommittableMessageSerializer.java | 19 +-
.../sink2/CommittableMessageTypeInfo.java | 2 +-
.../api/connector/sink2/CommittableSummary.java | 58 +++++-
.../connector/sink2/CommittableWithLineage.java | 43 +++-
.../connector/sink2/GlobalCommitterOperator.java | 5 +-
.../runtime/operators/sink/CommitterOperator.java | 27 +--
.../runtime/operators/sink/SinkWriterOperator.java | 69 ++++++-
.../CheckpointCommittableManagerImpl.java | 69 ++++---
.../sink/committables/CommittableCollector.java | 13 +-
.../CommittableCollectorSerializer.java | 16 +-
.../sink/committables/CommittableManager.java | 8 +-
.../committables/SubtaskCommittableManager.java | 52 +++--
.../sink2/CommittableMessageSerializerTest.java | 4 +-
.../connector/sink2/CommittableSummaryAssert.java | 51 +----
.../sink2/CommittableWithLinageAssert.java | 34 +---
.../sink2/GlobalCommitterOperatorTest.java | 9 +-
.../CheckpointCommittableManagerImplTest.java | 32 +--
.../committables/CommittableCollectorTest.java | 6 +-
.../SubtaskCommittableManagerTest.java | 57 +++---
.../SinkV1TransformationTranslatorITCase.java | 8 +-
.../api/graph/StreamingJobGraphGeneratorTest.java | 5 +-
.../operators/sink/CommitterOperatorTestBase.java | 15 +-
.../runtime/operators/sink/SinkTestUtil.java | 4 +-
.../SinkV2CommitterOperatorDeprecatedTest.java | 3 -
.../sink/SinkV2CommitterOperatorTest.java | 13 +-
.../SinkV2SinkWriterOperatorDeprecatedTest.java | 88 +++-----
.../sink/SinkV2SinkWriterOperatorTest.java | 76 ++++---
.../operators/sink/SinkWriterOperatorTestBase.java | 226 +++++++++------------
.../streaming/runtime/operators/sink/TestSink.java | 98 ++++-----
.../runtime/operators/sink/TestSinkV2.java | 111 +++++-----
.../sink/WithAdapterCommitterOperatorTest.java | 13 +-
.../sink/WithAdapterSinkWriterOperatorTest.java | 84 ++++----
.../operators/sink/deprecated/TestSinkV2.java | 82 +++-----
.../nodes/exec/common/CommonExecSinkITCase.java | 5 +-
.../flink/test/streaming/runtime/SinkITCase.java | 100 +++++++--
.../runtime/SinkV2MetricsDeprecatedITCase.java | 1 -
.../streaming/runtime/SinkV2MetricsITCase.java | 5 +-
43 files changed, 988 insertions(+), 863 deletions(-)
create mode 100644 flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerializerAdapter.java
(flink) 05/08: [FLINK-25920] Refactor & Revise SinkWriterOperatorTestBase
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4217408576552fc929d9c8331495a9282064cd9c
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Tue Sep 10 11:47:02 2024 +0200
[FLINK-25920] Refactor & Revise SinkWriterOperatorTestBase
The stateful SinkWriterOperatorTestBase test cases used EOI to manipulate the state which was never clean. In particular, it also stored the input elements in state until EOI arrived and emitted them all at once. For state restoration tests, we emitted records after EOI arrived.
This commit changed the writer state completely to just capture the record count, which is much more realistic than storing actual payload. The tests now directly assert on the state instead of output.
This commit also introduces an adaptor for serializing basic types in the writer state and replaces the hard-to-maintain SinkAndSuppliers with an InspectableSink in the sink writer tests that require an abstraction on top of the different Sink flavors.
---
.../core/io/SimpleVersionedSerializerAdapter.java | 58 ++++++
.../SinkV1TransformationTranslatorITCase.java | 8 +-
.../api/graph/StreamingJobGraphGeneratorTest.java | 5 +-
.../runtime/operators/sink/SinkTestUtil.java | 4 +-
.../SinkV2CommitterOperatorDeprecatedTest.java | 3 -
.../sink/SinkV2CommitterOperatorTest.java | 13 +-
.../SinkV2SinkWriterOperatorDeprecatedTest.java | 88 ++++-----
.../sink/SinkV2SinkWriterOperatorTest.java | 76 ++++----
.../operators/sink/SinkWriterOperatorTestBase.java | 217 +++++++++------------
.../streaming/runtime/operators/sink/TestSink.java | 98 ++++------
.../runtime/operators/sink/TestSinkV2.java | 111 +++++------
.../sink/WithAdapterCommitterOperatorTest.java | 13 +-
.../sink/WithAdapterSinkWriterOperatorTest.java | 84 ++++----
.../operators/sink/deprecated/TestSinkV2.java | 82 ++++----
.../nodes/exec/common/CommonExecSinkITCase.java | 5 +-
.../flink/test/streaming/runtime/SinkITCase.java | 3 -
.../runtime/SinkV2MetricsDeprecatedITCase.java | 1 -
.../streaming/runtime/SinkV2MetricsITCase.java | 5 +-
18 files changed, 400 insertions(+), 474 deletions(-)
diff --git a/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerializerAdapter.java b/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerializerAdapter.java
new file mode 100644
index 00000000000..03056be26a9
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/core/io/SimpleVersionedSerializerAdapter.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.core.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Adapter for {@link TypeSerializer} to {@link SimpleVersionedSerializer}. The implementation is
+ * naive and should only be used for non-critical paths and tests.
+ */
+@Internal
+public class SimpleVersionedSerializerAdapter<T>
+ implements SimpleVersionedSerializer<T>, Serializable {
+ private final TypeSerializer<T> serializer;
+
+ public SimpleVersionedSerializerAdapter(TypeSerializer<T> serializer) {
+ this.serializer = serializer;
+ }
+
+ public int getVersion() {
+ return serializer.snapshotConfiguration().getCurrentVersion();
+ }
+
+ public byte[] serialize(T value) throws IOException {
+ DataOutputSerializer dataOutputSerializer = new DataOutputSerializer(10);
+ serializer.serialize(value, dataOutputSerializer);
+ return dataOutputSerializer.getCopyOfBuffer();
+ }
+
+ public T deserialize(int version, byte[] serialized) throws IOException {
+ DataInputDeserializer dataInputDeserializer = new DataInputDeserializer(serialized);
+ T value = serializer.deserialize(dataInputDeserializer);
+ dataInputDeserializer.releaseArrays();
+ return value;
+ }
+}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV1TransformationTranslatorITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV1TransformationTranslatorITCase.java
index 8f4ddf77389..2685d57b6da 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV1TransformationTranslatorITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/SinkV1TransformationTranslatorITCase.java
@@ -121,13 +121,7 @@ class SinkV1TransformationTranslatorITCase
void generateWriterGlobalCommitterTopology() {
final StreamGraph streamGraph =
buildGraph(
- TestSink.newBuilder()
- .setCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
- .setGlobalCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
- .setDefaultGlobalCommitter()
- .build(),
+ TestSink.newBuilder().setDefaultGlobalCommitter().build(),
runtimeExecutionMode);
final StreamNode sourceNode = findNodeName(streamGraph, node -> node.contains("Source"));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
index 34085974363..1972d9e7697 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
@@ -34,6 +34,7 @@ import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.api.connector.sink2.CommitterInitContext;
import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
@@ -55,6 +56,7 @@ import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.execution.CheckpointingMode;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializerAdapter;
import org.apache.flink.core.memory.ManagedMemoryUseCase;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -115,7 +117,6 @@ import org.apache.flink.streaming.api.transformations.MultipleInputTransformatio
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.StreamExchangeMode;
-import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2;
import org.apache.flink.streaming.runtime.partitioner.BroadcastPartitioner;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
@@ -2684,7 +2685,7 @@ class StreamingJobGraphGeneratorTest {
@Override
public SimpleVersionedSerializer<String> getWriteResultSerializer() {
- return new TestSinkV2.StringSerializer();
+ return new SimpleVersionedSerializerAdapter<>(StringSerializer.INSTANCE);
}
@Override
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkTestUtil.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkTestUtil.java
index b457b5a0b78..cfdae58596c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkTestUtil.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkTestUtil.java
@@ -52,7 +52,7 @@ class SinkTestUtil {
static byte[] toBytes(String obj) {
try {
return SimpleVersionedSerialization.writeVersionAndSerialize(
- TestSinkV2.StringSerializer.INSTANCE, obj);
+ TestSinkV2.COMMITTABLE_SERIALIZER, obj);
} catch (IOException e) {
throw new IllegalStateException(e);
}
@@ -83,7 +83,7 @@ class SinkTestUtil {
static String fromBytes(byte[] obj) {
try {
return SimpleVersionedSerialization.readVersionAndDeSerialize(
- TestSinkV2.StringSerializer.INSTANCE, obj);
+ TestSinkV2.COMMITTABLE_SERIALIZER, obj);
} catch (IOException e) {
throw new IllegalStateException(e);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorDeprecatedTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorDeprecatedTest.java
index 51d2fa0d8d0..3aacaa6d1f8 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorDeprecatedTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorDeprecatedTest.java
@@ -35,7 +35,6 @@ class SinkV2CommitterOperatorDeprecatedTest extends CommitterOperatorTestBase {
(TwoPhaseCommittingSink<?, String>)
TestSinkV2.newBuilder()
.setCommitter(committer)
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.setWithPostCommitTopology(true)
.build(),
() -> committer.successfulCommits);
@@ -47,7 +46,6 @@ class SinkV2CommitterOperatorDeprecatedTest extends CommitterOperatorTestBase {
(TwoPhaseCommittingSink<?, String>)
TestSinkV2.newBuilder()
.setCommitter(new TestSinkV2.RetryOnceCommitter())
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.setWithPostCommitTopology(true)
.build(),
() -> 0);
@@ -60,7 +58,6 @@ class SinkV2CommitterOperatorDeprecatedTest extends CommitterOperatorTestBase {
(TwoPhaseCommittingSink<?, String>)
TestSinkV2.newBuilder()
.setCommitter(committer)
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.setWithPostCommitTopology(false)
.build(),
() -> committer.successfulCommits);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorTest.java
index 349e3961e43..0112b5cf862 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2CommitterOperatorTest.java
@@ -30,7 +30,6 @@ class SinkV2CommitterOperatorTest extends CommitterOperatorTestBase {
(SupportsCommitter<String>)
TestSinkV2.newBuilder()
.setCommitter(committer)
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.setWithPostCommitTopology(true)
.build(),
() -> committer.successfulCommits);
@@ -42,7 +41,6 @@ class SinkV2CommitterOperatorTest extends CommitterOperatorTestBase {
(SupportsCommitter<String>)
TestSinkV2.newBuilder()
.setCommitter(new TestSinkV2.RetryOnceCommitter())
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.setWithPostCommitTopology(true)
.build(),
() -> 0);
@@ -52,12 +50,11 @@ class SinkV2CommitterOperatorTest extends CommitterOperatorTestBase {
SinkAndCounters sinkWithoutPostCommit() {
ForwardingCommitter committer = new ForwardingCommitter();
return new SinkAndCounters(
- (SupportsCommitter<String>)
- TestSinkV2.newBuilder()
- .setCommitter(committer)
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
- .setWithPostCommitTopology(false)
- .build(),
+ TestSinkV2.newBuilder()
+ .setCommitter(committer)
+ .setWithPostCommitTopology(false)
+ .build()
+ .asSupportsCommitter(),
() -> committer.successfulCommits);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorDeprecatedTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorDeprecatedTest.java
index bf06065b17c..4e0c3ad0cec 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorDeprecatedTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorDeprecatedTest.java
@@ -18,16 +18,13 @@
package org.apache.flink.streaming.runtime.operators.sink;
+import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.runtime.operators.sink.deprecated.TestSinkV2;
-import org.apache.flink.shaded.guava32.com.google.common.collect.ImmutableList;
-
-import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.List;
/**
@@ -35,67 +32,49 @@ import java.util.List;
*/
@Deprecated
class SinkV2SinkWriterOperatorDeprecatedTest extends SinkWriterOperatorTestBase {
-
@Override
- SinkAndSuppliers sinkWithoutCommitter() {
+ InspectableSink sinkWithoutCommitter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter = new TestSinkV2.DefaultSinkWriter<>();
- return new SinkAndSuppliers(
- TestSinkV2.<Integer>newBuilder().setWriter(sinkWriter).build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ return new InspectableSink(TestSinkV2.<Integer>newBuilder().setWriter(sinkWriter).build());
}
@Override
- SinkAndSuppliers sinkWithCommitter() {
+ InspectableSink sinkWithCommitter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter =
new TestSinkV2.DefaultCommittingSinkWriter<>();
- return new SinkAndSuppliers(
+ return new InspectableSink(
TestSinkV2.<Integer>newBuilder()
- .setWriter(sinkWriter)
.setDefaultCommitter()
- .build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ .setWriter(sinkWriter)
+ .build());
}
@Override
- SinkAndSuppliers sinkWithTimeBasedWriter() {
+ InspectableSink sinkWithTimeBasedWriter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter = new TimeBasedBufferingSinkWriter();
- return new SinkAndSuppliers(
+ return new InspectableSink(
TestSinkV2.<Integer>newBuilder()
.setWriter(sinkWriter)
.setDefaultCommitter()
- .build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ .build());
}
@Override
- SinkAndSuppliers sinkWithSnapshottingWriter(boolean withState, String stateName) {
- SnapshottingBufferingSinkWriter sinkWriter = new SnapshottingBufferingSinkWriter();
+ InspectableSink sinkWithState(boolean withState, String stateName) {
+ TestSinkV2.DefaultSinkWriter<Integer> sinkWriter =
+ new TestSinkV2.DefaultStatefulSinkWriter<>();
TestSinkV2.Builder<Integer> builder =
- TestSinkV2.newBuilder()
- .setWriter(sinkWriter)
+ TestSinkV2.<Integer>newBuilder()
.setDefaultCommitter()
- .setWithPostCommitTopology(true);
+ .setWithPostCommitTopology(true)
+ .setWriter(sinkWriter);
if (withState) {
builder.setWriterState(true);
}
if (stateName != null) {
builder.setCompatibleStateNames(stateName);
}
- return new SinkAndSuppliers(
- builder.build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> sinkWriter.lastCheckpointId,
- () -> new TestSinkV2.StringSerializer());
+ return new InspectableSink(builder.build());
}
private static class TimeBasedBufferingSinkWriter
@@ -125,31 +104,30 @@ class SinkV2SinkWriterOperatorDeprecatedTest extends SinkWriterOperatorTestBase
}
}
- private static class SnapshottingBufferingSinkWriter
- extends TestSinkV2.DefaultStatefulSinkWriter {
- public static final int NOT_SNAPSHOTTED = -1;
- long lastCheckpointId = NOT_SNAPSHOTTED;
- boolean endOfInput = false;
+ static class InspectableSink extends AbstractInspectableSink<TestSinkV2<Integer>> {
+ InspectableSink(TestSinkV2<Integer> sink) {
+ super(sink);
+ }
+
+ @Override
+ public long getLastCheckpointId() {
+ return getSink().getWriter().lastCheckpointId;
+ }
@Override
- public void flush(boolean endOfInput) throws IOException, InterruptedException {
- this.endOfInput = endOfInput;
+ public List<String> getRecordsOfCurrentCheckpoint() {
+ return getSink().getWriter().elements;
}
@Override
- public List<String> snapshotState(long checkpointId) throws IOException {
- lastCheckpointId = checkpointId;
- return super.snapshotState(checkpointId);
+ public List<Watermark> getWatermarks() {
+ return getSink().getWriter().watermarks;
}
@Override
- public Collection<String> prepareCommit() {
- if (!endOfInput) {
- return ImmutableList.of();
- }
- List<String> result = elements;
- elements = new ArrayList<>();
- return result;
+ public int getRecordCountFromState() {
+ return ((TestSinkV2.DefaultStatefulSinkWriter<?>) getSink().getWriter())
+ .getRecordCount();
}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorTest.java
index d46cf0c05fc..2d41721cbfb 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkV2SinkWriterOperatorTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.runtime.operators.sink;
+import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -30,67 +31,49 @@ import java.util.Collection;
import java.util.List;
class SinkV2SinkWriterOperatorTest extends SinkWriterOperatorTestBase {
-
@Override
- SinkAndSuppliers sinkWithoutCommitter() {
+ InspectableSink sinkWithoutCommitter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter = new TestSinkV2.DefaultSinkWriter<>();
- return new SinkAndSuppliers(
- TestSinkV2.<Integer>newBuilder().setWriter(sinkWriter).build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ return new InspectableSink(TestSinkV2.<Integer>newBuilder().setWriter(sinkWriter).build());
}
@Override
- SinkAndSuppliers sinkWithCommitter() {
+ InspectableSink sinkWithCommitter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter =
new TestSinkV2.DefaultCommittingSinkWriter<>();
- return new SinkAndSuppliers(
+ return new InspectableSink(
TestSinkV2.<Integer>newBuilder()
.setWriter(sinkWriter)
.setDefaultCommitter()
- .build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ .build());
}
@Override
- SinkAndSuppliers sinkWithTimeBasedWriter() {
+ InspectableSink sinkWithTimeBasedWriter() {
TestSinkV2.DefaultSinkWriter<Integer> sinkWriter = new TimeBasedBufferingSinkWriter();
- return new SinkAndSuppliers(
+ return new InspectableSink(
TestSinkV2.<Integer>newBuilder()
.setWriter(sinkWriter)
.setDefaultCommitter()
- .build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- TestSinkV2.StringSerializer::new);
+ .build());
}
@Override
- SinkAndSuppliers sinkWithSnapshottingWriter(boolean withState, String stateName) {
- SnapshottingBufferingSinkWriter sinkWriter = new SnapshottingBufferingSinkWriter();
+ InspectableSink sinkWithState(boolean withState, String stateName) {
+ TestSinkV2.DefaultSinkWriter<Integer> sinkWriter =
+ new TestSinkV2.DefaultStatefulSinkWriter<>();
TestSinkV2.Builder<Integer> builder =
- TestSinkV2.newBuilder()
- .setWriter(sinkWriter)
+ TestSinkV2.<Integer>newBuilder()
.setDefaultCommitter()
- .setWithPostCommitTopology(true);
+ .setWithPostCommitTopology(true)
+ .setWriter(sinkWriter);
if (withState) {
builder.setWriterState(true);
}
if (stateName != null) {
builder.setCompatibleStateNames(stateName);
}
- return new SinkAndSuppliers(
- builder.build(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> sinkWriter.lastCheckpointId,
- () -> new TestSinkV2.StringSerializer());
+ return new InspectableSink(builder.build());
}
private static class TimeBasedBufferingSinkWriter
@@ -147,4 +130,31 @@ class SinkV2SinkWriterOperatorTest extends SinkWriterOperatorTestBase {
return result;
}
}
+
+ static class InspectableSink extends AbstractInspectableSink<TestSinkV2<Integer>> {
+ InspectableSink(TestSinkV2<Integer> sink) {
+ super(sink);
+ }
+
+ @Override
+ public long getLastCheckpointId() {
+ return getSink().getWriter().lastCheckpointId;
+ }
+
+ @Override
+ public List<String> getRecordsOfCurrentCheckpoint() {
+ return getSink().getWriter().elements;
+ }
+
+ @Override
+ public List<Watermark> getWatermarks() {
+ return getSink().getWriter().watermarks;
+ }
+
+ @Override
+ public int getRecordCountFromState() {
+ return ((TestSinkV2.DefaultStatefulSinkWriter<?>) getSink().getWriter())
+ .getRecordCount();
+ }
+ }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
index 4ede8a6ee83..d6a40dc24b3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
@@ -28,7 +28,6 @@ import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerial
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
-import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataOutputSerializer;
@@ -66,8 +65,6 @@ import java.util.Queue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
-import java.util.function.LongSupplier;
-import java.util.function.Supplier;
import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.apache.flink.streaming.runtime.operators.sink.SinkTestUtil.fromOutput;
@@ -77,21 +74,21 @@ abstract class SinkWriterOperatorTestBase {
@Test
void testNotEmitCommittablesWithoutCommitter() throws Exception {
- SinkAndSuppliers sinkAndSuppliers = sinkWithoutCommitter();
+ InspectableSink sink = sinkWithoutCommitter();
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkAndSuppliers.sink));
+ new SinkWriterOperatorFactory<>(sink.getSink()));
testHarness.open();
testHarness.processElement(1, 1);
assertThat(testHarness.getOutput()).isEmpty();
- assertThat(sinkAndSuppliers.elementSupplier.get())
+ assertThat(sink.getRecordsOfCurrentCheckpoint())
.containsOnly("(1,1," + Long.MIN_VALUE + ")");
testHarness.prepareSnapshotPreBarrier(1);
assertThat(testHarness.getOutput()).isEmpty();
// Elements are flushed
- assertThat(sinkAndSuppliers.elementSupplier.get()).isEmpty();
+ assertThat(sink.getRecordsOfCurrentCheckpoint()).isEmpty();
testHarness.close();
}
@@ -99,10 +96,10 @@ abstract class SinkWriterOperatorTestBase {
void testWatermarkPropagatedToSinkWriter() throws Exception {
final long initialTime = 0;
- SinkAndSuppliers sinkAndSuppliers = sinkWithoutCommitter();
+ InspectableSink sink = sinkWithoutCommitter();
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkAndSuppliers.sink));
+ new SinkWriterOperatorFactory<>(sink.getSink()));
testHarness.open();
testHarness.processWatermark(initialTime);
@@ -110,7 +107,7 @@ abstract class SinkWriterOperatorTestBase {
assertThat(testHarness.getOutput())
.containsExactly(new Watermark(initialTime), new Watermark(initialTime + 1));
- assertThat(sinkAndSuppliers.watermarkSupplier.get())
+ assertThat(sink.getWatermarks())
.containsExactly(
new org.apache.flink.api.common.eventtime.Watermark(initialTime),
new org.apache.flink.api.common.eventtime.Watermark(initialTime + 1));
@@ -123,7 +120,7 @@ abstract class SinkWriterOperatorTestBase {
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkWithTimeBasedWriter().sink));
+ new SinkWriterOperatorFactory<>(sinkWithTimeBasedWriter().getSink()));
testHarness.open();
@@ -150,7 +147,7 @@ abstract class SinkWriterOperatorTestBase {
void testEmitOnFlushWithCommitter() throws Exception {
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkWithCommitter().sink));
+ new SinkWriterOperatorFactory<>(sinkWithCommitter().getSink()));
testHarness.open();
assertThat(testHarness.getOutput()).isEmpty();
@@ -168,7 +165,7 @@ abstract class SinkWriterOperatorTestBase {
@Test
void testEmitOnEndOfInputInBatchMode() throws Exception {
final SinkWriterOperatorFactory<Integer, Integer> writerOperatorFactory =
- new SinkWriterOperatorFactory<>(sinkWithCommitter().sink);
+ new SinkWriterOperatorFactory<>(sinkWithCommitter().getSink());
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(writerOperatorFactory);
@@ -186,9 +183,10 @@ abstract class SinkWriterOperatorTestBase {
final long initialTime = 0;
- final SinkAndSuppliers sinkAndSuppliers = sinkWithSnapshottingWriter(stateful, null);
+ final InspectableSink sink = sinkWithState(stateful, null);
+ Sink<Integer> sink2 = sink.getSink();
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
- createTestHarnessWithBufferingSinkWriter(sinkAndSuppliers.sink);
+ new OneInputStreamOperatorTestHarness<>(new SinkWriterOperatorFactory<>(sink2));
testHarness.open();
@@ -199,39 +197,23 @@ abstract class SinkWriterOperatorTestBase {
testHarness.prepareSnapshotPreBarrier(1L);
OperatorSubtaskState snapshot = testHarness.snapshot(1L, 1L);
- // we see the watermark and the committable summary, so the committables must be stored in
- // state
- assertThat(testHarness.getOutput()).hasSize(2).contains(new Watermark(initialTime));
- assertThat(sinkAndSuppliers.lastCheckpointSupplier.getAsLong())
- .isEqualTo(stateful ? 1L : -1L);
+ assertThat(sink.getRecordCountFromState()).isEqualTo(2);
+ assertThat(sink.getLastCheckpointId()).isEqualTo(stateful ? 1L : -1L);
testHarness.close();
- final SinkAndSuppliers restoredSink = sinkWithSnapshottingWriter(stateful, null);
+ final InspectableSink restoredSink = sinkWithState(stateful, null);
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>>
- restoredTestHarness = createTestHarnessWithBufferingSinkWriter(restoredSink.sink);
+ restoredTestHarness =
+ new OneInputStreamOperatorTestHarness<>(
+ new SinkWriterOperatorFactory<>(restoredSink.getSink()));
restoredTestHarness.initializeState(snapshot);
restoredTestHarness.open();
- // this will flush out the committables that were restored
- restoredTestHarness.endInput();
- final long checkpointId = 2;
- restoredTestHarness.prepareSnapshotPreBarrier(checkpointId);
- restoredTestHarness.notifyOfCompletedCheckpoint(checkpointId);
-
- if (stateful) {
- assertBasicOutput(restoredTestHarness.getOutput(), 2, EOI);
- } else {
- assertThat(fromOutput(restoredTestHarness.getOutput()).get(0).asRecord().getValue())
- .isInstanceOf(CommittableSummary.class)
- .satisfies(
- cs ->
- SinkV2Assertions.assertThat((CommittableSummary<?>) cs)
- .hasOverallCommittables(0)
- .hasPendingCommittables(0)
- .hasFailedCommittables(0));
- }
+ // check that the previous state is correctly restored
+ assertThat(restoredSink.getRecordCountFromState()).isEqualTo(stateful ? 2 : 0);
+
restoredTestHarness.close();
}
@@ -244,66 +226,46 @@ abstract class SinkWriterOperatorTestBase {
"bit", "mention", "thick", "stick", "stir", "easy", "sleep", "forth",
"cost", "prompt");
- SinkAndSuppliers sinkAndSuppliers =
- sinkWithSnapshottingWriter(stateful, DummySinkOperator.DUMMY_SINK_STATE_NAME);
+ InspectableSink sink = sinkWithState(stateful, CompatibleStateSinkOperator.SINK_STATE_NAME);
+ int expectedState = 5;
final OneInputStreamOperatorTestHarness<String, String> previousSink =
new OneInputStreamOperatorTestHarness<>(
- new DummySinkOperator(sinkAndSuppliers.serializerSupplier.get()),
+ new CompatibleStateSinkOperator<>(
+ TestSinkV2.WRITER_SERIALIZER, expectedState),
StringSerializer.INSTANCE);
OperatorSubtaskState previousSinkState =
TestHarnessUtil.buildSubtaskState(previousSink, previousSinkInputs);
- // 2. Load previous sink state and verify the output
+ // 2. Load previous sink state and verify state
+ Sink<Integer> sink3 = sink.getSink();
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>>
compatibleWriterOperator =
- createTestHarnessWithBufferingSinkWriter(sinkAndSuppliers.sink);
-
- final List<String> expectedOutput1 =
- stateful ? new ArrayList<>(previousSinkInputs) : new ArrayList<>();
- expectedOutput1.add(Tuple3.of(1, 1, Long.MIN_VALUE).toString());
+ new OneInputStreamOperatorTestHarness<>(
+ new SinkWriterOperatorFactory<>(sink3));
// load the state from previous sink
compatibleWriterOperator.initializeState(previousSinkState);
+ assertThat(sink.getRecordCountFromState()).isEqualTo(stateful ? expectedState : 0);
- compatibleWriterOperator.open();
-
- compatibleWriterOperator.processElement(1, 1);
-
- // this will flush out the committables that were restored from previous sink
- compatibleWriterOperator.endInput();
- compatibleWriterOperator.prepareSnapshotPreBarrier(1);
-
- OperatorSubtaskState operatorStateWithoutPreviousState =
- compatibleWriterOperator.snapshot(1L, 1L);
+ // 3. do another snapshot and check if this also can be restored without compabitible state
+ // name
+ compatibleWriterOperator.prepareSnapshotPreBarrier(1L);
+ OperatorSubtaskState snapshot = compatibleWriterOperator.snapshot(1L, 1L);
compatibleWriterOperator.close();
- assertEmitted(expectedOutput1, compatibleWriterOperator.getOutput());
-
- // 3. Restore the sink without previous sink's state
- SinkAndSuppliers sinkAndSuppliers2 =
- sinkWithSnapshottingWriter(stateful, DummySinkOperator.DUMMY_SINK_STATE_NAME);
+ // 4. Restore the sink without previous sink's state
+ InspectableSink sink2 =
+ sinkWithState(stateful, CompatibleStateSinkOperator.SINK_STATE_NAME);
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>>
restoredSinkOperator =
- createTestHarnessWithBufferingSinkWriter(sinkAndSuppliers2.sink);
- final List<String> expectedOutput2 =
- Arrays.asList(
- Tuple3.of(2, 2, Long.MIN_VALUE).toString(),
- Tuple3.of(3, 3, Long.MIN_VALUE).toString());
-
- restoredSinkOperator.initializeState(operatorStateWithoutPreviousState);
-
- restoredSinkOperator.open();
+ new OneInputStreamOperatorTestHarness<>(
+ new SinkWriterOperatorFactory<>(sink2.getSink()));
- restoredSinkOperator.processElement(2, 2);
- restoredSinkOperator.processElement(3, 3);
+ restoredSinkOperator.initializeState(snapshot);
+ assertThat(sink.getRecordCountFromState()).isEqualTo(stateful ? expectedState : 0);
- // this will flush out the committables that were restored
- restoredSinkOperator.endInput();
- restoredSinkOperator.prepareSnapshotPreBarrier(2);
-
- assertEmitted(expectedOutput2, restoredSinkOperator.getOutput());
restoredSinkOperator.close();
}
@@ -311,10 +273,10 @@ abstract class SinkWriterOperatorTestBase {
void testRestoreCommitterState() throws Exception {
final List<String> committables = Arrays.asList("state1", "state2");
- SinkAndSuppliers sinkAndSuppliers = sinkWithCommitter();
+ InspectableSink sink = sinkWithCommitter();
final OneInputStreamOperatorTestHarness<String, String> committer =
new OneInputStreamOperatorTestHarness<>(
- new TestCommitterOperator(sinkAndSuppliers.serializerSupplier.get()),
+ new TestCommitterOperator(TestSinkV2.COMMITTABLE_SERIALIZER),
StringSerializer.INSTANCE);
final OperatorSubtaskState committerState =
@@ -322,7 +284,7 @@ abstract class SinkWriterOperatorTestBase {
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkAndSuppliers.sink));
+ new SinkWriterOperatorFactory<>(sink.getSink()));
testHarness.initializeState(committerState);
@@ -362,16 +324,16 @@ abstract class SinkWriterOperatorTestBase {
@ParameterizedTest
@ValueSource(booleans = {true, false})
void testHandleEndInputInStreamingMode(boolean isCheckpointingEnabled) throws Exception {
- SinkAndSuppliers sinkAndSuppliers = sinkWithCommitter();
+ InspectableSink sink = sinkWithCommitter();
final OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>> testHarness =
new OneInputStreamOperatorTestHarness<>(
- new SinkWriterOperatorFactory<>(sinkAndSuppliers.sink));
+ new SinkWriterOperatorFactory<>(sink.getSink()));
testHarness.open();
testHarness.processElement(1, 1);
assertThat(testHarness.getOutput()).isEmpty();
final String record = "(1,1," + Long.MIN_VALUE + ")";
- assertThat(sinkAndSuppliers.elementSupplier.get()).containsOnly(record);
+ assertThat(sink.getRecordsOfCurrentCheckpoint()).containsOnly(record);
testHarness.endInput();
@@ -380,7 +342,7 @@ abstract class SinkWriterOperatorTestBase {
}
assertEmitted(Collections.singletonList(record), testHarness.getOutput());
- assertThat(sinkAndSuppliers.elementSupplier.get()).isEmpty();
+ assertThat(sink.getRecordsOfCurrentCheckpoint()).isEmpty();
testHarness.close();
}
@@ -556,13 +518,6 @@ abstract class SinkWriterOperatorTestBase {
assertThat(committables).containsExactlyInAnyOrderElementsOf(records);
}
- private static OneInputStreamOperatorTestHarness<Integer, CommittableMessage<Integer>>
- createTestHarnessWithBufferingSinkWriter(Sink sink) throws Exception {
- final SinkWriterOperatorFactory<Integer, Integer> writerOperatorFactory =
- new SinkWriterOperatorFactory<>(sink);
- return new OneInputStreamOperatorTestHarness<>(writerOperatorFactory);
- }
-
private static void assertBasicOutput(
Collection<Object> queuedOutput, int numberOfCommittables, long checkpointId) {
List<StreamElement> output = fromOutput(queuedOutput);
@@ -622,19 +577,22 @@ abstract class SinkWriterOperatorTestBase {
}
}
- private static class DummySinkOperator extends AbstractStreamOperator<String>
+ /** Writes state to test whether the sink can read from alternative state names. */
+ private static class CompatibleStateSinkOperator<T> extends AbstractStreamOperator<String>
implements OneInputStreamOperator<String, String> {
- static final String DUMMY_SINK_STATE_NAME = "dummy_sink_state";
+ static final String SINK_STATE_NAME = "compatible_sink_state";
static final ListStateDescriptor<byte[]> SINK_STATE_DESC =
- new ListStateDescriptor<>(
- DUMMY_SINK_STATE_NAME, BytePrimitiveArraySerializer.INSTANCE);
- ListState<String> sinkState;
- private final SimpleVersionedSerializer<String> serializer;
+ new ListStateDescriptor<>(SINK_STATE_NAME, BytePrimitiveArraySerializer.INSTANCE);
+ ListState<T> sinkState;
+ private final SimpleVersionedSerializer<T> serializer;
+ private final T initialState;
- public DummySinkOperator(SimpleVersionedSerializer<String> serializer) {
+ public CompatibleStateSinkOperator(
+ SimpleVersionedSerializer<T> serializer, T initialState) {
this.serializer = serializer;
+ this.initialState = initialState;
}
public void initializeState(StateInitializationContext context) throws Exception {
@@ -643,11 +601,14 @@ abstract class SinkWriterOperatorTestBase {
new SimpleVersionedListState<>(
context.getOperatorStateStore().getListState(SINK_STATE_DESC),
serializer);
+ if (!context.isRestored()) {
+ sinkState.add(initialState);
+ }
}
@Override
- public void processElement(StreamRecord<String> element) throws Exception {
- sinkState.add(element.getValue());
+ public void processElement(StreamRecord<String> element) {
+ // do nothing
}
}
@@ -672,32 +633,42 @@ abstract class SinkWriterOperatorTestBase {
}
}
- abstract SinkAndSuppliers sinkWithoutCommitter();
+ abstract InspectableSink sinkWithoutCommitter();
+
+ abstract InspectableSink sinkWithTimeBasedWriter();
+
+ abstract InspectableSink sinkWithState(boolean withState, String stateName);
+
+ abstract InspectableSink sinkWithCommitter();
- abstract SinkAndSuppliers sinkWithTimeBasedWriter();
+ /**
+ * Basic abstraction to access the different flavors of sinks. Remove once the older interfaces
+ * are removed.
+ */
+ interface InspectableSink {
+ long getLastCheckpointId();
- abstract SinkAndSuppliers sinkWithSnapshottingWriter(boolean withState, String stateName);
+ List<String> getRecordsOfCurrentCheckpoint();
- abstract SinkAndSuppliers sinkWithCommitter();
+ List<org.apache.flink.api.common.eventtime.Watermark> getWatermarks();
- static class SinkAndSuppliers {
- org.apache.flink.api.connector.sink2.Sink<Integer> sink;
- Supplier<List<String>> elementSupplier;
- Supplier<List<org.apache.flink.api.common.eventtime.Watermark>> watermarkSupplier;
- LongSupplier lastCheckpointSupplier;
- Supplier<SimpleVersionedSerializer<String>> serializerSupplier;
+ int getRecordCountFromState();
- public SinkAndSuppliers(
- org.apache.flink.api.connector.sink2.Sink<Integer> sink,
- Supplier<List<String>> elementSupplier,
- Supplier<List<org.apache.flink.api.common.eventtime.Watermark>> watermarkSupplier,
- LongSupplier lastCheckpointSupplier,
- Supplier<SimpleVersionedSerializer<String>> serializerSupplier) {
+ Sink<Integer> getSink();
+ }
+
+ abstract static class AbstractInspectableSink<
+ S extends org.apache.flink.api.connector.sink2.Sink<Integer>>
+ implements InspectableSink {
+ private final S sink;
+
+ protected AbstractInspectableSink(S sink) {
this.sink = sink;
- this.elementSupplier = elementSupplier;
- this.watermarkSupplier = watermarkSupplier;
- this.lastCheckpointSupplier = lastCheckpointSupplier;
- this.serializerSupplier = serializerSupplier;
+ }
+
+ @Override
+ public S getSink() {
+ return sink;
}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSink.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSink.java
index a062181cc88..22b988a00db 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSink.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSink.java
@@ -25,7 +25,7 @@ import org.apache.flink.api.connector.sink.Sink;
import org.apache.flink.api.connector.sink.SinkWriter;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializerAdapter;
import org.apache.flink.streaming.api.transformations.SinkV1Adapter;
import javax.annotation.Nullable;
@@ -56,13 +56,17 @@ import static org.assertj.core.api.Assertions.assertThat;
* @deprecated Use {@link TestSinkV2} instead.
*/
@Deprecated
-public class TestSink<T> implements Sink<T, String, String, String> {
+public class TestSink<T> implements Sink<T, String, Integer, String> {
+ public static final SimpleVersionedSerializerAdapter<String> COMMITTABLE_SERIALIZER =
+ TestSinkV2.COMMITTABLE_SERIALIZER;
+ public static final SimpleVersionedSerializerAdapter<Integer> WRITER_SERIALIZER =
+ TestSinkV2.WRITER_SERIALIZER;
public static final String END_OF_INPUT_STR = "end of input";
private final DefaultSinkWriter<T> writer;
- @Nullable private final SimpleVersionedSerializer<String> writerStateSerializer;
+ @Nullable private final SimpleVersionedSerializer<Integer> writerStateSerializer;
@Nullable private final Committer<String> committer;
@@ -76,7 +80,7 @@ public class TestSink<T> implements Sink<T, String, String, String> {
private TestSink(
DefaultSinkWriter<T> writer,
- @Nullable SimpleVersionedSerializer<String> writerStateSerializer,
+ @Nullable SimpleVersionedSerializer<Integer> writerStateSerializer,
@Nullable Committer<String> committer,
@Nullable SimpleVersionedSerializer<String> committableSerializer,
@Nullable GlobalCommitter<String, String> globalCommitter,
@@ -92,7 +96,7 @@ public class TestSink<T> implements Sink<T, String, String, String> {
}
@Override
- public SinkWriter<T, String, String> createWriter(InitContext context, List<String> states) {
+ public SinkWriter<T, String, Integer> createWriter(InitContext context, List<Integer> states) {
writer.init(context);
writer.restoredFrom(states);
writer.setProcessingTimerService(context.getProcessingTimeService());
@@ -120,7 +124,7 @@ public class TestSink<T> implements Sink<T, String, String, String> {
}
@Override
- public Optional<SimpleVersionedSerializer<String>> getWriterStateSerializer() {
+ public Optional<SimpleVersionedSerializer<Integer>> getWriterStateSerializer() {
return Optional.ofNullable(writerStateSerializer);
}
@@ -137,30 +141,30 @@ public class TestSink<T> implements Sink<T, String, String, String> {
return SinkV1Adapter.wrap(this);
}
+ public DefaultSinkWriter<T> getWriter() {
+ return writer;
+ }
+
/** A builder class for {@link TestSink}. */
public static class Builder<T> {
- private DefaultSinkWriter writer = new DefaultSinkWriter();
+ private DefaultSinkWriter<T> writer = new DefaultSinkWriter<>();
- private SimpleVersionedSerializer<String> writerStateSerializer;
+ private SimpleVersionedSerializer<Integer> writerStateSerializer;
private Committer<String> committer;
- private SimpleVersionedSerializer<String> committableSerializer;
-
private GlobalCommitter<String, String> globalCommitter;
- private SimpleVersionedSerializer<String> globalCommittableSerializer;
-
private Collection<String> compatibleStateNames = Collections.emptyList();
public <W> Builder<W> setWriter(DefaultSinkWriter<W> writer) {
- this.writer = checkNotNull(writer);
+ this.writer = (DefaultSinkWriter<T>) checkNotNull(writer);
return (Builder<W>) this;
}
public Builder<T> withWriterState() {
- this.writerStateSerializer = StringCommittableSerializer.INSTANCE;
+ this.writerStateSerializer = WRITER_SERIALIZER;
return this;
}
@@ -169,39 +173,23 @@ public class TestSink<T> implements Sink<T, String, String, String> {
return this;
}
- public Builder<T> setCommittableSerializer(
- SimpleVersionedSerializer<String> committableSerializer) {
- this.committableSerializer = committableSerializer;
- return this;
- }
-
public Builder<T> setDefaultCommitter() {
this.committer = new DefaultCommitter();
- this.committableSerializer = StringCommittableSerializer.INSTANCE;
return this;
}
public Builder<T> setDefaultCommitter(Supplier<Queue<String>> queueSupplier) {
this.committer = new DefaultCommitter(queueSupplier);
- this.committableSerializer = StringCommittableSerializer.INSTANCE;
- return this;
- }
-
- public Builder<T> setGlobalCommittableSerializer(
- SimpleVersionedSerializer<String> globalCommittableSerializer) {
- this.globalCommittableSerializer = globalCommittableSerializer;
return this;
}
public Builder<T> setDefaultGlobalCommitter() {
this.globalCommitter = new DefaultGlobalCommitter("");
- this.globalCommittableSerializer = StringCommittableSerializer.INSTANCE;
return this;
}
public Builder<T> setGlobalCommitter(Supplier<Queue<String>> queueSupplier) {
this.globalCommitter = new DefaultGlobalCommitter(queueSupplier);
- this.globalCommittableSerializer = StringCommittableSerializer.INSTANCE;
return this;
}
@@ -219,9 +207,9 @@ public class TestSink<T> implements Sink<T, String, String, String> {
writer,
writerStateSerializer,
committer,
- committableSerializer,
+ committer == null && globalCommitter == null ? null : COMMITTABLE_SERIALIZER,
globalCommitter,
- globalCommittableSerializer,
+ globalCommitter == null ? null : COMMITTABLE_SERIALIZER,
compatibleStateNames);
}
}
@@ -230,7 +218,7 @@ public class TestSink<T> implements Sink<T, String, String, String> {
/** Base class for out testing {@link SinkWriter Writers}. */
public static class DefaultSinkWriter<T>
- implements SinkWriter<T, String, String>, Serializable {
+ implements SinkWriter<T, String, Integer>, Serializable {
protected List<String> elements;
@@ -238,6 +226,10 @@ public class TestSink<T> implements Sink<T, String, String, String> {
protected ProcessingTimeService processingTimerService;
+ private int recordCount;
+
+ protected long lastCheckpointId = -1;
+
protected DefaultSinkWriter() {
this.elements = new ArrayList<>();
this.watermarks = new ArrayList<>();
@@ -247,6 +239,7 @@ public class TestSink<T> implements Sink<T, String, String, String> {
public void write(T element, Context context) {
elements.add(
Tuple3.of(element, context.timestamp(), context.currentWatermark()).toString());
+ recordCount++;
}
@Override
@@ -262,20 +255,27 @@ public class TestSink<T> implements Sink<T, String, String, String> {
}
@Override
- public List<String> snapshotState(long checkpointId) throws IOException {
- return Collections.emptyList();
+ public List<Integer> snapshotState(long checkpointId) throws IOException {
+ lastCheckpointId = checkpointId;
+ return Collections.singletonList(recordCount);
}
@Override
public void close() throws Exception {}
- void restoredFrom(List<String> states) {}
+ void restoredFrom(List<Integer> states) {
+ recordCount = states.isEmpty() ? 0 : states.get(0);
+ }
void setProcessingTimerService(ProcessingTimeService processingTimerService) {
this.processingTimerService = processingTimerService;
}
public void init(InitContext context) {}
+
+ public int getRecordCount() {
+ return recordCount;
+ }
}
// -------------------------------------- Sink Committer ---------------------------------------
@@ -393,30 +393,4 @@ public class TestSink<T> implements Sink<T, String, String, String> {
commit(Collections.singletonList(END_OF_INPUT_STR));
}
}
-
- /**
- * We introduce this {@link StringCommittableSerializer} is because that all the fields of
- * {@link TestSink} should be serializable.
- */
- public static class StringCommittableSerializer
- implements SimpleVersionedSerializer<String>, Serializable {
-
- public static final StringCommittableSerializer INSTANCE =
- new StringCommittableSerializer();
-
- @Override
- public int getVersion() {
- return SimpleVersionedStringSerializer.INSTANCE.getVersion();
- }
-
- @Override
- public byte[] serialize(String obj) {
- return SimpleVersionedStringSerializer.INSTANCE.serialize(obj);
- }
-
- @Override
- public String deserialize(int version, byte[] serialized) throws IOException {
- return SimpleVersionedStringSerializer.INSTANCE.deserialize(version, serialized);
- }
- }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java
index e11f71c37ee..9caf7a8f339 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/TestSinkV2.java
@@ -19,6 +19,8 @@
package org.apache.flink.streaming.runtime.operators.sink;
import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.api.connector.sink2.CommitterInitContext;
import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
@@ -30,6 +32,7 @@ import org.apache.flink.api.connector.sink2.SupportsWriterState;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializerAdapter;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessageTypeInfo;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
@@ -37,7 +40,6 @@ import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
import org.apache.flink.streaming.api.connector.sink2.SupportsPostCommitTopology;
import org.apache.flink.streaming.api.connector.sink2.SupportsPreCommitTopology;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava32.com.google.common.collect.ImmutableSet;
@@ -61,6 +63,10 @@ import static org.assertj.core.api.Assertions.assertThat;
/** A {@link Sink} for all the sink related tests. */
public class TestSinkV2<InputT> implements Sink<InputT> {
+ public static final SimpleVersionedSerializerAdapter<String> COMMITTABLE_SERIALIZER =
+ new SimpleVersionedSerializerAdapter<>(StringSerializer.INSTANCE);
+ public static final SimpleVersionedSerializerAdapter<Integer> WRITER_SERIALIZER =
+ new SimpleVersionedSerializerAdapter<>(IntSerializer.INSTANCE);
private final DefaultSinkWriter<InputT> writer;
@@ -81,11 +87,18 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return new Builder<>();
}
+ public static <InputT> Builder<InputT> newBuilder(DefaultSinkWriter<InputT> writer) {
+ return new Builder<InputT>().setWriter(writer);
+ }
+
+ public SupportsCommitter<String> asSupportsCommitter() {
+ throw new UnsupportedOperationException("No committter");
+ }
+
/** A builder class for {@link TestSinkV2}. */
public static class Builder<InputT> {
private DefaultSinkWriter<InputT> writer = null;
private DefaultCommitter committer;
- private SimpleVersionedSerializer<String> committableSerializer;
private boolean withPostCommitTopology = false;
private boolean withPreCommitTopology = false;
private boolean withWriterState = false;
@@ -101,22 +114,14 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return this;
}
- public Builder<InputT> setCommittableSerializer(
- SimpleVersionedSerializer<String> committableSerializer) {
- this.committableSerializer = committableSerializer;
- return this;
- }
-
public Builder<InputT> setDefaultCommitter() {
this.committer = new DefaultCommitter();
- this.committableSerializer = StringSerializer.INSTANCE;
return this;
}
public Builder<InputT> setDefaultCommitter(
Supplier<Queue<Committer.CommitRequest<String>>> queueSupplier) {
this.committer = new DefaultCommitter(queueSupplier);
- this.committableSerializer = StringSerializer.INSTANCE;
return this;
}
@@ -155,7 +160,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
if (!withPreCommitTopology) {
// TwoPhaseCommittingSink with a stateless writer and a committer
return new TestSinkV2TwoPhaseCommittingSink<>(
- writer, committableSerializer, committer);
+ writer, COMMITTABLE_SERIALIZER, committer);
} else {
// TwoPhaseCommittingSink with a stateless writer, pre commit topology,
// committer
@@ -163,9 +168,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
writer instanceof DefaultCommittingSinkWriter,
"Please provide a DefaultCommittingSinkWriter instance");
return new TestSinkV2WithPreCommitTopology<>(
- (DefaultCommittingSinkWriter) writer,
- committableSerializer,
- committer);
+ writer, COMMITTABLE_SERIALIZER, committer);
}
} else {
if (withWriterState) {
@@ -174,9 +177,9 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
Preconditions.checkArgument(
writer instanceof DefaultStatefulSinkWriter,
"Please provide a DefaultStatefulSinkWriter instance");
- return new TestStatefulSinkV2(
- (DefaultStatefulSinkWriter) writer,
- committableSerializer,
+ return new TestStatefulSinkV2<>(
+ (DefaultStatefulSinkWriter<InputT>) writer,
+ COMMITTABLE_SERIALIZER,
committer,
compatibleStateNames);
} else {
@@ -186,9 +189,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
writer instanceof DefaultCommittingSinkWriter,
"Please provide a DefaultCommittingSinkWriter instance");
return new TestSinkV2WithPostCommitTopology<>(
- (DefaultCommittingSinkWriter) writer,
- committableSerializer,
- committer);
+ writer, COMMITTABLE_SERIALIZER, committer);
}
}
}
@@ -215,6 +216,11 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return committer;
}
+ @Override
+ public SupportsCommitter<String> asSupportsCommitter() {
+ return this;
+ }
+
@Override
public SimpleVersionedSerializer<String> getCommittableSerializer() {
return committableSerializer;
@@ -268,19 +274,19 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return withLineage.map(old -> old + "Transformed");
}
})
- .returns(CommittableMessageTypeInfo.of(StringSerializer::new));
+ .returns(CommittableMessageTypeInfo.of(() -> COMMITTABLE_SERIALIZER));
}
@Override
public SimpleVersionedSerializer<String> getWriteResultSerializer() {
- return new StringSerializer();
+ return new SimpleVersionedSerializerAdapter<>(StringSerializer.INSTANCE);
}
}
private static class TestStatefulSinkV2<InputT> extends TestSinkV2WithPostCommitTopology<InputT>
- implements SupportsWriterState<InputT, String>,
+ implements SupportsWriterState<InputT, Integer>,
SupportsWriterState.WithCompatibleState {
- private String compatibleState;
+ private final String compatibleState;
public TestStatefulSinkV2(
DefaultStatefulSinkWriter<InputT> writer,
@@ -297,8 +303,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public StatefulSinkWriter<InputT, String> restoreWriter(
- WriterInitContext context, Collection<String> recoveredState) {
+ public StatefulSinkWriter<InputT, Integer> restoreWriter(
+ WriterInitContext context, Collection<Integer> recoveredState) {
DefaultStatefulSinkWriter<InputT> statefulWriter =
(DefaultStatefulSinkWriter) getWriter();
@@ -307,8 +313,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public SimpleVersionedSerializer<String> getWriterStateSerializer() {
- return new StringSerializer();
+ public SimpleVersionedSerializer<Integer> getWriterStateSerializer() {
+ return WRITER_SERIALIZER;
}
@Override
@@ -326,6 +332,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
protected List<Watermark> watermarks;
+ public long lastCheckpointId = -1;
+
protected DefaultSinkWriter() {
this.elements = new ArrayList<>();
this.watermarks = new ArrayList<>();
@@ -380,15 +388,27 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
*/
protected static class DefaultStatefulSinkWriter<InputT>
extends DefaultCommittingSinkWriter<InputT>
- implements StatefulSinkWriter<InputT, String> {
+ implements StatefulSinkWriter<InputT, Integer> {
+ private int recordCount = 0;
@Override
- public List<String> snapshotState(long checkpointId) throws IOException {
- return elements;
+ public void write(InputT element, Context context) {
+ super.write(element, context);
+ recordCount++;
+ }
+
+ public int getRecordCount() {
+ return recordCount;
}
- protected void restore(Collection<String> recoveredState) {
- this.elements = new ArrayList<>(recoveredState);
+ @Override
+ public List<Integer> snapshotState(long checkpointId) throws IOException {
+ lastCheckpointId = checkpointId;
+ return Collections.singletonList(recordCount);
+ }
+
+ protected void restore(Collection<Integer> recoveredState) {
+ this.recordCount = recoveredState.isEmpty() ? 0 : recoveredState.iterator().next();
}
}
@@ -464,29 +484,4 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
});
}
}
-
- /**
- * We introduce this {@link StringSerializer} is because that all the fields of {@link
- * TestSinkV2} should be serializable.
- */
- public static class StringSerializer
- implements SimpleVersionedSerializer<String>, Serializable {
-
- public static final StringSerializer INSTANCE = new StringSerializer();
-
- @Override
- public int getVersion() {
- return SimpleVersionedStringSerializer.INSTANCE.getVersion();
- }
-
- @Override
- public byte[] serialize(String obj) {
- return SimpleVersionedStringSerializer.INSTANCE.serialize(obj);
- }
-
- @Override
- public String deserialize(int version, byte[] serialized) throws IOException {
- return SimpleVersionedStringSerializer.INSTANCE.deserialize(version, serialized);
- }
- }
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterCommitterOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterCommitterOperatorTest.java
index 6d4c3586f38..f817a6810af 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterCommitterOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterCommitterOperatorTest.java
@@ -33,8 +33,6 @@ class WithAdapterCommitterOperatorTest extends CommitterOperatorTestBase {
TestSink.newBuilder()
.setCommitter(committer)
.setDefaultGlobalCommitter()
- .setCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
.build()
.asV2(),
() -> committer.successfulCommits);
@@ -47,8 +45,6 @@ class WithAdapterCommitterOperatorTest extends CommitterOperatorTestBase {
TestSink.newBuilder()
.setCommitter(new TestSink.RetryOnceCommitter())
.setDefaultGlobalCommitter()
- .setCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
.build()
.asV2(),
() -> 0);
@@ -59,12 +55,7 @@ class WithAdapterCommitterOperatorTest extends CommitterOperatorTestBase {
ForwardingCommitter committer = new ForwardingCommitter();
return new SinkAndCounters(
(SupportsCommitter<String>)
- TestSink.newBuilder()
- .setCommitter(committer)
- .setCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
- .build()
- .asV2(),
+ TestSink.newBuilder().setCommitter(committer).build().asV2(),
() -> committer.successfulCommits);
}
@@ -78,6 +69,6 @@ class WithAdapterCommitterOperatorTest extends CommitterOperatorTestBase {
}
@Override
- public void close() throws Exception {}
+ public void close() {}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterSinkWriterOperatorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterSinkWriterOperatorTest.java
index 5af5ac5a679..4e320816aea 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterSinkWriterOperatorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/WithAdapterSinkWriterOperatorTest.java
@@ -18,65 +18,46 @@
package org.apache.flink.streaming.runtime.operators.sink;
+import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.connector.sink.Sink;
import org.apache.flink.api.java.tuple.Tuple3;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
class WithAdapterSinkWriterOperatorTest extends SinkWriterOperatorTestBase {
-
@Override
- SinkAndSuppliers sinkWithoutCommitter() {
+ InspectableSink sinkWithoutCommitter() {
TestSink.DefaultSinkWriter<Integer> sinkWriter = new TestSink.DefaultSinkWriter<>();
- return new SinkAndSuppliers(
- TestSink.newBuilder().setWriter(sinkWriter).build().asV2(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- () -> new TestSink.StringCommittableSerializer());
+ return new InspectableSink(TestSink.newBuilder().setWriter(sinkWriter).build());
}
@Override
- SinkAndSuppliers sinkWithCommitter() {
+ InspectableSink sinkWithCommitter() {
TestSink.DefaultSinkWriter<Integer> sinkWriter = new TestSink.DefaultSinkWriter<>();
- return new SinkAndSuppliers(
- TestSink.newBuilder().setWriter(sinkWriter).setDefaultCommitter().build().asV2(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- () -> new TestSink.StringCommittableSerializer());
+ return new InspectableSink(
+ TestSink.newBuilder().setWriter(sinkWriter).setDefaultCommitter().build());
}
@Override
- SinkAndSuppliers sinkWithTimeBasedWriter() {
+ InspectableSink sinkWithTimeBasedWriter() {
TestSink.DefaultSinkWriter<Integer> sinkWriter = new TimeBasedBufferingSinkWriter();
- return new SinkAndSuppliers(
- TestSink.newBuilder().setWriter(sinkWriter).setDefaultCommitter().build().asV2(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> -1,
- () -> new TestSink.StringCommittableSerializer());
+ return new InspectableSink(
+ TestSink.newBuilder().setWriter(sinkWriter).setDefaultCommitter().build());
}
@Override
- SinkAndSuppliers sinkWithSnapshottingWriter(boolean withState, String stateName) {
- SnapshottingBufferingSinkWriter sinkWriter = new SnapshottingBufferingSinkWriter();
+ InspectableSink sinkWithState(boolean withState, String stateName) {
+ TestSink.DefaultSinkWriter<Integer> sinkWriter = new TestSink.DefaultSinkWriter<>();
TestSink.Builder<Integer> builder =
TestSink.newBuilder().setWriter(sinkWriter).setDefaultCommitter();
if (withState) {
builder.withWriterState();
+ if (stateName != null) {
+ builder.setCompatibleStateNames(stateName);
+ }
}
- if (stateName != null) {
- builder.setCompatibleStateNames(stateName);
- }
- return new SinkAndSuppliers(
- builder.build().asV2(),
- () -> sinkWriter.elements,
- () -> sinkWriter.watermarks,
- () -> sinkWriter.lastCheckpointId,
- () -> new TestSink.StringCommittableSerializer());
+ return new InspectableSink(builder.build());
}
private static class TimeBasedBufferingSinkWriter extends TestSink.DefaultSinkWriter<Integer>
@@ -103,30 +84,33 @@ class WithAdapterSinkWriterOperatorTest extends SinkWriterOperatorTestBase {
}
}
- private static class SnapshottingBufferingSinkWriter
- extends TestSink.DefaultSinkWriter<Integer> {
- public static final int NOT_SNAPSHOTTED = -1;
- long lastCheckpointId = NOT_SNAPSHOTTED;
+ class InspectableSink
+ extends AbstractInspectableSink<org.apache.flink.api.connector.sink2.Sink<Integer>> {
+ private final TestSink<Integer> sink;
+
+ InspectableSink(TestSink<Integer> sink) {
+ super(sink.asV2());
+ this.sink = sink;
+ }
@Override
- public List<String> snapshotState(long checkpointId) {
- lastCheckpointId = checkpointId;
- return elements;
+ public long getLastCheckpointId() {
+ return sink.getWriter().lastCheckpointId;
}
@Override
- void restoredFrom(List<String> states) {
- this.elements = new ArrayList<>(states);
+ public List<String> getRecordsOfCurrentCheckpoint() {
+ return sink.getWriter().elements;
}
@Override
- public List<String> prepareCommit(boolean flush) {
- if (!flush) {
- return Collections.emptyList();
- }
- List<String> result = elements;
- elements = new ArrayList<>();
- return result;
+ public List<Watermark> getWatermarks() {
+ return sink.getWriter().watermarks;
+ }
+
+ @Override
+ public int getRecordCountFromState() {
+ return sink.getWriter().getRecordCount();
}
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/deprecated/TestSinkV2.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/deprecated/TestSinkV2.java
index 5bb25605d75..7aabbcb4e49 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/deprecated/TestSinkV2.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/deprecated/TestSinkV2.java
@@ -26,10 +26,10 @@ import org.apache.flink.api.connector.sink2.StatefulSink;
import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.io.SimpleVersionedSerializerAdapter;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.connector.sink2.WithPostCommitTopology;
import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.functions.sink.filesystem.bucketassigners.SimpleVersionedStringSerializer;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.guava32.com.google.common.collect.ImmutableSet;
@@ -60,6 +60,10 @@ import static org.assertj.core.api.Assertions.assertThat;
@Deprecated
public class TestSinkV2<InputT> implements Sink<InputT> {
+ public static final SimpleVersionedSerializerAdapter<String> COMMITTABLE_SERIALIZER =
+ org.apache.flink.streaming.runtime.operators.sink.TestSinkV2.COMMITTABLE_SERIALIZER;
+ public static final SimpleVersionedSerializerAdapter<Integer> WRITER_SERIALIZER =
+ org.apache.flink.streaming.runtime.operators.sink.TestSinkV2.WRITER_SERIALIZER;
private final DefaultSinkWriter<InputT> writer;
private TestSinkV2(DefaultSinkWriter<InputT> writer) {
@@ -71,7 +75,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return writer;
}
- DefaultSinkWriter<InputT> getWriter() {
+ public DefaultSinkWriter<InputT> getWriter() {
return writer;
}
@@ -83,7 +87,6 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
public static class Builder<InputT> {
private DefaultSinkWriter<InputT> writer = null;
private DefaultCommitter committer;
- private SimpleVersionedSerializer<String> committableSerializer;
private boolean withPostCommitTopology = false;
private boolean withWriterState = false;
private String compatibleStateNames;
@@ -98,22 +101,14 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
return this;
}
- public Builder<InputT> setCommittableSerializer(
- SimpleVersionedSerializer<String> committableSerializer) {
- this.committableSerializer = committableSerializer;
- return this;
- }
-
public Builder<InputT> setDefaultCommitter() {
this.committer = new DefaultCommitter();
- this.committableSerializer = StringSerializer.INSTANCE;
return this;
}
public Builder<InputT> setDefaultCommitter(
Supplier<Queue<Committer.CommitRequest<String>>> queueSupplier) {
this.committer = new DefaultCommitter(queueSupplier);
- this.committableSerializer = StringSerializer.INSTANCE;
return this;
}
@@ -146,7 +141,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
if (!withPostCommitTopology) {
// TwoPhaseCommittingSink with a stateless writer and a committer
return new TestSinkV2TwoPhaseCommittingSink<>(
- writer, committableSerializer, committer);
+ writer, COMMITTABLE_SERIALIZER, committer);
} else {
if (withWriterState) {
// TwoPhaseCommittingSink with a stateful writer and a committer and post
@@ -156,7 +151,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
"Please provide a DefaultStatefulSinkWriter instance");
return new TestStatefulSinkV2(
(DefaultStatefulSinkWriter) writer,
- committableSerializer,
+ COMMITTABLE_SERIALIZER,
committer,
compatibleStateNames);
} else {
@@ -167,7 +162,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
"Please provide a DefaultCommittingSinkWriter instance");
return new TestSinkV2WithPostCommitTopology<>(
(DefaultCommittingSinkWriter) writer,
- committableSerializer,
+ COMMITTABLE_SERIALIZER,
committer);
}
}
@@ -225,7 +220,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
private static class TestStatefulSinkV2<InputT> extends TestSinkV2WithPostCommitTopology<InputT>
- implements StatefulSink<InputT, String>, StatefulSink.WithCompatibleState {
+ implements StatefulSink<InputT, Integer>, StatefulSink.WithCompatibleState {
private String compatibleState;
public TestStatefulSinkV2(
@@ -243,8 +238,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public StatefulSinkWriter<InputT, String> restoreWriter(
- InitContext context, Collection<String> recoveredState) {
+ public StatefulSinkWriter<InputT, Integer> restoreWriter(
+ InitContext context, Collection<Integer> recoveredState) {
DefaultStatefulSinkWriter<InputT> statefulWriter =
(DefaultStatefulSinkWriter) getWriter();
@@ -253,8 +248,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public SimpleVersionedSerializer<String> getWriterStateSerializer() {
- return new StringSerializer();
+ public SimpleVersionedSerializer<Integer> getWriterStateSerializer() {
+ return WRITER_SERIALIZER;
}
@Override
@@ -272,6 +267,8 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
public List<Watermark> watermarks;
+ public long lastCheckpointId = -1;
+
public DefaultSinkWriter() {
this.elements = new ArrayList<>();
this.watermarks = new ArrayList<>();
@@ -327,15 +324,27 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
*/
public static class DefaultStatefulSinkWriter<InputT>
extends DefaultCommittingSinkWriter<InputT>
- implements StatefulSink.StatefulSinkWriter<InputT, String> {
+ implements StatefulSink.StatefulSinkWriter<InputT, Integer> {
+ private int recordCount;
@Override
- public List<String> snapshotState(long checkpointId) throws IOException {
- return elements;
+ public List<Integer> snapshotState(long checkpointId) throws IOException {
+ lastCheckpointId = checkpointId;
+ return Collections.singletonList(recordCount);
}
- protected void restore(Collection<String> recoveredState) {
- this.elements = new ArrayList<>(recoveredState);
+ @Override
+ public void write(InputT element, Context context) {
+ super.write(element, context);
+ recordCount++;
+ }
+
+ public int getRecordCount() {
+ return recordCount;
+ }
+
+ protected void restore(Collection<Integer> recoveredState) {
+ this.recordCount = recoveredState.isEmpty() ? 0 : recoveredState.iterator().next();
}
}
@@ -411,29 +420,4 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
});
}
}
-
- /**
- * We introduce this {@link StringSerializer} is because that all the fields of {@link
- * TestSinkV2} should be serializable.
- */
- public static class StringSerializer
- implements SimpleVersionedSerializer<String>, Serializable {
-
- public static final StringSerializer INSTANCE = new StringSerializer();
-
- @Override
- public int getVersion() {
- return SimpleVersionedStringSerializer.INSTANCE.getVersion();
- }
-
- @Override
- public byte[] serialize(String obj) {
- return SimpleVersionedStringSerializer.INSTANCE.serialize(obj);
- }
-
- @Override
- public String deserialize(int version, byte[] serialized) throws IOException {
- return SimpleVersionedStringSerializer.INSTANCE.deserialize(version, serialized);
- }
- }
}
diff --git a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
index f5b31cb66a7..662568d4afe 100644
--- a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
+++ b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSinkITCase.java
@@ -492,10 +492,7 @@ class CommonExecSinkITCase {
private static TestSink<RowData> buildRecordWriterTestSink(
TestSink.DefaultSinkWriter<RowData> writer) {
- return TestSink.newBuilder()
- .setWriter(writer)
- .setCommittableSerializer(TestSink.StringCommittableSerializer.INSTANCE)
- .build();
+ return TestSink.newBuilder().setWriter(writer).build();
}
private TableFactoryHarness.SinkBase buildRuntimeSinkProvider(
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
index bbc08c9798e..0a3de995c9a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
@@ -226,7 +226,6 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
stream.sinkTo(
TestSink.newBuilder()
- .setCommittableSerializer(TestSink.StringCommittableSerializer.INSTANCE)
.setGlobalCommitter(
(Supplier<Queue<String>> & Serializable) () -> GLOBAL_COMMIT_QUEUE)
.build());
@@ -252,8 +251,6 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
env.fromData(SOURCE_DATA)
.sinkTo(
TestSink.newBuilder()
- .setCommittableSerializer(
- TestSink.StringCommittableSerializer.INSTANCE)
.setGlobalCommitter(
(Supplier<Queue<String>> & Serializable)
() -> GLOBAL_COMMIT_QUEUE)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsDeprecatedITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsDeprecatedITCase.java
index 3dea88bcf65..95a0e4c1bec 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsDeprecatedITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsDeprecatedITCase.java
@@ -147,7 +147,6 @@ public class SinkV2MetricsDeprecatedITCase extends TestLogger {
.sinkTo(
TestSinkV2.<Long>newBuilder()
.setCommitter(new MetricCommitter(beforeLatch, afterLatch))
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
.build())
.name(TEST_SINK_NAME);
JobClient jobClient = env.executeAsync();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
index bb227552a3e..4a67d601a0c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkV2MetricsITCase.java
@@ -140,9 +140,8 @@ public class SinkV2MetricsITCase extends TestLogger {
env.fromSequence(0, numCommittables - 1)
.returns(BasicTypeInfo.LONG_TYPE_INFO)
.sinkTo(
- TestSinkV2.<Long>newBuilder()
- .setCommitter(new MetricCommitter(beforeLatch, afterLatch))
- .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
+ (TestSinkV2.<Long>newBuilder()
+ .setCommitter(new MetricCommitter(beforeLatch, afterLatch)))
.build())
.name(TEST_SINK_NAME);
JobClient jobClient = env.executeAsync();
(flink) 04/08: [FLINK-25920] Only process complete batches of committables
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1d32f1b539cd326ae9d616fb5a8ba57dca860bfd
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Wed Sep 11 10:22:29 2024 +0200
[FLINK-25920] Only process complete batches of committables
The committer is supposed to commit all committables at once for a given subtask (so that it can potentially optimize committables on the fly). With UCs, we could potentially see notifyCheckpointCompleted before receiving all committables. The CommittableSummary was built and is used to detect that.
So far, we enforced completeness only for the most current committables belonging the respective checkpoint being completed. However, we should also enforce it to all subsumed committables. In fact, we probably implicitly do it but we have the extra code path which allows subsumed committables to be incomplete. This commit simplifies the code a bit by always enforcing completeness.
---
.../connector/sink2/GlobalCommitterOperator.java | 5 ++---
.../runtime/operators/sink/CommitterOperator.java | 22 +++++++---------------
.../CheckpointCommittableManagerImpl.java | 9 ++++-----
.../sink/committables/CommittableManager.java | 8 +++-----
.../CheckpointCommittableManagerImplTest.java | 18 +++++++++++-------
5 files changed, 27 insertions(+), 35 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
index 5776ae43b00..86697081436 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
@@ -175,8 +175,7 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
sinkV1State = globalCommitter.commit(sinkV1State);
}
for (CheckpointCommittableManager<CommT> committable : getCommittables(checkpointId)) {
- boolean fullyReceived = committable.getCheckpointId() == lastCompletedCheckpointId;
- committable.commit(fullyReceived, committer);
+ committable.commit(committer);
}
}
@@ -186,7 +185,7 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
committableCollector.getEndOfInputCommittable();
if (endOfInputCommittable != null) {
do {
- endOfInputCommittable.commit(false, committer);
+ endOfInputCommittable.commit(committer);
} while (!committableCollector.isFinished());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index 484d7a712c3..e33594475e8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -149,30 +149,23 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
endInput = true;
if (!isCheckpointingEnabled || isBatchMode) {
// There will be no final checkpoint, all committables should be committed here
- notifyCheckpointComplete(EOI);
+ commitAndEmitCheckpoints();
}
}
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- if (endInput) {
- // This is the final checkpoint, all committables should be committed
- lastCompletedCheckpointId = Long.MAX_VALUE;
- } else {
- lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
- }
+ lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
}
private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
+ long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
do {
for (CheckpointCommittableManager<CommT> manager :
- committableCollector.getCheckpointCommittablesUpTo(lastCompletedCheckpointId)) {
- // wait for all committables of the current manager before submission
- boolean fullyReceived =
- !endInput && manager.getCheckpointId() == lastCompletedCheckpointId;
- commitAndEmit(manager, fullyReceived);
+ committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
+ commitAndEmit(manager);
}
// !committableCollector.isFinished() indicates that we should retry
// Retry should be done here if this is a final checkpoint (indicated by endInput)
@@ -185,10 +178,9 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
}
}
- private void commitAndEmit(CommittableManager<CommT> committableManager, boolean fullyReceived)
+ private void commitAndEmit(CommittableManager<CommT> committableManager)
throws IOException, InterruptedException {
- Collection<CommittableWithLineage<CommT>> committed =
- committableManager.commit(fullyReceived, committer);
+ Collection<CommittableWithLineage<CommT>> committed = committableManager.commit(committer);
if (emitDownstream && !committed.isEmpty()) {
output.collect(new StreamRecord<>(committableManager.getSummary()));
for (CommittableWithLineage<CommT> committable : committed) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index 09a33b72651..d98ec256e50 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -120,10 +120,9 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
}
@Override
- public Collection<CommittableWithLineage<CommT>> commit(
- boolean fullyReceived, Committer<CommT> committer)
+ public Collection<CommittableWithLineage<CommT>> commit(Committer<CommT> committer)
throws IOException, InterruptedException {
- Collection<CommitRequestImpl<CommT>> requests = getPendingRequests(fullyReceived);
+ Collection<CommitRequestImpl<CommT>> requests = getPendingRequests(true);
requests.forEach(CommitRequestImpl::setSelected);
committer.commit(new ArrayList<>(requests));
requests.forEach(CommitRequestImpl::setCommittedIfNoError);
@@ -132,9 +131,9 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
return committed;
}
- Collection<CommitRequestImpl<CommT>> getPendingRequests(boolean fullyReceived) {
+ Collection<CommitRequestImpl<CommT>> getPendingRequests(boolean onlyIfFullyReceived) {
return subtasksCommittableManagers.values().stream()
- .filter(subtask -> !fullyReceived || subtask.hasReceivedAll())
+ .filter(subtask -> !onlyIfFullyReceived || subtask.hasReceivedAll())
.flatMap(SubtaskCommittableManager::getPendingRequests)
.collect(Collectors.toList());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableManager.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableManager.java
index 9b5aa9deccb..f6f176d6748 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableManager.java
@@ -37,16 +37,14 @@ public interface CommittableManager<CommT> {
CommittableSummary<CommT> getSummary();
/**
- * Commits all due committables.
+ * Commits all due committables if all respective committables of the specific subtask and
+ * checkpoint have been received.
*
- * @param fullyReceived only commit committables if all committables of this checkpoint for a
- * subtask are received
* @param committer used to commit to the external system
* @return successfully committed committables with meta information
* @throws IOException
* @throws InterruptedException
*/
- Collection<CommittableWithLineage<CommT>> commit(
- boolean fullyReceived, Committer<CommT> committer)
+ Collection<CommittableWithLineage<CommT>> commit(Committer<CommT> committer)
throws IOException, InterruptedException;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
index 0ae88631a58..5bcebda497f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
@@ -75,17 +75,21 @@ class CheckpointCommittableManagerImplTest {
final Committer<Integer> committer = new NoOpCommitter();
// Only commit fully received committables
- Collection<CommittableWithLineage<Integer>> commitRequests =
- checkpointCommittables.commit(true, committer);
- assertThat(commitRequests)
+ assertThat(checkpointCommittables.commit(committer))
.hasSize(1)
.satisfiesExactly(c -> assertThat(c.getCommittable()).isEqualTo(3));
+ // Even on retry
+ assertThat(checkpointCommittables.commit(committer)).isEmpty();
+
+ // Add missing committable
+ checkpointCommittables.addCommittable(new CommittableWithLineage<>(5, 1L, 2));
// Commit all committables
- commitRequests = checkpointCommittables.commit(false, committer);
- assertThat(commitRequests)
- .hasSize(1)
- .satisfiesExactly(c -> assertThat(c.getCommittable()).isEqualTo(4));
+ assertThat(checkpointCommittables.commit(committer))
+ .hasSize(2)
+ .satisfiesExactly(
+ c -> assertThat(c.getCommittable()).isEqualTo(4),
+ c -> assertThat(c.getCommittable()).isEqualTo(5));
}
@Test
(flink) 02/08: [FLINK-25920] Turn CommittableManager#merge functional
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8ce679ae8476c8f2dc6d765103cf24a57b653470
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Sep 12 09:29:39 2024 +0200
[FLINK-25920] Turn CommittableManager#merge functional
Remove the side-effect and create a new (rather cheap) instance of the managers.
---
.../CheckpointCommittableManagerImpl.java | 5 ++--
.../CommittableCollectorSerializer.java | 16 ++---------
.../committables/SubtaskCommittableManager.java | 21 ++++++++------
.../SubtaskCommittableManagerTest.java | 33 +++++++++++-----------
4 files changed, 36 insertions(+), 39 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index 7740a87da42..09a33b72651 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -147,14 +147,15 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
CheckpointCommittableManagerImpl<CommT> merge(CheckpointCommittableManagerImpl<CommT> other) {
checkArgument(other.checkpointId == checkpointId);
+ CheckpointCommittableManagerImpl<CommT> merged = copy();
for (Map.Entry<Integer, SubtaskCommittableManager<CommT>> subtaskEntry :
other.subtasksCommittableManagers.entrySet()) {
- subtasksCommittableManagers.merge(
+ merged.subtasksCommittableManagers.merge(
subtaskEntry.getKey(),
subtaskEntry.getValue(),
SubtaskCommittableManager::merge);
}
- return this;
+ return merged;
}
CheckpointCommittableManagerImpl<CommT> copy() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
index 0cd59305786..7bb6d769a48 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
@@ -166,19 +166,9 @@ public final class CommittableCollectorSerializer<CommT>
for (SubtaskCommittableManager<CommT> subtaskCommittableManager :
subtaskCommittableManagers) {
- // check if we already have manager for current
- // subtaskCommittableManager.getSubtaskId() if yes,
- // then merge them.
- SubtaskCommittableManager<CommT> mergedManager =
- subtasksCommittableManagers.computeIfPresent(
- subtaskId,
- (key, manager) -> manager.merge(subtaskCommittableManager));
-
- // This is new subtaskId, lets add the mapping.
- if (mergedManager == null) {
- subtasksCommittableManagers.put(
- subtaskCommittableManager.getSubtaskId(), subtaskCommittableManager);
- }
+ // merge in case we already have a manager for that subtaskId
+ subtasksCommittableManagers.merge(
+ subtaskId, subtaskCommittableManager, SubtaskCommittableManager::merge);
}
return new CheckpointCommittableManagerImpl<>(
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
index 381cec977f3..a459930ad68 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
@@ -39,12 +39,12 @@ import static org.apache.flink.util.Preconditions.checkState;
/** Manages the committables coming from one subtask. */
class SubtaskCommittableManager<CommT> {
private final Deque<CommitRequestImpl<CommT>> requests;
- private int numExpectedCommittables;
+ private final int numExpectedCommittables;
private final long checkpointId;
private final int subtaskId;
private int numDrained;
private int numFailed;
- private SinkCommitterMetricGroup metricGroup;
+ private final SinkCommitterMetricGroup metricGroup;
SubtaskCommittableManager(
int numExpectedCommittables,
@@ -186,12 +186,17 @@ class SubtaskCommittableManager<CommT> {
}
SubtaskCommittableManager<CommT> merge(SubtaskCommittableManager<CommT> other) {
- checkArgument(other.getSubtaskId() == this.getSubtaskId());
- this.numExpectedCommittables += other.numExpectedCommittables;
- this.requests.addAll(other.requests);
- this.numDrained += other.numDrained;
- this.numFailed += other.numFailed;
- return this;
+ checkArgument(other.getSubtaskId() == this.getSubtaskId(), "Different subtasks.");
+ checkArgument(other.getCheckpointId() == this.getCheckpointId(), "Different checkpoints.");
+ return new SubtaskCommittableManager<>(
+ Stream.concat(requests.stream(), other.requests.stream())
+ .collect(Collectors.toList()),
+ numExpectedCommittables + other.numExpectedCommittables,
+ numDrained + other.numDrained,
+ numFailed + other.numFailed,
+ subtaskId,
+ checkpointId,
+ metricGroup);
}
SubtaskCommittableManager<CommT> copy() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
index 0bb4f0d3db0..dd4fdf91d6e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
@@ -111,21 +111,22 @@ class SubtaskCommittableManagerTest {
1,
2L,
METRIC_GROUP);
- subtaskCommittableManager.merge(
- new SubtaskCommittableManager<>(
- Arrays.asList(
- new CommitRequestImpl<>(2, METRIC_GROUP),
- new CommitRequestImpl<>(3, METRIC_GROUP)),
- 10,
- 2,
- 3,
- 1,
- 2L,
- METRIC_GROUP));
- assertThat(subtaskCommittableManager.getNumCommittables()).isEqualTo(11);
- assertThat(subtaskCommittableManager.getNumDrained()).isEqualTo(3);
- assertThat(subtaskCommittableManager.isFinished()).isFalse();
- assertThat(subtaskCommittableManager.getNumFailed()).isEqualTo(5);
- assertThat(subtaskCommittableManager.getPendingRequests()).hasSize(3);
+ SubtaskCommittableManager<Integer> merged =
+ subtaskCommittableManager.merge(
+ new SubtaskCommittableManager<>(
+ Arrays.asList(
+ new CommitRequestImpl<>(2, METRIC_GROUP),
+ new CommitRequestImpl<>(3, METRIC_GROUP)),
+ 10,
+ 2,
+ 3,
+ 1,
+ 2L,
+ METRIC_GROUP));
+ assertThat(merged.getNumCommittables()).isEqualTo(11);
+ assertThat(merged.getNumDrained()).isEqualTo(3);
+ assertThat(merged.isFinished()).isFalse();
+ assertThat(merged.getNumFailed()).isEqualTo(5);
+ assertThat(merged.getPendingRequests()).hasSize(3);
}
}
(flink) 08/08: [FLINK-25920] Fix AbstractStreamingWriter sending after EOI
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 6d60f41a9b1369cd9af6fe0e0fb384f39b3166ff
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Sun Sep 15 22:30:50 2024 +0200
[FLINK-25920] Fix AbstractStreamingWriter sending after EOI
AbstractStreamingWriter send partition info twice on EOI. This commit ensures that we are not resending partition information even after restarting from a final checkpoint.
---
.../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 36 +++++++-------
.../file/table/stream/AbstractStreamingWriter.java | 56 ++++++++++++++++++++--
2 files changed, 70 insertions(+), 22 deletions(-)
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
index 91d8cab4e67..b9d189c8fe6 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
@@ -560,29 +560,30 @@ Method <org.apache.flink.connector.file.table.batch.compact.BatchFileWriter.open
Method <org.apache.flink.connector.file.table.batch.compact.BatchFileWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (BatchFileWriter.java:116)
Method <org.apache.flink.connector.file.table.batch.compact.BatchFileWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<T>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (BatchFileWriter.java:0)
Method <org.apache.flink.connector.file.table.batch.compact.BatchFileWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (BatchFileWriter.java:0)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketCreated(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.getBucketId()> in (AbstractStreamingWriter.java:106)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketCreated(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.getBucketId()> in (AbstractStreamingWriter.java:125)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketCreated(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> has generic parameter type <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket<IN, java.lang.String>> with type argument depending on <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket> in (AbstractStreamingWriter.java:0)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketCreated(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket> in (AbstractStreamingWriter.java:0)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketInactive(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.getBucketId()> in (AbstractStreamingWriter.java:111)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketInactive(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.getBucketId()> in (AbstractStreamingWriter.java:130)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketInactive(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> has generic parameter type <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket<IN, java.lang.String>> with type argument depending on <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket> in (AbstractStreamingWriter.java:0)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1.bucketInactive(org.apache.flink.streaming.api.functions.sink.filesystem.Bucket)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.Bucket> in (AbstractStreamingWriter.java:0)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.close()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.close()> in (AbstractStreamingWriter.java:167)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.commitUpToCheckpoint(long)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.commitUpToCheckpoint(long)> in (AbstractStreamingWriter.java:90)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.endInput()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onProcessingTime(long)> in (AbstractStreamingWriter.java:157)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.endInput()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.snapshotState(long)> in (AbstractStreamingWriter.java:158)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.<init>(org.apache.flink.streaming.api.functions.sink.filesystem.Buckets, boolean, org.apache.flink.api.common.state.OperatorStateStore, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> in (AbstractStreamingWriter.java:122)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.setBucketLifeCycleListener(org.apache.flink.streaming.api.functions.sink.filesystem.BucketLifeCycleListener)> in (AbstractStreamingWriter.java:101)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.setFileLifeCycleListener(org.apache.flink.streaming.api.functions.sink.filesystem.FileLifeCycleListener)> in (AbstractStreamingWriter.java:115)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getProcessingTimeService()> in (AbstractStreamingWriter.java:122)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (AbstractStreamingWriter.java:98)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(java.lang.Object, long, java.lang.Long, long)> in (AbstractStreamingWriter.java:142)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getTimestamp()> in (AbstractStreamingWriter.java:145)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (AbstractStreamingWriter.java:143)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.hasTimestamp()> in (AbstractStreamingWriter.java:145)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService.getCurrentProcessingTime()> in (AbstractStreamingWriter.java:144)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.close()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.close()> in (AbstractStreamingWriter.java:213)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.commitUpToCheckpoint(long)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.commitUpToCheckpoint(long)> in (AbstractStreamingWriter.java:109)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.endInput()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.onProcessingTime(long)> in (AbstractStreamingWriter.java:202)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.endInput()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.snapshotState(long)> in (AbstractStreamingWriter.java:203)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.<init>(org.apache.flink.streaming.api.functions.sink.filesystem.Buckets, boolean, org.apache.flink.api.common.state.OperatorStateStore, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService, long)> in (AbstractStreamingWriter.java:141)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.shaded.guava32.com.google.common.collect.Lists.newArrayList(java.lang.Iterable)> in (AbstractStreamingWriter.java:164)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.setBucketLifeCycleListener(org.apache.flink.streaming.api.functions.sink.filesystem.BucketLifeCycleListener)> in (AbstractStreamingWriter.java:120)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.setFileLifeCycleListener(org.apache.flink.streaming.api.functions.sink.filesystem.FileLifeCycleListener)> in (AbstractStreamingWriter.java:134)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getProcessingTimeService()> in (AbstractStreamingWriter.java:141)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (AbstractStreamingWriter.java:117)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.onElement(java.lang.Object, long, java.lang.Long, long)> in (AbstractStreamingWriter.java:183)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getTimestamp()> in (AbstractStreamingWriter.java:186)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (AbstractStreamingWriter.java:184)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.hasTimestamp()> in (AbstractStreamingWriter.java:186)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService.getCurrentProcessingTime()> in (AbstractStreamingWriter.java:185)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<IN>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (AbstractStreamingWriter.java:0)
Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (AbstractStreamingWriter.java:0)
-Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.snapshotState(org.apache.flink.runtime.state.StateSnapshotContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.snapshotState(long)> in (AbstractStreamingWriter.java:131)
+Method <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.snapshotState(org.apache.flink.runtime.state.StateSnapshotContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSinkHelper.snapshotState(long)> in (AbstractStreamingWriter.java:171)
Method <org.apache.flink.connector.file.table.stream.PartitionCommitTrigger.create(boolean, org.apache.flink.api.common.state.OperatorStateStore, org.apache.flink.configuration.Configuration, java.lang.ClassLoader, java.util.List, org.apache.flink.streaming.runtime.tasks.ProcessingTimeService)> has parameter of type <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService> in (PartitionCommitTrigger.java:0)
Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.commitPartitions(long)> calls method <org.apache.flink.table.utils.PartitionPathUtils.extractPartitionSpecFromPath(org.apache.flink.core.fs.Path)> in (PartitionCommitter.java:167)
Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.commitPartitions(long)> calls method <org.apache.flink.table.utils.PartitionPathUtils.generatePartitionPath(java.util.LinkedHashMap)> in (PartitionCommitter.java:172)
@@ -647,6 +648,7 @@ Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCo
Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCompressors.<clinit>()> calls method <org.apache.flink.api.common.io.compression.DeflateInflaterInputStreamFactory.getInstance()> in (StandardDeCompressors.java:43)
Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCompressors.<clinit>()> calls method <org.apache.flink.api.common.io.compression.GzipInflaterInputStreamFactory.getInstance()> in (StandardDeCompressors.java:44)
Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCompressors.<clinit>()> calls method <org.apache.flink.api.common.io.compression.XZInputStreamFactory.getInstance()> in (StandardDeCompressors.java:46)
+Static Initializer <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter.<clinit>()> gets field <org.apache.flink.api.common.typeutils.base.BooleanSerializer.INSTANCE> in (AbstractStreamingWriter.java:74)
Static Initializer <org.apache.flink.connector.file.table.stream.PartitionTimeCommitTrigger.<clinit>()> calls constructor <org.apache.flink.api.common.typeutils.base.ListSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> in (PartitionTimeCommitTrigger.java:52)
Static Initializer <org.apache.flink.connector.file.table.stream.PartitionTimeCommitTrigger.<clinit>()> calls constructor <org.apache.flink.api.common.typeutils.base.MapSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer, org.apache.flink.api.common.typeutils.TypeSerializer)> in (PartitionTimeCommitTrigger.java:56)
Static Initializer <org.apache.flink.connector.file.table.stream.PartitionTimeCommitTrigger.<clinit>()> gets field <org.apache.flink.api.common.typeutils.base.LongSerializer.INSTANCE> in (PartitionTimeCommitTrigger.java:56)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
index 8a326311ce0..ca156ef890e 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java
@@ -18,6 +18,9 @@
package org.apache.flink.connector.file.table.stream;
+import org.apache.flink.api.common.state.ListState;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
@@ -33,6 +36,12 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.shaded.guava32.com.google.common.collect.Lists;
+
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
/**
* Operator for file system sink. It is a operator version of {@link StreamingFileSink}. It can send
* file and bucket information to downstream.
@@ -58,6 +67,16 @@ public abstract class AbstractStreamingWriter<IN, OUT> extends AbstractStreamOpe
protected transient long currentWatermark;
+ /**
+ * Used to remember that EOI has already happened so that we don't emit the last committables of
+ * the final checkpoints twice.
+ */
+ private static final ListStateDescriptor<Boolean> END_OF_INPUT_STATE_DESC =
+ new ListStateDescriptor<>("end_of_input_state", BooleanSerializer.INSTANCE);
+
+ private boolean endOfInput;
+ private ListState<Boolean> endOfInputState;
+
public AbstractStreamingWriter(
long bucketCheckInterval,
StreamingFileSink.BucketsBuilder<
@@ -123,6 +142,27 @@ public abstract class AbstractStreamingWriter<IN, OUT> extends AbstractStreamOpe
bucketCheckInterval);
currentWatermark = Long.MIN_VALUE;
+
+ // Figure out if we have seen end of input before and if we should anything downstream. We
+ // have the following
+ // cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink that didn't handle EOI
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we regularly handle EOI, potentially resulting in unnecessary .
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI.
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true,
+ // we recover from a final checkpoint (for this subtask) and can ignore another EOI
+ // else we have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are
+ // true and do the same as above. As soon as one entry is false, we regularly start
+ // the writer and potentially emit duplicate summaries if we indeed recovered from a
+ // final checkpoint.
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ List<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
+ endOfInput = !previousState.isEmpty() && !previousState.contains(false);
}
@Override
@@ -139,6 +179,7 @@ public abstract class AbstractStreamingWriter<IN, OUT> extends AbstractStreamOpe
@Override
public void processElement(StreamRecord<IN> element) throws Exception {
+ checkState(!endOfInput, "Received element after endOfInput: %s", element);
helper.onElement(
element.getValue(),
getProcessingTimeService().getCurrentProcessingTime(),
@@ -149,15 +190,20 @@ public abstract class AbstractStreamingWriter<IN, OUT> extends AbstractStreamOpe
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- commitUpToCheckpoint(checkpointId);
+ if (!this.endOfInput) {
+ commitUpToCheckpoint(checkpointId);
+ }
}
@Override
public void endInput() throws Exception {
- buckets.onProcessingTime(Long.MAX_VALUE);
- helper.snapshotState(Long.MAX_VALUE);
- output.emitWatermark(new Watermark(Long.MAX_VALUE));
- commitUpToCheckpoint(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ buckets.onProcessingTime(Long.MAX_VALUE);
+ helper.snapshotState(Long.MAX_VALUE);
+ output.emitWatermark(new Watermark(Long.MAX_VALUE));
+ commitUpToCheckpoint(Long.MAX_VALUE);
+ }
}
@Override
(flink) 07/08: [FLINK-25920] Improve logging in committable handling of the sink
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2cdd3f05ba81bd5be9f4d3161a6d3ab47386470a
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Sep 5 13:24:03 2024 +0200
[FLINK-25920] Improve logging in committable handling of the sink
---
.../runtime/operators/sink/SinkWriterOperator.java | 9 +++++++--
.../CheckpointCommittableManagerImpl.java | 12 ++++++++++++
.../sink/committables/SubtaskCommittableManager.java | 20 ++++++++++++++++++++
3 files changed, 39 insertions(+), 2 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
index 35d6ca6f7ad..f2b6a533bcd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
@@ -289,7 +289,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
int numberOfParallelSubtasks,
long checkpointId,
Collection<CommT> committables) {
- output.collect(
+ emit(
new StreamRecord<>(
new CommittableSummary<>(
indexOfThisSubtask,
@@ -299,13 +299,18 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
committables.size(),
0)));
for (CommT committable : committables) {
- output.collect(
+ emit(
new StreamRecord<>(
new CommittableWithLineage<>(
committable, checkpointId, indexOfThisSubtask)));
}
}
+ private void emit(StreamRecord<CommittableMessage<CommT>> message) {
+ LOG.debug("Sending message to committer: {}", message);
+ output.collect(message);
+ }
+
private WriterInitContext createInitContext(OptionalLong restoredCheckpointId) {
return new InitContextImpl(
getRuntimeContext(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index a217116055c..9cbe0b16eaa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -24,6 +24,9 @@ import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
@@ -43,6 +46,9 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
private final int numberOfSubtasks;
private final SinkCommitterMetricGroup metricGroup;
+ private static final Logger LOG =
+ LoggerFactory.getLogger(CheckpointCommittableManagerImpl.class);
+
CheckpointCommittableManagerImpl(
int subtaskId,
int numberOfSubtasks,
@@ -82,6 +88,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
SubtaskCommittableManager<CommT> merged =
subtasksCommittableManagers.merge(
summary.getSubtaskId(), manager, SubtaskCommittableManager::merge);
+ LOG.debug("Adding EOI summary (new={}}, merged={}}).", manager, merged);
} else {
SubtaskCommittableManager<CommT> existing =
subtasksCommittableManagers.putIfAbsent(summary.getSubtaskId(), manager);
@@ -90,6 +97,11 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
String.format(
"Received duplicate committable summary for checkpoint %s + subtask %s (new=%s, old=%s). Please check the status of FLINK-25920",
checkpointId, summary.getSubtaskId(), manager, existing));
+ } else {
+ LOG.debug(
+ "Setting the summary for checkpointId {} with {}",
+ this.checkpointId,
+ manager);
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
index a459930ad68..2421e6fb234 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
@@ -209,4 +209,24 @@ class SubtaskCommittableManager<CommT> {
checkpointId,
metricGroup);
}
+
+ @Override
+ public String toString() {
+ return "SubtaskCommittableManager{"
+ + "requests="
+ + requests
+ + ", numExpectedCommittables="
+ + numExpectedCommittables
+ + ", checkpointId="
+ + checkpointId
+ + ", subtaskId="
+ + subtaskId
+ + ", numDrained="
+ + numDrained
+ + ", numFailed="
+ + numFailed
+ + ", metricGroup="
+ + metricGroup
+ + '}';
+ }
}
(flink) 01/08: [FLINK-25920] Straighten EOI handling in CommittableCollector
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit c56def03b1debc1c26759d57c017ae19b3a05eea
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Sep 5 13:21:40 2024 +0200
[FLINK-25920] Straighten EOI handling in CommittableCollector
In some parts of the sink, EOI is treated as checkpointId=null and in some checkpointId=MAX. The code of CheckpointCommittableManagerImpl implies that a null is valid however the serializer actually breaks then. In practice, checkpointId=MAX is used all the time by accident.
This commit replaces the nullable checkpointIds with a primitive long EOI=MAX, so that we always use the special value instead of null. The serializer already used that value, so it actually simplifies many places and doesn't break any existing state.
---
.../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 148 ++++++++++-----------
.../sink/compactor/operator/CompactorOperator.java | 10 +-
.../operator/CompactorOperatorStateHandler.java | 10 +-
.../api/connector/sink2/CommittableMessage.java | 22 ++-
.../sink2/CommittableMessageSerializer.java | 19 +--
.../sink2/CommittableMessageTypeInfo.java | 2 +-
.../api/connector/sink2/CommittableSummary.java | 58 +++++++-
.../connector/sink2/CommittableWithLineage.java | 43 +++++-
.../runtime/operators/sink/CommitterOperator.java | 7 +-
.../runtime/operators/sink/SinkWriterOperator.java | 4 +-
.../CheckpointCommittableManagerImpl.java | 28 ++--
.../sink/committables/CommittableCollector.java | 11 +-
.../committables/SubtaskCommittableManager.java | 11 +-
.../sink2/CommittableMessageSerializerTest.java | 4 +-
.../connector/sink2/CommittableSummaryAssert.java | 10 +-
.../sink2/CommittableWithLinageAssert.java | 13 +-
.../sink2/GlobalCommitterOperatorTest.java | 9 +-
.../committables/CommittableCollectorTest.java | 6 +-
.../operators/sink/CommitterOperatorTestBase.java | 15 +--
.../operators/sink/SinkWriterOperatorTestBase.java | 11 +-
20 files changed, 245 insertions(+), 196 deletions(-)
diff --git a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
index 107dc673280..91d8cab4e67 100644
--- a/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
+++ b/flink-architecture-tests/flink-architecture-tests-production/archunit-violations/f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e
@@ -54,9 +54,9 @@ Constructor <org.apache.flink.connector.datagen.table.types.DataGeneratorMapper.
Constructor <org.apache.flink.connector.datagen.table.types.RowDataGenerator.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.util.List, float)> depends on component type <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator> in (RowDataGenerator.java:0)
Constructor <org.apache.flink.connector.datagen.table.types.RowDataGenerator.<init>([Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;, java.util.List, float)> has parameter of type <[Lorg.apache.flink.streaming.api.functions.source.datagen.DataGenerator;> in (RowDataGenerator.java:0)
Constructor <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.<init>(org.apache.flink.core.fs.Path, long, org.apache.flink.api.common.serialization.BulkWriter$Factory, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner, org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.CheckpointRollingPolicy, org.apache.flink.connector.file.sink.writer.FileWriterBucketFactory, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileCo [...]
-Constructor <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.<init>(org.apache.flink.core.fs.Path, org.apache.flink.api.common.serialization.BulkWriter$Factory, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:533)
+Constructor <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.<init>(org.apache.flink.core.fs.Path, org.apache.flink.api.common.serialization.BulkWriter$Factory, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:548)
Constructor <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.<init>(org.apache.flink.core.fs.Path, long, org.apache.flink.api.common.serialization.Encoder, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner, org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy, org.apache.flink.connector.file.sink.writer.FileWriterBucketFactory, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apac [...]
-Constructor <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.<init>(org.apache.flink.core.fs.Path, org.apache.flink.api.common.serialization.Encoder, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:355)
+Constructor <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.<init>(org.apache.flink.core.fs.Path, org.apache.flink.api.common.serialization.Encoder, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:370)
Constructor <org.apache.flink.connector.file.sink.FileSinkCommittable.<init>(java.lang.String, org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$InProgressFileRecoverable)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$InProgressFileRecoverable> in (FileSinkCommittable.java:0)
Constructor <org.apache.flink.connector.file.sink.FileSinkCommittable.<init>(java.lang.String, org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$PendingFileRecoverable)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$PendingFileRecoverable> in (FileSinkCommittable.java:0)
Constructor <org.apache.flink.connector.file.sink.FileSinkCommittable.<init>(java.lang.String, org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$PendingFileRecoverable, org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$InProgressFileRecoverable, org.apache.flink.core.fs.Path)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$InProgressFileRecoverable> in (FileSinkCommittable.java:0)
@@ -100,7 +100,7 @@ Constructor <org.apache.flink.connector.file.src.assigners.LocalityAwareSplitAss
Constructor <org.apache.flink.connector.file.src.assigners.LocalityAwareSplitAssigner.<init>(java.util.Collection)> calls constructor <org.apache.flink.metrics.SimpleCounter.<init>()> in (LocalityAwareSplitAssigner.java:81)
Constructor <org.apache.flink.connector.file.src.impl.ContinuousFileSplitEnumerator.<init>(org.apache.flink.api.connector.source.SplitEnumeratorContext, org.apache.flink.connector.file.src.enumerate.FileEnumerator, org.apache.flink.connector.file.src.assigners.FileSplitAssigner, [Lorg.apache.flink.core.fs.Path;, java.util.Collection, long)> has parameter of type <[Lorg.apache.flink.core.fs.Path;> in (ContinuousFileSplitEnumerator.java:0)
Constructor <org.apache.flink.connector.file.table.ColumnarRowIterator.<init>(org.apache.flink.table.data.columnar.ColumnarRowData, java.lang.Runnable)> has parameter of type <org.apache.flink.table.data.columnar.ColumnarRowData> in (ColumnarRowIterator.java:0)
-Constructor <org.apache.flink.connector.file.table.FileSystemOutputFormat$Builder.<init>()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.<init>(java.lang.String, java.lang.String)> in (FileSystemOutputFormat.java:216)
+Constructor <org.apache.flink.connector.file.table.FileSystemOutputFormat$Builder.<init>()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.<init>(java.lang.String, java.lang.String)> in (FileSystemOutputFormat.java:235)
Constructor <org.apache.flink.connector.file.table.FileSystemOutputFormat.<init>(org.apache.flink.connector.file.table.FileSystemFactory, org.apache.flink.connector.file.table.TableMetaStoreFactory, boolean, boolean, org.apache.flink.core.fs.Path, [Ljava.lang.String;, boolean, java.util.LinkedHashMap, org.apache.flink.connector.file.table.OutputFormatFactory, org.apache.flink.connector.file.table.PartitionComputer, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig [...]
Constructor <org.apache.flink.connector.file.table.FileSystemOutputFormat.<init>(org.apache.flink.connector.file.table.FileSystemFactory, org.apache.flink.connector.file.table.TableMetaStoreFactory, boolean, boolean, org.apache.flink.core.fs.Path, [Ljava.lang.String;, boolean, java.util.LinkedHashMap, org.apache.flink.connector.file.table.OutputFormatFactory, org.apache.flink.connector.file.table.PartitionComputer, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig [...]
Constructor <org.apache.flink.connector.file.table.PartitionTempFileManager.<init>(org.apache.flink.connector.file.table.FileSystemFactory, org.apache.flink.core.fs.Path, int, int)> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.<init>(java.lang.String, java.lang.String)> in (PartitionTempFileManager.java:71)
@@ -174,8 +174,8 @@ Field <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter.
Field <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter.writer> has type <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter> in (CompactBucketWriter.java:0)
Field <org.apache.flink.connector.print.table.PrintTableSinkFactory$RowDataPrintFunction.writer> has type <org.apache.flink.api.common.functions.util.PrintSinkOutputWriter> in (PrintTableSinkFactory.java:0)
Method <org.apache.flink.connector.base.source.hybrid.HybridSource$HybridSourceBuilder.addSource(org.apache.flink.connector.base.source.hybrid.HybridSource$SourceFactory, org.apache.flink.api.connector.source.Boundedness)> calls method <org.apache.flink.api.java.ClosureCleaner.clean(java.lang.Object, org.apache.flink.api.common.ExecutionConfig$ClosureCleanerLevel, boolean)> in (HybridSource.java:246)
-Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> calls method <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits.signalIntermediateNoMoreSplits(int)> in (HybridSourceSplitEnumerator.java:448)
-Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> checks instanceof <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits> in (HybridSourceSplitEnumerator.java:440)
+Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> calls method <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits.signalIntermediateNoMoreSplits(int)> in (HybridSourceSplitEnumerator.java:450)
+Method <org.apache.flink.connector.base.source.hybrid.HybridSourceSplitEnumerator.checkAndSignalNoMoreSplits(org.apache.flink.api.connector.source.SplitEnumeratorContext, int, int, int)> checks instanceof <org.apache.flink.api.connector.source.SupportsIntermediateNoMoreSplits> in (HybridSourceSplitEnumerator.java:442)
Method <org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.getNumAliveFetchers()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (SplitFetcherManager.java:0)
Method <org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue.take()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FutureCompletingBlockingQueue.java:0)
Method <org.apache.flink.connector.datagen.functions.FromElementsGeneratorFunction.getSerializer()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FromElementsGeneratorFunction.java:0)
@@ -183,7 +183,7 @@ Method <org.apache.flink.connector.datagen.functions.IndexLookupGeneratorFunctio
Method <org.apache.flink.connector.datagen.source.DataGeneratorSource.getGeneratorFunction()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (DataGeneratorSource.java:0)
Method <org.apache.flink.connector.datagen.source.GeneratorSourceReaderFactory.createReader(org.apache.flink.api.connector.source.SourceReaderContext)> calls constructor <org.apache.flink.api.connector.source.util.ratelimit.RateLimitedSourceReader.<init>(org.apache.flink.api.connector.source.SourceReader, org.apache.flink.api.connector.source.util.ratelimit.RateLimiter)> in (GeneratorSourceReaderFactory.java:63)
Method <org.apache.flink.connector.datagen.source.GeneratorSourceReaderFactory.createReader(org.apache.flink.api.connector.source.SourceReaderContext)> calls method <org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy.createRateLimiter(int)> in (GeneratorSourceReaderFactory.java:62)
-Method <org.apache.flink.connector.datagen.table.DataGenTableSource.createSource()> calls constructor <org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource.<init>(org.apache.flink.streaming.api.functions.source.datagen.DataGenerator, long, java.lang.Long)> in (DataGenTableSource.java:67)
+Method <org.apache.flink.connector.datagen.table.DataGenTableSource.createSource()> calls constructor <org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource.<init>(org.apache.flink.streaming.api.functions.source.datagen.DataGenerator, long, java.lang.Long)> in (DataGenTableSource.java:72)
Method <org.apache.flink.connector.datagen.table.DataGenTableSource.createSource()> has return type <org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource> in (DataGenTableSource.java:0)
Method <org.apache.flink.connector.datagen.table.DataGenTableSource.createSource()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (DataGenTableSource.java:0)
Method <org.apache.flink.connector.datagen.table.DataGeneratorContainer.getGenerator()> has return type <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator> in (DataGeneratorContainer.java:0)
@@ -217,10 +217,10 @@ Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(or
Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.TinyIntType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:192)
Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.VarBinaryType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withVarLen(boolean)> in (RandomGeneratorVisitor.java:179)
Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.VarCharType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:158)
-Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.VarCharType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withVarLen(boolean)> in (RandomGeneratorVisitor.java:172)
-Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.intGenerator(int, int)> in (RandomGeneratorVisitor.java:292)
-Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:293)
-Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.ZonedTimestampType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:331)
+Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.VarCharType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withVarLen(boolean)> in (RandomGeneratorVisitor.java:159)
+Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.intGenerator(int, int)> in (RandomGeneratorVisitor.java:287)
+Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.YearMonthIntervalType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:288)
+Method <org.apache.flink.connector.datagen.table.RandomGeneratorVisitor.visit(org.apache.flink.table.types.logical.ZonedTimestampType)> calls method <org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator.withNullRate(float)> in (RandomGeneratorVisitor.java:326)
Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor$2.next()> calls method <org.apache.flink.shaded.guava32.com.google.common.primitives.Longs.toByteArray(long)> in (SequenceGeneratorVisitor.java:221)
Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor.getSequenceBytesGenerator(long, long)> has return type <org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator> in (SequenceGeneratorVisitor.java:0)
Method <org.apache.flink.connector.datagen.table.SequenceGeneratorVisitor.getSequenceStringGenerator(long, long)> has return type <org.apache.flink.streaming.api.functions.source.datagen.SequenceGenerator> in (SequenceGeneratorVisitor.java:0)
@@ -240,37 +240,37 @@ Method <org.apache.flink.connector.datagen.table.types.RowDataGenerator.next()>
Method <org.apache.flink.connector.datagen.table.types.RowDataGenerator.open(java.lang.String, org.apache.flink.runtime.state.FunctionInitializationContext, org.apache.flink.api.common.functions.RuntimeContext)> calls method <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator.open(java.lang.String, org.apache.flink.runtime.state.FunctionInitializationContext, org.apache.flink.api.common.functions.RuntimeContext)> in (RowDataGenerator.java:54)
Method <org.apache.flink.connector.datagen.table.types.RowDataGenerator.snapshotState(org.apache.flink.runtime.state.FunctionSnapshotContext)> calls method <org.apache.flink.streaming.api.functions.source.datagen.DataGenerator.snapshotState(org.apache.flink.runtime.state.FunctionSnapshotContext)> in (RowDataGenerator.java:61)
Method <org.apache.flink.connector.file.sink.FileSink$BucketsBuilder.createBucketWriter()> has return type <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter> in (FileSink.java:0)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createBucketWriter()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.BulkBucketWriter.<init>(org.apache.flink.core.fs.RecoverableWriter, org.apache.flink.api.common.serialization.BulkWriter$Factory)> in (FileSink.java:674)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createBucketWriter()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.BulkBucketWriter.<init>(org.apache.flink.core.fs.RecoverableWriter, org.apache.flink.api.common.serialization.BulkWriter$Factory)> in (FileSink.java:698)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createBucketWriter()> has return type <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter> in (FileSink.java:0)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:638)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartPrefix()> in (FileSink.java:639)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartSuffix()> in (FileSink.java:640)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:668)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:669)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:669)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:668)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:658)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:659)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:658)
-Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:659)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:692)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:693)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:693)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:692)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:682)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:683)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:682)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:683)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.withOutputFileConfig(org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig> in (FileSink.java:0)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createBucketWriter()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.RowWiseBucketWriter.<init>(org.apache.flink.core.fs.RecoverableWriter, org.apache.flink.api.common.serialization.Encoder)> in (FileSink.java:480)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createBucketWriter()> calls constructor <org.apache.flink.streaming.api.functions.sink.filesystem.RowWiseBucketWriter.<init>(org.apache.flink.core.fs.RecoverableWriter, org.apache.flink.api.common.serialization.Encoder)> in (FileSink.java:495)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createBucketWriter()> has return type <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter> in (FileSink.java:0)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:435)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartPrefix()> in (FileSink.java:436)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartSuffix()> in (FileSink.java:437)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:474)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:475)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:475)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:474)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:464)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:465)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:464)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:465)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:489)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:490)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:490)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getCommittableSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:489)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:479)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter.getProperties()> in (FileSink.java:480)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getInProgressFileRecoverableSerializer()> in (FileSink.java:479)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.getWriterStateSerializer()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.WriterProperties.getPendingFileRecoverableSerializer()> in (FileSink.java:480)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.withOutputFileConfig(org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig> in (FileSink.java:0)
-Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:231)
-Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:243)
-Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:275)
+Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:246)
+Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:258)
+Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSink.java:290)
Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> has generic parameter type <org.apache.flink.streaming.api.datastream.DataStream<org.apache.flink.streaming.api.connector.sink2.CommittableMessage<org.apache.flink.connector.file.sink.FileSinkCommittable>>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (FileSink.java:0)
Method <org.apache.flink.connector.file.sink.FileSink.addPreCommitTopology(org.apache.flink.streaming.api.datastream.DataStream)> has generic return type <org.apache.flink.streaming.api.datastream.DataStream<org.apache.flink.streaming.api.connector.sink2.CommittableMessage<org.apache.flink.connector.file.sink.FileSinkCommittable>>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (FileSink.java:0)
Method <org.apache.flink.connector.file.sink.FileSinkCommittable.getInProgressFileToCleanup()> has return type <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter$InProgressFileRecoverable> in (FileSinkCommittable.java:0)
@@ -330,26 +330,26 @@ Method <org.apache.flink.connector.file.sink.compactor.operator.CompactService.g
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactService.getWriterType(org.apache.flink.connector.file.sink.compactor.FileCompactor)> gets field <org.apache.flink.streaming.api.functions.sink.filesystem.CompactingFileWriter$Type.RECORD_WISE> in (CompactService.java:169)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactService.getWriterType(org.apache.flink.connector.file.sink.compactor.FileCompactor)> has return type <org.apache.flink.streaming.api.functions.sink.filesystem.CompactingFileWriter$Type> in (CompactService.java:0)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactService.open()> calls method <org.apache.flink.runtime.util.Hardware.getNumberCPUCores()> in (CompactService.java:70)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserialize(int, [B)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (CompactorOperator.java:301)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserializeV1(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (CompactorOperator.java:325)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserializeV1(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readLong()> in (CompactorOperator.java:328)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserialize(int, [B)> calls constructor <org.apache.flink.core.memory.DataInputDeserializer.<init>([B)> in (CompactorOperator.java:299)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserializeV1(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readInt()> in (CompactorOperator.java:323)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserializeV1(org.apache.flink.core.memory.DataInputDeserializer)> calls method <org.apache.flink.core.memory.DataInputDeserializer.readLong()> in (CompactorOperator.java:326)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.deserializeV1(org.apache.flink.core.memory.DataInputDeserializer)> has parameter of type <org.apache.flink.core.memory.DataInputDeserializer> in (CompactorOperator.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (CompactorOperator.java:292)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls method <org.apache.flink.core.memory.DataOutputSerializer.getCopyOfBuffer()> in (CompactorOperator.java:295)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (CompactorOperator.java:293)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serializeV1(java.util.Map, org.apache.flink.core.memory.DataOutputSerializer)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (CompactorOperator.java:315)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serializeV1(java.util.Map, org.apache.flink.core.memory.DataOutputSerializer)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeLong(long)> in (CompactorOperator.java:317)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (CompactorOperator.java:290)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls method <org.apache.flink.core.memory.DataOutputSerializer.getCopyOfBuffer()> in (CompactorOperator.java:293)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serialize(java.util.Map)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (CompactorOperator.java:291)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serializeV1(java.util.Map, org.apache.flink.core.memory.DataOutputSerializer)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (CompactorOperator.java:313)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serializeV1(java.util.Map, org.apache.flink.core.memory.DataOutputSerializer)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeLong(long)> in (CompactorOperator.java:315)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator$RemainingRequestsSerializer.serializeV1(java.util.Map, org.apache.flink.core.memory.DataOutputSerializer)> has parameter of type <org.apache.flink.core.memory.DataOutputSerializer> in (CompactorOperator.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.<init>(int, int, java.lang.Long, int, int, int)> in (CompactorOperator.java:254)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, java.lang.Long, int)> in (CompactorOperator.java:262)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperator.java:256)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperator.java:263)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:250)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:251)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(java.lang.Long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:262)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.<init>(int, int, long, int, int, int)> in (CompactorOperator.java:252)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, long, int)> in (CompactorOperator.java:260)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperator.java:254)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperator.java:261)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:248)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:249)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.emitCompacted(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactorOperator.java:260)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.getAllTasksFuture()> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (CompactorOperator.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.streaming.api.operators.util.SimpleVersionedListState.<init>(org.apache.flink.api.common.state.ListState, org.apache.flink.core.io.SimpleVersionedSerializer)> in (CompactorOperator.java:187)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactorOperator.java:135)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.streaming.api.operators.util.SimpleVersionedListState.<init>(org.apache.flink.api.common.state.ListState, org.apache.flink.core.io.SimpleVersionedSerializer)> in (CompactorOperator.java:185)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactorOperator.java:133)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.connector.file.sink.compactor.operator.CompactorRequest>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactorOperator.java:0)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactorOperator.java:0)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorFactory.createStreamOperator(org.apache.flink.streaming.api.operators.StreamOperatorParameters)> calls method <org.apache.flink.streaming.api.graph.StreamConfig.getOperatorName()> in (CompactorOperatorFactory.java:84)
@@ -361,10 +361,12 @@ Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperato
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorFactory.createStreamOperator(org.apache.flink.streaming.api.operators.StreamOperatorParameters)> has generic parameter type <org.apache.flink.streaming.api.operators.StreamOperatorParameters<org.apache.flink.streaming.api.connector.sink2.CommittableMessage<org.apache.flink.connector.file.sink.FileSinkCommittable>>> with type argument depending on <org.apache.flink.streaming.api.operators.StreamOperatorParame [...]
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorFactory.createStreamOperator(org.apache.flink.streaming.api.operators.StreamOperatorParameters)> has parameter of type <org.apache.flink.streaming.api.operators.StreamOperatorParameters> in (CompactorOperatorFactory.java:0)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorFactory.createStreamOperator(org.apache.flink.streaming.api.operators.StreamOperatorParameters)> has type parameter 'T' depending on <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (CompactorOperatorFactory.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.<init>(int, int, java.lang.Long, int, int, int)> in (CompactorOperatorStateHandler.java:177)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, java.lang.Long, int)> in (CompactorOperatorStateHandler.java:184)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.<init>(int, int, long, int, int, int)> in (CompactorOperatorStateHandler.java:177)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, long, int)> in (CompactorOperatorStateHandler.java:184)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperatorStateHandler.java:177)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperatorStateHandler.java:184)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getCheckpointIdOrEOI()> in (CompactorOperatorStateHandler.java:174)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getCheckpointIdOrEOI()> in (CompactorOperatorStateHandler.java:183)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getNumberOfCommittables()> in (CompactorOperatorStateHandler.java:175)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getNumberOfFailedCommittables()> in (CompactorOperatorStateHandler.java:177)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getNumberOfPendingCommittables()> in (CompactorOperatorStateHandler.java:176)
@@ -373,11 +375,9 @@ Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperato
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableSummary.getSubtaskId()> in (CompactorOperatorStateHandler.java:184)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> has generic parameter type <org.apache.flink.streaming.api.connector.sink2.CommittableSummary<org.apache.flink.connector.file.sink.FileSinkCommittable>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableSummary> in (CompactorOperatorStateHandler.java:0)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.appendCompactingResultsToSummary(org.apache.flink.streaming.api.connector.sink2.CommittableSummary)> has parameter of type <org.apache.flink.streaming.api.connector.sink2.CommittableSummary> in (CompactorOperatorStateHandler.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.getCheckpointId(org.apache.flink.streaming.api.connector.sink2.CommittableMessage)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableMessage.getCheckpointId()> in (CompactorOperatorStateHandler.java:279)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.getCheckpointId(org.apache.flink.streaming.api.connector.sink2.CommittableMessage)> has generic parameter type <org.apache.flink.streaming.api.connector.sink2.CommittableMessage<org.apache.flink.connector.file.sink.FileSinkCommittable>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (CompactorOperatorStateHandler.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.getCheckpointId(org.apache.flink.streaming.api.connector.sink2.CommittableMessage)> has parameter of type <org.apache.flink.streaming.api.connector.sink2.CommittableMessage> in (CompactorOperatorStateHandler.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, java.lang.Long, int)> in (CompactorOperatorStateHandler.java:218)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls constructor <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.<init>(java.lang.Object, long, int)> in (CompactorOperatorStateHandler.java:218)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactorOperatorStateHandler.java:218)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.getCheckpointIdOrEOI()> in (CompactorOperatorStateHandler.java:207)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.getCommittable()> in (CompactorOperatorStateHandler.java:196)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> calls method <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage.getSubtaskId()> in (CompactorOperatorStateHandler.java:218)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandler.handleHiddenCommittable(org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage)> has generic parameter type <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage<org.apache.flink.connector.file.sink.FileSinkCommittable>> with type argument depending on <org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage> in (CompactorOperatorStateHandler.java:0)
@@ -417,7 +417,7 @@ Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequest
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequestSerializer.serialize(org.apache.flink.connector.file.sink.compactor.operator.CompactorRequest)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeInt(int)> in (CompactorRequestSerializer.java:53)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequestSerializer.serializeV1(org.apache.flink.connector.file.sink.compactor.operator.CompactorRequest, org.apache.flink.core.memory.DataOutputSerializer)> calls method <org.apache.flink.core.memory.DataOutputSerializer.writeUTF(java.lang.String)> in (CompactorRequestSerializer.java:73)
Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequestSerializer.serializeV1(org.apache.flink.connector.file.sink.compactor.operator.CompactorRequest, org.apache.flink.core.memory.DataOutputSerializer)> has parameter of type <org.apache.flink.core.memory.DataOutputSerializer> in (CompactorRequestSerializer.java:0)
-Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequestTypeInfo.createSerializer(org.apache.flink.api.common.serialization.SerializerConfig)> calls constructor <org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy.<init>(org.apache.flink.util.function.SerializableSupplier)> in (CompactorRequestTypeInfo.java:80)
+Method <org.apache.flink.connector.file.sink.compactor.operator.CompactorRequestTypeInfo.createSerializer(org.apache.flink.api.common.serialization.SerializerConfig)> calls constructor <org.apache.flink.core.io.SimpleVersionedSerializerTypeSerializerProxy.<init>(org.apache.flink.util.function.SerializableSupplier)> in (CompactorRequestTypeInfo.java:81)
Method <org.apache.flink.connector.file.sink.writer.DefaultFileWriterBucketFactory.getNewBucket(java.lang.String, org.apache.flink.core.fs.Path, org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter, org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has generic parameter type <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter<IN, java.lang.String>> with ty [...]
Method <org.apache.flink.connector.file.sink.writer.DefaultFileWriterBucketFactory.getNewBucket(java.lang.String, org.apache.flink.core.fs.Path, org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter, org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter> in (DefaultFileWriterBucketFactory [...]
Method <org.apache.flink.connector.file.sink.writer.DefaultFileWriterBucketFactory.getNewBucket(java.lang.String, org.apache.flink.core.fs.Path, org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter, org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy, org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig> in (DefaultFileWriterBucketFac [...]
@@ -513,7 +513,7 @@ Method <org.apache.flink.connector.file.table.FileInfoExtractorBulkFormat.wrapRe
Method <org.apache.flink.connector.file.table.FileInfoExtractorBulkFormat.wrapReader(org.apache.flink.connector.file.src.reader.BulkFormat$Reader, org.apache.flink.connector.file.src.FileSourceSplit)> calls method <org.apache.flink.table.utils.PartitionPathUtils.extractPartitionSpecFromPath(org.apache.flink.core.fs.Path)> in (FileInfoExtractorBulkFormat.java:140)
Method <org.apache.flink.connector.file.table.FileSystemCommitter.commitPartitionsWithFiles(java.util.Map)> calls method <org.apache.flink.table.utils.PartitionPathUtils.extractPartitionSpecFromPath(org.apache.flink.core.fs.Path)> in (FileSystemCommitter.java:146)
Method <org.apache.flink.connector.file.table.FileSystemOutputFormat$Builder.setOutputFileConfig(org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig)> has parameter of type <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig> in (FileSystemOutputFormat.java:0)
-Method <org.apache.flink.connector.file.table.FileSystemOutputFormat$Builder.setStagingPath(org.apache.flink.core.fs.Path)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FileSystemOutputFormat.java:291)
+Method <org.apache.flink.connector.file.table.FileSystemOutputFormat$Builder.setStagingPath(org.apache.flink.core.fs.Path)> is annotated with <org.apache.flink.annotation.VisibleForTesting> in (FileSystemOutputFormat.java:0)
Method <org.apache.flink.connector.file.table.FileSystemTableSink$TableBucketAssigner.getBucketId(org.apache.flink.table.data.RowData, org.apache.flink.streaming.api.functions.sink.filesystem.BucketAssigner$Context)> calls method <org.apache.flink.table.utils.PartitionPathUtils.generatePartitionPath(java.util.LinkedHashMap)> in (FileSystemTableSink.java:553)
Method <org.apache.flink.connector.file.table.FileSystemTableSink.createBatchSink(org.apache.flink.streaming.api.datastream.DataStream, org.apache.flink.table.connector.sink.DynamicTableSink$Context, int, boolean)> calls method <org.apache.flink.api.dag.Transformation.setParallelism(int, boolean)> in (FileSystemTableSink.java:208)
Method <org.apache.flink.connector.file.table.FileSystemTableSink.createBatchSink(org.apache.flink.streaming.api.datastream.DataStream, org.apache.flink.table.connector.sink.DynamicTableSink$Context, int, boolean)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSystemTableSink.java:189)
@@ -589,7 +589,7 @@ Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.commitPa
Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (PartitionCommitter.java:140)
Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.connector.file.table.stream.PartitionCommitInfo>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (PartitionCommitter.java:0)
Method <org.apache.flink.connector.file.table.stream.PartitionCommitter.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (PartitionCommitter.java:0)
-Method <org.apache.flink.connector.file.table.stream.PartitionTimeCommitPredicate.isPartitionCommittable(org.apache.flink.connector.file.table.stream.PartitionCommitPredicate$PredicateContext)> calls method <org.apache.flink.table.utils.PartitionPathUtils.extractPartitionValues(org.apache.flink.core.fs.Path)> in (PartitionTimeCommitPredicate.java:71)
+Method <org.apache.flink.connector.file.table.stream.PartitionTimeCommitPredicate.isPartitionCommittable(org.apache.flink.connector.file.table.stream.PartitionCommitPredicate$PredicateContext)> calls method <org.apache.flink.table.utils.PartitionPathUtils.extractPartitionValues(org.apache.flink.core.fs.Path)> in (PartitionTimeCommitPredicate.java:70)
Method <org.apache.flink.connector.file.table.stream.ProcTimeCommitTrigger$1.currentProcTime()> calls method <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService.getCurrentProcessingTime()> in (ProcTimeCommitTrigger.java:111)
Method <org.apache.flink.connector.file.table.stream.ProcTimeCommitTrigger.addPartition(java.lang.String)> calls method <org.apache.flink.streaming.runtime.tasks.ProcessingTimeService.getCurrentProcessingTime()> in (ProcTimeCommitTrigger.java:76)
Method <org.apache.flink.connector.file.table.stream.StreamingFileWriter.closePartFileForPartitions()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.closePartFileForBucket(java.lang.Object)> in (StreamingFileWriter.java:130)
@@ -607,33 +607,33 @@ Method <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter
Method <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter.commit()> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.closeForCommit()> in (CompactBucketWriter.java:49)
Method <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter.factory(org.apache.flink.util.function.SupplierWithException)> has generic parameter type <org.apache.flink.util.function.SupplierWithException<org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter<T, java.lang.String>, java.io.IOException>> with type argument depending on <org.apache.flink.streaming.api.functions.sink.filesystem.BucketWriter> in (CompactBucketWriter.java:0)
Method <org.apache.flink.connector.file.table.stream.compact.CompactBucketWriter.write(java.lang.Object)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.InProgressFileWriter.write(java.lang.Object, long)> in (CompactBucketWriter.java:44)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.coordinate(long, java.util.Map)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactCoordinator.java:190)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.coordinate(long, java.util.Map)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactCoordinator.java:198)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.ListSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactCoordinator.java:116)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.MapSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer, org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactCoordinator.java:116)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.coordinate(long, java.util.Map)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactCoordinator.java:192)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.coordinate(long, java.util.Map)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactCoordinator.java:200)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.ListSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactCoordinator.java:118)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.MapSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer, org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactCoordinator.java:118)
Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.<init>(java.lang.Class, org.apache.flink.api.common.serialization.SerializerConfig)> in (CompactCoordinator.java:118)
Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> gets field <org.apache.flink.api.common.typeutils.base.LongSerializer.INSTANCE> in (CompactCoordinator.java:107)
Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> gets field <org.apache.flink.api.common.typeutils.base.StringSerializer.INSTANCE> in (CompactCoordinator.java:107)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactCoordinator.java:127)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactCoordinator.java:129)
Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.connector.file.table.stream.compact.CompactMessages$CoordinatorInput>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactCoordinator.java:0)
Method <org.apache.flink.connector.file.table.stream.compact.CompactCoordinator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactCoordinator.java:0)
Method <org.apache.flink.connector.file.table.stream.compact.CompactFileWriter.commitUpToCheckpoint(long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactFileWriter.java:63)
Method <org.apache.flink.connector.file.table.stream.compact.CompactFileWriter.commitUpToCheckpoint(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactFileWriter.java:62)
Method <org.apache.flink.connector.file.table.stream.compact.CompactFileWriter.commitUpToCheckpoint(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactFileWriter.java:63)
Method <org.apache.flink.connector.file.table.stream.compact.CompactFileWriter.onPartFileOpened(java.lang.String, org.apache.flink.core.fs.Path)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactFileWriter.java:52)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactOperator.java:162)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:160)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:161)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.ListSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactOperator.java:109)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.MapSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer, org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactOperator.java:109)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls constructor <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.<init>(java.lang.Object)> in (CompactOperator.java:164)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:162)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.endCompaction(long)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:163)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.ListSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactOperator.java:111)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.common.typeutils.base.MapSerializer.<init>(org.apache.flink.api.common.typeutils.TypeSerializer, org.apache.flink.api.common.typeutils.TypeSerializer)> in (CompactOperator.java:111)
Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> calls constructor <org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.<init>(java.lang.Class, org.apache.flink.api.common.serialization.SerializerConfig)> in (CompactOperator.java:111)
Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.initializeState(org.apache.flink.runtime.state.StateInitializationContext)> gets field <org.apache.flink.api.common.typeutils.base.LongSerializer.INSTANCE> in (CompactOperator.java:103)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.runtime.execution.Environment.getTaskManagerInfo()> in (CompactOperator.java:139)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo.getConfiguration()> in (CompactOperator.java:140)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:125)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:126)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactOperator.java:121)
-Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.tasks.StreamTask.getEnvironment()> in (CompactOperator.java:138)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.runtime.execution.Environment.getTaskManagerInfo()> in (CompactOperator.java:141)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.runtime.taskmanager.TaskManagerRuntimeInfo.getConfiguration()> in (CompactOperator.java:142)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:127)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (CompactOperator.java:128)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.streamrecord.StreamRecord.getValue()> in (CompactOperator.java:123)
+Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> calls method <org.apache.flink.streaming.runtime.tasks.StreamTask.getEnvironment()> in (CompactOperator.java:140)
Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has generic parameter type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord<org.apache.flink.connector.file.table.stream.compact.CompactMessages$CoordinatorOutput>> with type argument depending on <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactOperator.java:0)
Method <org.apache.flink.connector.file.table.stream.compact.CompactOperator.processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord)> has parameter of type <org.apache.flink.streaming.runtime.streamrecord.StreamRecord> in (CompactOperator.java:0)
Method <org.apache.flink.connector.print.table.PrintTableSinkFactory$RowDataPrintFunction.invoke(org.apache.flink.table.data.RowData, org.apache.flink.streaming.api.functions.sink.SinkFunction$Context)> calls method <org.apache.flink.api.common.functions.util.PrintSinkOutputWriter.write(java.lang.Object)> in (PrintTableSinkFactory.java:189)
@@ -641,7 +641,7 @@ Method <org.apache.flink.connector.print.table.PrintTableSinkFactory$RowDataPrin
Method <org.apache.flink.connector.print.table.PrintTableSinkFactory$RowDataPrintFunction.open(org.apache.flink.api.common.functions.OpenContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (PrintTableSinkFactory.java:181)
Method <org.apache.flink.connector.print.table.PrintTableSinkFactory$RowDataPrintFunction.open(org.apache.flink.api.common.functions.OpenContext)> calls method <org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getTaskInfo()> in (PrintTableSinkFactory.java:182)
Static Initializer <org.apache.flink.connector.file.sink.compactor.operator.CompactCoordinator.<clinit>()> gets field <org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.INSTANCE> in (CompactCoordinator.java:67)
-Static Initializer <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.<clinit>()> gets field <org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.INSTANCE> in (CompactorOperator.java:85)
+Static Initializer <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator.<clinit>()> gets field <org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer.INSTANCE> in (CompactorOperator.java:83)
Static Initializer <org.apache.flink.connector.file.src.FileSourceSplitSerializer.<clinit>()> calls constructor <org.apache.flink.core.memory.DataOutputSerializer.<init>(int)> in (FileSourceSplitSerializer.java:42)
Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCompressors.<clinit>()> calls method <org.apache.flink.api.common.io.compression.Bzip2InputStreamFactory.getInstance()> in (StandardDeCompressors.java:45)
Static Initializer <org.apache.flink.connector.file.src.compression.StandardDeCompressors.<clinit>()> calls method <org.apache.flink.api.common.io.compression.DeflateInflaterInputStreamFactory.getInstance()> in (StandardDeCompressors.java:43)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
index cf9f06252f5..0cb573d466f 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
@@ -46,8 +46,6 @@ import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -138,15 +136,15 @@ public class CompactorOperator
@Override
public void endInput() throws Exception {
// add collecting requests into the final snapshot
- checkpointRequests.put(Long.MAX_VALUE, collectingRequests);
+ checkpointRequests.put(CommittableMessage.EOI, collectingRequests);
collectingRequests = new ArrayList<>();
// submit all requests and wait until they are done
- submitUntil(Long.MAX_VALUE);
+ submitUntil(CommittableMessage.EOI);
assert checkpointRequests.isEmpty();
getAllTasksFuture().join();
- emitCompacted(null);
+ emitCompacted(CommittableMessage.EOI);
assert compactingRequests.isEmpty();
}
@@ -223,7 +221,7 @@ public class CompactorOperator
canSubmit.clear();
}
- private void emitCompacted(@Nullable Long checkpointId) throws Exception {
+ private void emitCompacted(long checkpointId) throws Exception {
List<FileSinkCommittable> compacted = new ArrayList<>();
Iterator<Tuple2<CompactorRequest, CompletableFuture<Iterable<FileSinkCommittable>>>> iter =
compactingRequests.iterator();
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
index 7f8108db8ff..02bcbbdcac7 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperatorStateHandler.java
@@ -171,7 +171,7 @@ public class CompactorOperatorStateHandler
new CommittableSummary<>(
summary.getSubtaskId(),
summary.getNumberOfSubtasks(),
- getCheckpointId(summary),
+ summary.getCheckpointIdOrEOI(),
summary.getNumberOfCommittables() + results.size(),
summary.getNumberOfPendingCommittables() + results.size(),
summary.getNumberOfFailedCommittables())));
@@ -180,7 +180,7 @@ public class CompactorOperatorStateHandler
new StreamRecord<>(
new CommittableWithLineage<>(
committable,
- getCheckpointId(summary),
+ summary.getCheckpointIdOrEOI(),
summary.getSubtaskId())));
}
}
@@ -204,7 +204,7 @@ public class CompactorOperatorStateHandler
// cleanup request to the next summary, since the count of pending committable
// for this checkpoint is immutable now
Iterable<FileSinkCommittable> result = submit(request).get();
- Long checkpointId = getCheckpointId(message);
+ Long checkpointId = message.getCheckpointIdOrEOI();
boolean pendingFileSent = false;
for (FileSinkCommittable c : result) {
if (c.hasPendingFile()) {
@@ -275,10 +275,6 @@ public class CompactorOperatorStateHandler
remainingRequestsState.update(Collections.singletonList(requestsMap));
}
- private Long getCheckpointId(CommittableMessage<FileSinkCommittable> message) {
- return message.getCheckpointId().isPresent() ? message.getCheckpointId().getAsLong() : null;
- }
-
private CompletableFuture<Iterable<FileSinkCommittable>> submit(CompactorRequest request) {
CompletableFuture<Iterable<FileSinkCommittable>> resultFuture = new CompletableFuture<>();
compactService.submit(request, resultFuture);
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessage.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessage.java
index 754b5e374f4..7db0c29ecc6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessage.java
@@ -22,15 +22,33 @@ import org.apache.flink.annotation.Experimental;
import java.util.OptionalLong;
-/** The message send from {@link SinkWriter} to {@link Committer}. */
+/** The message send from {@code SinkWriter} to {@code Committer}. */
@Experimental
public interface CommittableMessage<CommT> {
+ /**
+ * Special value for checkpointId for the end of input in case of batch commit or final
+ * checkpoint.
+ */
+ long EOI = Long.MAX_VALUE;
+
/** The subtask that created this committable. */
int getSubtaskId();
/**
* Returns the checkpoint id or empty if the message does not belong to a checkpoint. In that
* case, the committable was created at the end of input (e.g., in batch mode).
+ *
+ * @see #getCheckpointIdOrEOI()
+ */
+ @Deprecated
+ default OptionalLong getCheckpointId() {
+ long checkpointIdOrEOI = getCheckpointIdOrEOI();
+ return checkpointIdOrEOI == EOI ? OptionalLong.empty() : OptionalLong.of(checkpointIdOrEOI);
+ }
+
+ /**
+ * Returns the checkpoint id or EOI if this message belong to the final checkpoint or the batch
+ * commit.
*/
- OptionalLong getCheckpointId();
+ long getCheckpointIdOrEOI();
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java
index 0fac7380dd9..cf251dd2d84 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java
@@ -42,7 +42,6 @@ public class CommittableMessageSerializer<CommT>
@VisibleForTesting static final int VERSION = 1;
private static final int COMMITTABLE = 1;
private static final int SUMMARY = 2;
- private static final long EOI = Long.MAX_VALUE;
private final SimpleVersionedSerializer<CommT> committableSerializer;
@@ -64,14 +63,14 @@ public class CommittableMessageSerializer<CommT>
committableSerializer,
((CommittableWithLineage<CommT>) obj).getCommittable(),
out);
- writeCheckpointId(out, obj);
+ out.writeLong(obj.getCheckpointIdOrEOI());
out.writeInt(obj.getSubtaskId());
} else if (obj instanceof CommittableSummary) {
out.writeByte(SUMMARY);
out.writeInt(obj.getSubtaskId());
CommittableSummary<?> committableSummary = (CommittableSummary<?>) obj;
out.writeInt(committableSummary.getNumberOfSubtasks());
- writeCheckpointId(out, obj);
+ out.writeLong(obj.getCheckpointIdOrEOI());
out.writeInt(committableSummary.getNumberOfCommittables());
out.writeInt(committableSummary.getNumberOfPendingCommittables());
out.writeInt(committableSummary.getNumberOfFailedCommittables());
@@ -91,13 +90,13 @@ public class CommittableMessageSerializer<CommT>
return new CommittableWithLineage<>(
SimpleVersionedSerialization.readVersionAndDeSerialize(
committableSerializer, in),
- readCheckpointId(in),
+ in.readLong(),
in.readInt());
case SUMMARY:
return new CommittableSummary<>(
in.readInt(),
in.readInt(),
- readCheckpointId(in),
+ in.readLong(),
in.readInt(),
in.readInt(),
in.readInt());
@@ -109,14 +108,4 @@ public class CommittableMessageSerializer<CommT>
+ StringUtils.byteToHexString(serialized));
}
}
-
- private void writeCheckpointId(DataOutputSerializer out, CommittableMessage<CommT> obj)
- throws IOException {
- out.writeLong(obj.getCheckpointId().orElse(EOI));
- }
-
- private Long readCheckpointId(DataInputDeserializer in) throws IOException {
- long checkpointId = in.readLong();
- return checkpointId == EOI ? null : checkpointId;
- }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageTypeInfo.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageTypeInfo.java
index ad8460ee9eb..c22868a7cf7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageTypeInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageTypeInfo.java
@@ -50,7 +50,6 @@ public class CommittableMessageTypeInfo<CommT> extends TypeInformation<Committab
* @param committableSerializerFactory factory to create the serializer for a {@link
* CommittableMessage}
* @param <CommT> type of the committable
- * @return
*/
public static <CommT> TypeInformation<CommittableMessage<CommT>> of(
SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializerFactory) {
@@ -86,6 +85,7 @@ public class CommittableMessageTypeInfo<CommT> extends TypeInformation<Committab
return 1;
}
+ @SuppressWarnings({"unchecked", "rawtypes"})
@Override
public Class<CommittableMessage<CommT>> getTypeClass() {
return (Class) CommittableMessage.class;
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummary.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummary.java
index 7171a5168a7..7f1fb003806 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummary.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummary.java
@@ -20,9 +20,7 @@ package org.apache.flink.streaming.api.connector.sink2;
import org.apache.flink.annotation.Experimental;
-import javax.annotation.Nullable;
-
-import java.util.OptionalLong;
+import java.util.Objects;
/**
* This class tracks the information about committables belonging to one checkpoint coming from one
@@ -38,7 +36,7 @@ public class CommittableSummary<CommT> implements CommittableMessage<CommT> {
/** May change after recovery. */
private final int numberOfSubtasks;
- @Nullable private final Long checkpointId;
+ private final long checkpointId;
/** The number of committables coming from the given subtask in the particular checkpoint. */
private final int numberOfCommittables;
/** The number of committables that have not been successfully committed. */
@@ -49,7 +47,7 @@ public class CommittableSummary<CommT> implements CommittableMessage<CommT> {
public CommittableSummary(
int subtaskId,
int numberOfSubtasks,
- @Nullable Long checkpointId,
+ long checkpointId,
int numberOfCommittables,
int numberOfPendingCommittables,
int numberOfFailedCommittables) {
@@ -69,8 +67,8 @@ public class CommittableSummary<CommT> implements CommittableMessage<CommT> {
return numberOfSubtasks;
}
- public OptionalLong getCheckpointId() {
- return checkpointId == null ? OptionalLong.empty() : OptionalLong.of(checkpointId);
+ public long getCheckpointIdOrEOI() {
+ return checkpointId;
}
public int getNumberOfCommittables() {
@@ -94,4 +92,50 @@ public class CommittableSummary<CommT> implements CommittableMessage<CommT> {
numberOfPendingCommittables,
numberOfFailedCommittables);
}
+
+ @Override
+ public String toString() {
+ return "CommittableSummary{"
+ + "subtaskId="
+ + subtaskId
+ + ", numberOfSubtasks="
+ + numberOfSubtasks
+ + ", checkpointId="
+ + checkpointId
+ + ", numberOfCommittables="
+ + numberOfCommittables
+ + ", numberOfPendingCommittables="
+ + numberOfPendingCommittables
+ + ", numberOfFailedCommittables="
+ + numberOfFailedCommittables
+ + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CommittableSummary<?> that = (CommittableSummary<?>) o;
+ return subtaskId == that.subtaskId
+ && numberOfSubtasks == that.numberOfSubtasks
+ && checkpointId == that.checkpointId
+ && numberOfCommittables == that.numberOfCommittables
+ && numberOfPendingCommittables == that.numberOfPendingCommittables
+ && numberOfFailedCommittables == that.numberOfFailedCommittables;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ subtaskId,
+ numberOfSubtasks,
+ checkpointId,
+ numberOfCommittables,
+ numberOfPendingCommittables,
+ numberOfFailedCommittables);
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLineage.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLineage.java
index 96683bfd2ca..bdfe9de262f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLineage.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLineage.java
@@ -21,9 +21,7 @@ package org.apache.flink.streaming.api.connector.sink2;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.api.connector.sink2.Committer;
-import javax.annotation.Nullable;
-
-import java.util.OptionalLong;
+import java.util.Objects;
import java.util.function.Function;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -35,10 +33,10 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
@Experimental
public class CommittableWithLineage<CommT> implements CommittableMessage<CommT> {
private final CommT committable;
- @Nullable private final Long checkpointId;
+ private final long checkpointId;
private final int subtaskId;
- public CommittableWithLineage(CommT committable, @Nullable Long checkpointId, int subtaskId) {
+ public CommittableWithLineage(CommT committable, long checkpointId, int subtaskId) {
this.committable = checkNotNull(committable);
this.checkpointId = checkpointId;
this.subtaskId = subtaskId;
@@ -52,11 +50,42 @@ public class CommittableWithLineage<CommT> implements CommittableMessage<CommT>
return subtaskId;
}
- public OptionalLong getCheckpointId() {
- return checkpointId == null ? OptionalLong.empty() : OptionalLong.of(checkpointId);
+ public long getCheckpointIdOrEOI() {
+ return checkpointId;
}
public <NewCommT> CommittableWithLineage<NewCommT> map(Function<CommT, NewCommT> mapper) {
return new CommittableWithLineage<>(mapper.apply(committable), checkpointId, subtaskId);
}
+
+ @Override
+ public String toString() {
+ return "CommittableWithLineage{"
+ + "committable="
+ + committable
+ + ", checkpointId="
+ + checkpointId
+ + ", subtaskId="
+ + subtaskId
+ + '}';
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ CommittableWithLineage<?> that = (CommittableWithLineage<?>) o;
+ return checkpointId == that.checkpointId
+ && subtaskId == that.subtaskId
+ && Objects.equals(committable, that.committable);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(committable, checkpointId, subtaskId);
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index 028d8317d80..484d7a712c3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -50,6 +50,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.OptionalLong;
+import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -148,7 +149,7 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
endInput = true;
if (!isCheckpointingEnabled || isBatchMode) {
// There will be no final checkpoint, all committables should be committed here
- notifyCheckpointComplete(Long.MAX_VALUE);
+ notifyCheckpointComplete(EOI);
}
}
@@ -208,8 +209,8 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
// in case of unaligned checkpoint, we may receive notifyCheckpointComplete before the
// committables
- OptionalLong checkpointId = element.getValue().getCheckpointId();
- if (checkpointId.isPresent() && checkpointId.getAsLong() <= lastCompletedCheckpointId) {
+ long checkpointId = element.getValue().getCheckpointIdOrEOI();
+ if (checkpointId <= lastCompletedCheckpointId) {
commitAndEmitCheckpoints();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
index ea9a60d6133..93f5ba66198 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
@@ -197,10 +197,10 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
public void endInput() throws Exception {
endOfInput = true;
sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ emitCommittables(CommittableMessage.EOI);
}
- private void emitCommittables(Long checkpointId) throws IOException, InterruptedException {
+ private void emitCommittables(long checkpointId) throws IOException, InterruptedException {
if (!emitDownstream) {
// To support SinkV1 topologies with only a writer we have to call prepareCommit
// although no committables are forwarded
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index 38abd2d3a33..7740a87da42 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -23,14 +23,11 @@ import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
-import java.util.Objects;
import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -40,7 +37,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
/** Mapping of subtask id to {@link SubtaskCommittableManager}. */
private final Map<Integer, SubtaskCommittableManager<CommT>> subtasksCommittableManagers;
- @Nullable private final Long checkpointId;
+ private final long checkpointId;
private final int subtaskId;
private final int numberOfSubtasks;
private final SinkCommitterMetricGroup metricGroup;
@@ -48,7 +45,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
CheckpointCommittableManagerImpl(
int subtaskId,
int numberOfSubtasks,
- @Nullable Long checkpointId,
+ long checkpointId,
SinkCommitterMetricGroup metricGroup) {
this(new HashMap<>(), subtaskId, numberOfSubtasks, checkpointId, metricGroup);
}
@@ -57,7 +54,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
Map<Integer, SubtaskCommittableManager<CommT>> subtasksCommittableManagers,
int subtaskId,
int numberOfSubtasks,
- @Nullable Long checkpointId,
+ long checkpointId,
SinkCommitterMetricGroup metricGroup) {
this.subtasksCommittableManagers = checkNotNull(subtasksCommittableManagers);
this.subtaskId = subtaskId;
@@ -68,7 +65,6 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
@Override
public long getCheckpointId() {
- checkNotNull(checkpointId);
return checkpointId;
}
@@ -77,16 +73,14 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
}
void upsertSummary(CommittableSummary<CommT> summary) {
+ SubtaskCommittableManager<CommT> manager =
+ new SubtaskCommittableManager<>(
+ summary.getNumberOfCommittables(),
+ subtaskId,
+ summary.getCheckpointIdOrEOI(),
+ metricGroup);
SubtaskCommittableManager<CommT> existing =
- subtasksCommittableManagers.putIfAbsent(
- summary.getSubtaskId(),
- new SubtaskCommittableManager<>(
- summary.getNumberOfCommittables(),
- subtaskId,
- summary.getCheckpointId().isPresent()
- ? summary.getCheckpointId().getAsLong()
- : null,
- metricGroup));
+ subtasksCommittableManagers.putIfAbsent(summary.getSubtaskId(), manager);
if (existing != null) {
throw new UnsupportedOperationException(
"Currently it is not supported to update the CommittableSummary for a checkpoint coming from the same subtask. Please check the status of FLINK-25920");
@@ -152,7 +146,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
}
CheckpointCommittableManagerImpl<CommT> merge(CheckpointCommittableManagerImpl<CommT> other) {
- checkArgument(Objects.equals(other.checkpointId, checkpointId));
+ checkArgument(other.checkpointId == checkpointId);
for (Map.Entry<Integer, SubtaskCommittableManager<CommT>> subtaskEntry :
other.subtasksCommittableManagers.entrySet()) {
subtasksCommittableManagers.merge(
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
index 1b3d5bffc81..801c8446850 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
@@ -20,7 +20,7 @@ package org.apache.flink.streaming.runtime.operators.sink.committables;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.functions.RuntimeContext;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.InitContext;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
@@ -157,8 +157,7 @@ public class CommittableCollector<CommT> {
}
/**
- * Returns {@link CheckpointCommittableManager} that is currently hold by the collector and
- * associated with the {@link CommittableCollector#EOI} checkpoint id.
+ * Returns {@link CommittableManager} belonging to the last input.
*
* @return {@link CheckpointCommittableManager}
*/
@@ -235,12 +234,12 @@ public class CommittableCollector<CommT> {
private void addSummary(CommittableSummary<CommT> summary) {
checkpointCommittables
.computeIfAbsent(
- summary.getCheckpointId().orElse(EOI),
+ summary.getCheckpointIdOrEOI(),
key ->
new CheckpointCommittableManagerImpl<>(
subtaskId,
numberOfSubtasks,
- summary.getCheckpointId().orElse(EOI),
+ summary.getCheckpointIdOrEOI(),
metricGroup))
.upsertSummary(summary);
}
@@ -252,7 +251,7 @@ public class CommittableCollector<CommT> {
private CheckpointCommittableManagerImpl<CommT> getCheckpointCommittables(
CommittableMessage<CommT> committable) {
CheckpointCommittableManagerImpl<CommT> committables =
- this.checkpointCommittables.get(committable.getCheckpointId().orElse(EOI));
+ this.checkpointCommittables.get(committable.getCheckpointIdOrEOI());
return checkNotNull(committables, "Unknown checkpoint for %s", committable);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
index 1baadeb40a8..381cec977f3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
@@ -22,8 +22,6 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
-import javax.annotation.Nullable;
-
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
@@ -42,7 +40,7 @@ import static org.apache.flink.util.Preconditions.checkState;
class SubtaskCommittableManager<CommT> {
private final Deque<CommitRequestImpl<CommT>> requests;
private int numExpectedCommittables;
- @Nullable private final Long checkpointId;
+ private final long checkpointId;
private final int subtaskId;
private int numDrained;
private int numFailed;
@@ -51,7 +49,7 @@ class SubtaskCommittableManager<CommT> {
SubtaskCommittableManager(
int numExpectedCommittables,
int subtaskId,
- @Nullable Long checkpointId,
+ long checkpointId,
SinkCommitterMetricGroup metricGroup) {
this(
Collections.emptyList(),
@@ -69,7 +67,7 @@ class SubtaskCommittableManager<CommT> {
int numDrained,
int numFailed,
int subtaskId,
- @Nullable Long checkpointId,
+ long checkpointId,
SinkCommitterMetricGroup metricGroup) {
this.checkpointId = checkpointId;
this.subtaskId = subtaskId;
@@ -179,8 +177,7 @@ class SubtaskCommittableManager<CommT> {
}
@VisibleForTesting
- @Nullable
- Long getCheckpointId() {
+ long getCheckpointId() {
return checkpointId;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializerTest.java
index b3850b41935..4c7b0d7096c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializerTest.java
@@ -40,7 +40,7 @@ class CommittableMessageSerializerTest {
assertThat(message).isInstanceOf(CommittableWithLineage.class);
final CommittableWithLineage<Integer> copy = (CommittableWithLineage<Integer>) message;
assertThat(copy.getCommittable()).isEqualTo(1);
- assertThat(copy.getCheckpointId()).isPresent().hasValue(2L);
+ assertThat(copy.getCheckpointIdOrEOI()).isEqualTo(2L);
assertThat(copy.getSubtaskId()).isEqualTo(3);
}
@@ -56,7 +56,7 @@ class CommittableMessageSerializerTest {
final CommittableSummary<Integer> copy = (CommittableSummary<Integer>) message;
assertThat(copy.getSubtaskId()).isEqualTo(1);
assertThat(copy.getNumberOfSubtasks()).isEqualTo(2);
- assertThat(copy.getCheckpointId()).isPresent().hasValue(3L);
+ assertThat(copy.getCheckpointIdOrEOI()).isEqualTo(3L);
assertThat(copy.getNumberOfCommittables()).isEqualTo(4);
assertThat(copy.getNumberOfPendingCommittables()).isEqualTo(5);
assertThat(copy.getNumberOfFailedCommittables()).isEqualTo(6);
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
index fb1062a5091..e02e838aeb3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
@@ -20,8 +20,6 @@ package org.apache.flink.streaming.api.connector.sink2;
import org.assertj.core.api.AbstractAssert;
-import javax.annotation.Nullable;
-
import static org.assertj.core.api.Assertions.assertThat;
/** Custom assertions for {@link CommittableSummary}. */
@@ -75,13 +73,9 @@ public class CommittableSummaryAssert
return this;
}
- public CommittableSummaryAssert hasCheckpointId(@Nullable Long checkpointId) {
+ public CommittableSummaryAssert hasCheckpointId(long checkpointId) {
isNotNull();
- if (checkpointId == null) {
- assertThat(actual.getCheckpointId()).isEmpty();
- } else {
- assertThat(actual.getCheckpointId()).hasValue(checkpointId);
- }
+ assertThat(actual.getCheckpointIdOrEOI()).isEqualTo(checkpointId);
return this;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
index 966b4330c19..0937c7454d9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
@@ -20,8 +20,6 @@ package org.apache.flink.streaming.api.connector.sink2;
import org.assertj.core.api.AbstractAssert;
-import javax.annotation.Nullable;
-
import static org.assertj.core.api.Assertions.assertThat;
/**
@@ -38,7 +36,8 @@ public class CommittableWithLinageAssert
public CommittableWithLinageAssert isEqualTo(CommittableWithLineage<?> committableWithLineage) {
isNotNull();
assertThat(actual.getSubtaskId()).isEqualTo(committableWithLineage.getSubtaskId());
- assertThat(actual.getCheckpointId()).isEqualTo(committableWithLineage.getCheckpointId());
+ assertThat(actual.getCheckpointIdOrEOI())
+ .isEqualTo(committableWithLineage.getCheckpointIdOrEOI());
assertThat(actual.getCommittable()).isEqualTo(committableWithLineage.getCommittable());
return this;
}
@@ -49,13 +48,9 @@ public class CommittableWithLinageAssert
return this;
}
- public CommittableWithLinageAssert hasCheckpointId(@Nullable Long checkpointId) {
+ public CommittableWithLinageAssert hasCheckpointId(long checkpointId) {
isNotNull();
- if (checkpointId == null) {
- assertThat(actual.getCheckpointId()).isEmpty();
- } else {
- assertThat(actual.getCheckpointId()).hasValue(checkpointId);
- }
+ assertThat(actual.getCheckpointIdOrEOI()).isEqualTo(checkpointId);
return this;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperatorTest.java
index 441a2b437a0..dc45e939b1c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperatorTest.java
@@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.assertj.core.api.Assertions.assertThat;
class GlobalCommitterOperatorTest {
@@ -97,15 +98,15 @@ class GlobalCommitterOperatorTest {
testHarness.open();
final CommittableSummary<Integer> committableSummary =
- new CommittableSummary<>(1, 2, null, 1, 1, 0);
+ new CommittableSummary<>(1, 2, EOI, 1, 1, 0);
testHarness.processElement(new StreamRecord<>(committableSummary));
final CommittableSummary<Integer> committableSummary2 =
- new CommittableSummary<>(2, 2, null, 1, 1, 0);
+ new CommittableSummary<>(2, 2, EOI, 1, 1, 0);
testHarness.processElement(new StreamRecord<>(committableSummary2));
- final CommittableWithLineage<Integer> first = new CommittableWithLineage<>(1, null, 1);
+ final CommittableWithLineage<Integer> first = new CommittableWithLineage<>(1, EOI, 1);
testHarness.processElement(new StreamRecord<>(first));
- final CommittableWithLineage<Integer> second = new CommittableWithLineage<>(2, null, 2);
+ final CommittableWithLineage<Integer> second = new CommittableWithLineage<>(2, EOI, 2);
testHarness.processElement(new StreamRecord<>(second));
testHarness.endInput();
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
index 5af5efed1a9..3571e60f2c9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.streaming.api.connector.sink2.SinkV2Assertions;
import org.junit.jupiter.api.Test;
+import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.assertj.core.api.Assertions.assertThat;
class CommittableCollectorTest {
@@ -50,13 +51,12 @@ class CommittableCollectorTest {
void testGetEndOfInputCommittable() {
final CommittableCollector<Integer> committableCollector =
new CommittableCollector<>(1, 1, METRIC_GROUP);
- CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, null, 1, 0, 0);
+ CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, EOI, 1, 0, 0);
committableCollector.addMessage(first);
CommittableManager<Integer> endOfInputCommittable =
committableCollector.getEndOfInputCommittable();
assertThat(endOfInputCommittable).isNotNull();
- SinkV2Assertions.assertThat(endOfInputCommittable.getSummary())
- .hasCheckpointId(Long.MAX_VALUE);
+ SinkV2Assertions.assertThat(endOfInputCommittable.getSummary()).hasCheckpointId(EOI);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java
index d1269ed89c1..6e7674b47a9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorTestBase.java
@@ -35,6 +35,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.List;
import java.util.function.IntSupplier;
+import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.apache.flink.streaming.runtime.operators.sink.SinkTestUtil.fromOutput;
import static org.apache.flink.streaming.runtime.operators.sink.SinkTestUtil.toCommittableSummary;
import static org.apache.flink.streaming.runtime.operators.sink.SinkTestUtil.toCommittableWithLinage;
@@ -171,15 +172,15 @@ abstract class CommitterOperatorTestBase {
testHarness.open();
final CommittableSummary<String> committableSummary =
- new CommittableSummary<>(1, 2, null, 1, 1, 0);
+ new CommittableSummary<>(1, 2, EOI, 1, 1, 0);
testHarness.processElement(new StreamRecord<>(committableSummary));
final CommittableSummary<String> committableSummary2 =
- new CommittableSummary<>(2, 2, null, 1, 1, 0);
+ new CommittableSummary<>(2, 2, EOI, 1, 1, 0);
testHarness.processElement(new StreamRecord<>(committableSummary2));
- final CommittableWithLineage<String> first = new CommittableWithLineage<>("1", null, 1);
+ final CommittableWithLineage<String> first = new CommittableWithLineage<>("1", EOI, 1);
testHarness.processElement(new StreamRecord<>(first));
- final CommittableWithLineage<String> second = new CommittableWithLineage<>("1", null, 2);
+ final CommittableWithLineage<String> second = new CommittableWithLineage<>("1", EOI, 2);
testHarness.processElement(new StreamRecord<>(second));
testHarness.endInput();
@@ -330,11 +331,7 @@ abstract class CommitterOperatorTestBase {
CommittableWithLineage<?> copyCommittableWithDifferentOrigin(
CommittableWithLineage<?> committable, int subtaskId) {
return new CommittableWithLineage<>(
- committable.getCommittable(),
- committable.getCheckpointId().isPresent()
- ? committable.getCheckpointId().getAsLong()
- : null,
- subtaskId);
+ committable.getCommittable(), committable.getCheckpointIdOrEOI(), subtaskId);
}
private OneInputStreamOperatorTestHarness<
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
index ecd7a3e54f3..4ede8a6ee83 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperatorTestBase.java
@@ -54,8 +54,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
-import javax.annotation.Nullable;
-
import java.io.IOException;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
@@ -71,6 +69,7 @@ import java.util.function.Consumer;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
+import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.apache.flink.streaming.runtime.operators.sink.SinkTestUtil.fromOutput;
import static org.assertj.core.api.Assertions.assertThat;
@@ -178,7 +177,7 @@ abstract class SinkWriterOperatorTestBase {
testHarness.processElement(1, 1);
testHarness.endInput();
- assertBasicOutput(testHarness.getOutput(), 1, Long.MAX_VALUE);
+ assertBasicOutput(testHarness.getOutput(), 1, EOI);
}
@ParameterizedTest
@@ -222,7 +221,7 @@ abstract class SinkWriterOperatorTestBase {
restoredTestHarness.notifyOfCompletedCheckpoint(checkpointId);
if (stateful) {
- assertBasicOutput(restoredTestHarness.getOutput(), 2, Long.MAX_VALUE);
+ assertBasicOutput(restoredTestHarness.getOutput(), 2, EOI);
} else {
assertThat(fromOutput(restoredTestHarness.getOutput()).get(0).asRecord().getValue())
.isInstanceOf(CommittableSummary.class)
@@ -565,9 +564,7 @@ abstract class SinkWriterOperatorTestBase {
}
private static void assertBasicOutput(
- Collection<Object> queuedOutput,
- int numberOfCommittables,
- @Nullable Long checkpointId) {
+ Collection<Object> queuedOutput, int numberOfCommittables, long checkpointId) {
List<StreamElement> output = fromOutput(queuedOutput);
assertThat(output).hasSize(numberOfCommittables + 1);
assertThat(output.get(0).asRecord().getValue())
(flink) 06/08: [FLINK-25920] Handle duplicate EOI in Sink
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 37e6724813bd92f8d323ddec80308241f693a5e1
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Sep 5 15:25:12 2024 +0200
[FLINK-25920] Handle duplicate EOI in Sink
In case of a failure after final checkpoint, EOI is called twice.
SinkWriter should ignore the second call to avoid emitting more dummy committables = transactional objects containing no data since no data can arrive when recovering from final checkpoint. The commit uses a boolean list state to remember if EOI has been emitted. The cases are discussed in code.
Since rescaling may still result in these dummy committables, the committer needs merge them into the CommittableCollector as these committables still need to be committed as systems like Kafka don't provide transactional isolation.
---
.../runtime/operators/sink/SinkWriterOperator.java | 58 ++++++++++++-
.../CheckpointCommittableManagerImpl.java | 27 +++---
.../sink/committables/CommittableCollector.java | 2 +-
.../CheckpointCommittableManagerImplTest.java | 14 ++--
.../flink/test/streaming/runtime/SinkITCase.java | 97 +++++++++++++++++++---
5 files changed, 165 insertions(+), 33 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
index 93f5ba66198..35d6ca6f7ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.serialization.SerializationSchema.Initializat
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
import org.apache.flink.api.connector.sink2.Sink;
@@ -53,6 +54,8 @@ import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.util.UserCodeClassLoader;
+import org.apache.flink.shaded.guava32.com.google.common.collect.Lists;
+
import javax.annotation.Nullable;
import java.io.IOException;
@@ -67,7 +70,7 @@ import static org.apache.flink.util.Preconditions.checkState;
/**
* An operator that processes records to be written into a {@link
- * org.apache.flink.api.connector.sink.Sink}. It also has a way to process committables with the
+ * org.apache.flink.api.connector.sink2.Sink}. It also has a way to process committables with the
* same parallelism or send them downstream to a {@link CommitterOperator} with a different
* parallelism.
*
@@ -90,6 +93,13 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Nullable private final SimpleVersionedSerializer<CommT> committableSerializer;
private final List<CommT> legacyCommittables = new ArrayList<>();
+ /**
+ * Used to remember that EOI has already happened so that we don't emit the last committables of
+ * the final checkpoints twice.
+ */
+ private static final ListStateDescriptor<Boolean> END_OF_INPUT_STATE_DESC =
+ new ListStateDescriptor<>("end_of_input_state", BooleanSerializer.INSTANCE);
+
/** The runtime information of the input element. */
private final Context<InputT> context;
@@ -107,6 +117,10 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
private final MailboxExecutor mailboxExecutor;
private boolean endOfInput = false;
+ /**
+ * Remembers the endOfInput state for (final) checkpoints iff the operator emits committables.
+ */
+ @Nullable private ListState<Boolean> endOfInputState;
SinkWriterOperator(
Sink<InputT> sink,
@@ -160,17 +174,48 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
legacyCommitterState.clear();
}
}
+
sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ if (emitDownstream) {
+ // Figure out if we have seen end of input before and if we can suppress creating
+ // transactions and sending them downstream to the CommitterOperator. We have the
+ // following
+ // cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink that didn't handle EOI
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we regularly handle EOI, potentially resulting in duplicate summaries
+ // that the CommitterOperator needs to handle.
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI.
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true,
+ // we recover from a final checkpoint (for this subtask) and can ignore another EOI
+ // else we have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are
+ // true and do the same as above. As soon as one entry is false, we regularly start
+ // the writer and potentially emit duplicate summaries if we indeed recovered from a
+ // final checkpoint.
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ ArrayList<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
+ endOfInput = !previousState.isEmpty() && !previousState.contains(false);
+ }
}
@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
writerStateHandler.snapshotState(context.getCheckpointId());
+ if (endOfInputState != null) {
+ endOfInputState.clear();
+ endOfInputState.add(this.endOfInput);
+ }
}
@Override
public void processElement(StreamRecord<InputT> element) throws Exception {
+ checkState(!endOfInput, "Received element after endOfInput: %s", element);
context.element = element;
sinkWriter.write(element.getValue(), context);
}
@@ -195,9 +240,14 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(CommittableMessage.EOI);
+ if (!endOfInput) {
+ endOfInput = true;
+ if (endOfInputState != null) {
+ endOfInputState.add(true);
+ }
+ sinkWriter.flush(true);
+ emitCommittables(CommittableMessage.EOI);
+ }
}
private void emitCommittables(long checkpointId) throws IOException, InterruptedException {
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index d98ec256e50..a217116055c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.operators.sink.committables;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
@@ -72,18 +73,24 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
return subtasksCommittableManagers.values();
}
- void upsertSummary(CommittableSummary<CommT> summary) {
+ void addSummary(CommittableSummary<CommT> summary) {
+ long checkpointId = summary.getCheckpointIdOrEOI();
SubtaskCommittableManager<CommT> manager =
new SubtaskCommittableManager<>(
- summary.getNumberOfCommittables(),
- subtaskId,
- summary.getCheckpointIdOrEOI(),
- metricGroup);
- SubtaskCommittableManager<CommT> existing =
- subtasksCommittableManagers.putIfAbsent(summary.getSubtaskId(), manager);
- if (existing != null) {
- throw new UnsupportedOperationException(
- "Currently it is not supported to update the CommittableSummary for a checkpoint coming from the same subtask. Please check the status of FLINK-25920");
+ summary.getNumberOfCommittables(), subtaskId, checkpointId, metricGroup);
+ if (checkpointId == CommittableMessage.EOI) {
+ SubtaskCommittableManager<CommT> merged =
+ subtasksCommittableManagers.merge(
+ summary.getSubtaskId(), manager, SubtaskCommittableManager::merge);
+ } else {
+ SubtaskCommittableManager<CommT> existing =
+ subtasksCommittableManagers.putIfAbsent(summary.getSubtaskId(), manager);
+ if (existing != null) {
+ throw new UnsupportedOperationException(
+ String.format(
+ "Received duplicate committable summary for checkpoint %s + subtask %s (new=%s, old=%s). Please check the status of FLINK-25920",
+ checkpointId, summary.getSubtaskId(), manager, existing));
+ }
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
index 801c8446850..2dac78c71ea 100644
--- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
+++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
@@ -241,7 +241,7 @@ public class CommittableCollector<CommT> {
numberOfSubtasks,
summary.getCheckpointIdOrEOI(),
metricGroup))
- .upsertSummary(summary);
+ .addSummary(summary);
}
private void addCommittable(CommittableWithLineage<CommT> committable) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
index 5bcebda497f..6c8687b63c9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
@@ -47,7 +47,7 @@ class CheckpointCommittableManagerImplTest {
assertThat(checkpointCommittables.getSubtaskCommittableManagers()).isEmpty();
final CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, 1L, 1, 0, 0);
- checkpointCommittables.upsertSummary(first);
+ checkpointCommittables.addSummary(first);
assertThat(checkpointCommittables.getSubtaskCommittableManagers())
.hasSize(1)
.satisfiesExactly(
@@ -60,7 +60,7 @@ class CheckpointCommittableManagerImplTest {
// Add different subtask id
final CommittableSummary<Integer> third = new CommittableSummary<>(2, 1, 2L, 2, 1, 1);
- checkpointCommittables.upsertSummary(third);
+ checkpointCommittables.addSummary(third);
assertThat(checkpointCommittables.getSubtaskCommittableManagers()).hasSize(2);
}
@@ -68,8 +68,8 @@ class CheckpointCommittableManagerImplTest {
void testCommit() throws IOException, InterruptedException {
final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
new CheckpointCommittableManagerImpl<>(1, 1, 1L, METRIC_GROUP);
- checkpointCommittables.upsertSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
- checkpointCommittables.upsertSummary(new CommittableSummary<>(2, 1, 1L, 2, 0, 0));
+ checkpointCommittables.addSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
+ checkpointCommittables.addSummary(new CommittableSummary<>(2, 1, 1L, 2, 0, 0));
checkpointCommittables.addCommittable(new CommittableWithLineage<>(3, 1L, 1));
checkpointCommittables.addCommittable(new CommittableWithLineage<>(4, 1L, 2));
@@ -96,10 +96,10 @@ class CheckpointCommittableManagerImplTest {
void testUpdateCommittableSummary() {
final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
new CheckpointCommittableManagerImpl<>(1, 1, 1L, METRIC_GROUP);
- checkpointCommittables.upsertSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
+ checkpointCommittables.addSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
assertThatThrownBy(
() ->
- checkpointCommittables.upsertSummary(
+ checkpointCommittables.addSummary(
new CommittableSummary<>(1, 1, 1L, 2, 0, 0)))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("FLINK-25920");
@@ -114,7 +114,7 @@ class CheckpointCommittableManagerImplTest {
final CheckpointCommittableManagerImpl<Integer> original =
new CheckpointCommittableManagerImpl<>(
subtaskId, numberOfSubtasks, checkpointId, METRIC_GROUP);
- original.upsertSummary(
+ original.addSummary(
new CommittableSummary<>(subtaskId, numberOfSubtasks, checkpointId, 1, 0, 0));
CheckpointCommittableManagerImpl<Integer> copy = original.copy();
diff --git a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
index 0a3de995c9a..383ea9cfd33 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/streaming/runtime/SinkITCase.java
@@ -17,25 +17,38 @@
package org.apache.flink.test.streaming.runtime;
+import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.RestartStrategyOptions;
import org.apache.flink.connector.datagen.source.TestDataGenerators;
+import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.runtime.operators.sink.TestSink;
-import org.apache.flink.test.util.AbstractTestBaseJUnit4;
+import org.apache.flink.streaming.runtime.operators.sink.TestSinkV2;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.flink.testutils.junit.SharedObjectsExtension;
+import org.apache.flink.testutils.junit.SharedReference;
-import org.junit.Before;
-import org.junit.Test;
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BooleanSupplier;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -48,7 +61,7 @@ import static org.hamcrest.Matchers.containsInAnyOrder;
/**
* Integration test for {@link org.apache.flink.api.connector.sink.Sink} run time implementation.
*/
-public class SinkITCase extends AbstractTestBaseJUnit4 {
+class SinkITCase extends AbstractTestBase {
static final List<Integer> SOURCE_DATA =
Arrays.asList(
895, 127, 148, 161, 148, 662, 822, 491, 275, 122, 850, 630, 682, 765, 434, 970,
@@ -109,14 +122,17 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
COMMIT_QUEUE_RECEIVE_ALL_DATA.getAsBoolean()
&& GLOBAL_COMMIT_QUEUE_RECEIVE_ALL_DATA.getAsBoolean();
- @Before
+ @RegisterExtension
+ private final SharedObjectsExtension sharedObjects = SharedObjectsExtension.create();
+
+ @BeforeEach
public void init() {
COMMIT_QUEUE.clear();
GLOBAL_COMMIT_QUEUE.clear();
}
@Test
- public void writerAndCommitterAndGlobalCommitterExecuteInStreamingMode() throws Exception {
+ void writerAndCommitterAndGlobalCommitterExecuteInStreamingMode() throws Exception {
final StreamExecutionEnvironment env = buildStreamEnv();
final DataStream<Integer> stream =
@@ -153,7 +169,7 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void writerAndCommitterAndGlobalCommitterExecuteInBatchMode() throws Exception {
+ void writerAndCommitterAndGlobalCommitterExecuteInBatchMode() throws Exception {
final StreamExecutionEnvironment env = buildBatchEnv();
env.fromData(SOURCE_DATA)
@@ -177,7 +193,7 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void writerAndCommitterExecuteInStreamingMode() throws Exception {
+ void writerAndCommitterExecuteInStreamingMode() throws Exception {
final StreamExecutionEnvironment env = buildStreamEnv();
final DataStream<Integer> stream =
@@ -198,8 +214,45 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
containsInAnyOrder(EXPECTED_COMMITTED_DATA_IN_STREAMING_MODE.toArray()));
}
+ /**
+ * Creates a bounded stream with a failing committer. The test verifies that the Sink correctly
+ * recovers and handles multiple endInput().
+ */
+ @Test
+ void duplicateEndInput() throws Exception {
+ // we need at least 2 attempts but add a bit of a safety margin for unexpected retries
+ int maxAttempts = 10;
+ final Configuration conf = new Configuration();
+ conf.set(CheckpointingOptions.TOLERABLE_FAILURE_NUMBER, maxAttempts);
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY, "fixed-delay");
+ conf.set(RestartStrategyOptions.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, maxAttempts);
+
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(conf);
+ env.enableCheckpointing(100);
+
+ AtomicBoolean failedOnce = new AtomicBoolean(false);
+ List<String> committed = new ArrayList<>();
+ FailingOnceCommitter committer =
+ new FailingOnceCommitter(
+ sharedObjects.add(failedOnce), sharedObjects.add(committed));
+ env.<Object>fromData("bounded")
+ .sinkTo(TestSinkV2.newBuilder().setCommitter(committer).build());
+
+ JobClient jobClient = env.executeAsync();
+ // wait for job to finish including restarts
+ jobClient.getJobExecutionResult().get();
+ // Did we successfully finish?
+ Assertions.assertThat(jobClient.getJobStatus().get()).isEqualTo(JobStatus.FINISHED);
+
+ // check that we error'ed once as expected
+ Assertions.assertThat(failedOnce).isTrue();
+ // but also eventually succeed to commit (size > 1 in case of unexpected retries)
+ Assertions.assertThat(committed).isNotEmpty();
+ }
+
@Test
- public void writerAndCommitterExecuteInBatchMode() throws Exception {
+ void writerAndCommitterExecuteInBatchMode() throws Exception {
final StreamExecutionEnvironment env = buildBatchEnv();
env.fromData(SOURCE_DATA)
@@ -214,7 +267,7 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void writerAndGlobalCommitterExecuteInStreamingMode() throws Exception {
+ void writerAndGlobalCommitterExecuteInStreamingMode() throws Exception {
final StreamExecutionEnvironment env = buildStreamEnv();
final DataStream<Integer> stream =
@@ -245,7 +298,7 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
}
@Test
- public void writerAndGlobalCommitterExecuteInBatchMode() throws Exception {
+ void writerAndGlobalCommitterExecuteInBatchMode() throws Exception {
final StreamExecutionEnvironment env = buildBatchEnv();
env.fromData(SOURCE_DATA)
@@ -280,4 +333,26 @@ public class SinkITCase extends AbstractTestBaseJUnit4 {
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
return env;
}
+
+ static class FailingOnceCommitter extends TestSinkV2.DefaultCommitter {
+ private final SharedReference<AtomicBoolean> failedOnce;
+ private final SharedReference<List<String>> committed;
+
+ public FailingOnceCommitter(
+ SharedReference<AtomicBoolean> failedOnce,
+ SharedReference<List<String>> committed) {
+ this.failedOnce = failedOnce;
+ this.committed = committed;
+ }
+
+ @Override
+ public void commit(Collection<CommitRequest<String>> committables) {
+ if (failedOnce.get().compareAndSet(false, true)) {
+ throw new RuntimeException("Fail to commit");
+ }
+ for (CommitRequest<String> committable : committables) {
+ this.committed.get().add(committable.getCommittable());
+ }
+ }
+ }
}
(flink) 03/08: [FLINK-25920] Improve sink test assertions
Posted by ar...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
arvid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit ad01d71f871c34cc12798babd231592d22b6f853
Author: Arvid Heise <ar...@apache.org>
AuthorDate: Thu Sep 5 13:28:51 2024 +0200
[FLINK-25920] Improve sink test assertions
Use the proper ObjectAssert as the base for CommittableSummaryAssert and CommittableWithLinageAssert.
---
.../connector/sink2/CommittableSummaryAssert.java | 43 ++++------------------
.../sink2/CommittableWithLinageAssert.java | 27 +++-----------
.../SubtaskCommittableManagerTest.java | 24 +++---------
3 files changed, 19 insertions(+), 75 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
index e02e838aeb3..ed6ea6440ee 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummaryAssert.java
@@ -18,64 +18,37 @@
package org.apache.flink.streaming.api.connector.sink2;
-import org.assertj.core.api.AbstractAssert;
-
-import static org.assertj.core.api.Assertions.assertThat;
+import org.assertj.core.api.AbstractObjectAssert;
/** Custom assertions for {@link CommittableSummary}. */
public class CommittableSummaryAssert
- extends AbstractAssert<CommittableSummaryAssert, CommittableSummary<?>> {
+ extends AbstractObjectAssert<CommittableSummaryAssert, CommittableSummary<?>> {
public CommittableSummaryAssert(CommittableSummary<?> summary) {
super(summary, CommittableSummaryAssert.class);
}
- public CommittableSummaryAssert isEqualTo(CommittableSummary<?> summary) {
- isNotNull();
- assertThat(actual.getSubtaskId()).isEqualTo(summary.getSubtaskId());
- assertThat(actual.getCheckpointId()).isEqualTo(summary.getCheckpointId());
- assertThat(actual.getNumberOfSubtasks()).isEqualTo(summary.getNumberOfSubtasks());
- assertThat(actual.getNumberOfCommittables()).isEqualTo(summary.getNumberOfCommittables());
- assertThat(actual.getNumberOfPendingCommittables())
- .isEqualTo(summary.getNumberOfPendingCommittables());
- assertThat(actual.getNumberOfFailedCommittables())
- .isEqualTo(summary.getNumberOfFailedCommittables());
- return this;
- }
-
public CommittableSummaryAssert hasSubtaskId(int subtaskId) {
- isNotNull();
- assertThat(actual.getSubtaskId()).isEqualTo(subtaskId);
- return this;
+ return returns(subtaskId, CommittableSummary::getSubtaskId);
}
public CommittableSummaryAssert hasNumberOfSubtasks(int numberOfSubtasks) {
- isNotNull();
- assertThat(actual.getNumberOfSubtasks()).isEqualTo(numberOfSubtasks);
- return this;
+ return returns(numberOfSubtasks, CommittableSummary::getNumberOfSubtasks);
}
public CommittableSummaryAssert hasOverallCommittables(int committableNumber) {
- isNotNull();
- assertThat(actual.getNumberOfCommittables()).isEqualTo(committableNumber);
- return this;
+ return returns(committableNumber, CommittableSummary::getNumberOfCommittables);
}
public CommittableSummaryAssert hasPendingCommittables(int committableNumber) {
- isNotNull();
- assertThat(actual.getNumberOfPendingCommittables()).isEqualTo(committableNumber);
- return this;
+ return returns(committableNumber, CommittableSummary::getNumberOfPendingCommittables);
}
public CommittableSummaryAssert hasFailedCommittables(int committableNumber) {
- isNotNull();
- assertThat(actual.getNumberOfFailedCommittables()).isEqualTo(committableNumber);
- return this;
+ return returns(committableNumber, CommittableSummary::getNumberOfFailedCommittables);
}
public CommittableSummaryAssert hasCheckpointId(long checkpointId) {
- isNotNull();
- assertThat(actual.getCheckpointIdOrEOI()).isEqualTo(checkpointId);
- return this;
+ return returns(checkpointId, CommittableSummary::getCheckpointIdOrEOI);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
index 0937c7454d9..853fe6235d7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLinageAssert.java
@@ -18,45 +18,28 @@
package org.apache.flink.streaming.api.connector.sink2;
-import org.assertj.core.api.AbstractAssert;
-
-import static org.assertj.core.api.Assertions.assertThat;
+import org.assertj.core.api.AbstractObjectAssert;
/**
* Custom assertions for {@link
* org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage}.
*/
public class CommittableWithLinageAssert
- extends AbstractAssert<CommittableWithLinageAssert, CommittableWithLineage<?>> {
+ extends AbstractObjectAssert<CommittableWithLinageAssert, CommittableWithLineage<?>> {
public CommittableWithLinageAssert(CommittableWithLineage<?> summary) {
super(summary, CommittableWithLinageAssert.class);
}
- public CommittableWithLinageAssert isEqualTo(CommittableWithLineage<?> committableWithLineage) {
- isNotNull();
- assertThat(actual.getSubtaskId()).isEqualTo(committableWithLineage.getSubtaskId());
- assertThat(actual.getCheckpointIdOrEOI())
- .isEqualTo(committableWithLineage.getCheckpointIdOrEOI());
- assertThat(actual.getCommittable()).isEqualTo(committableWithLineage.getCommittable());
- return this;
- }
-
public CommittableWithLinageAssert hasCommittable(Object committable) {
- isNotNull();
- assertThat(actual.getCommittable()).isEqualTo(committable);
- return this;
+ return returns(committable, CommittableWithLineage::getCommittable);
}
public CommittableWithLinageAssert hasCheckpointId(long checkpointId) {
- isNotNull();
- assertThat(actual.getCheckpointIdOrEOI()).isEqualTo(checkpointId);
- return this;
+ return returns(checkpointId, CommittableWithLineage::getCheckpointIdOrEOI);
}
public CommittableWithLinageAssert hasSubtaskId(int subtaskId) {
- isNotNull();
- assertThat(actual.getSubtaskId()).isEqualTo(subtaskId);
- return this;
+ return returns(subtaskId, CommittableWithLineage::getSubtaskId);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
index dd4fdf91d6e..7c252e9dee3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
@@ -30,6 +30,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.stream.IntStream;
+import static org.apache.flink.streaming.api.connector.sink2.SinkV2Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThat;
class SubtaskCommittableManagerTest {
@@ -40,10 +41,9 @@ class SubtaskCommittableManagerTest {
void testDrainCommittables() {
final SubtaskCommittableManager<Integer> subtaskCommittableManager =
new SubtaskCommittableManager<>(3, 1, 1L, METRIC_GROUP);
- final CommittableWithLineage<Integer> first = new CommittableWithLineage<Integer>(1, 1L, 1);
- final CommittableWithLineage<Integer> second =
- new CommittableWithLineage<Integer>(2, 1L, 1);
- final CommittableWithLineage<Integer> third = new CommittableWithLineage<Integer>(3, 1L, 1);
+ final CommittableWithLineage<Integer> first = new CommittableWithLineage<>(1, 1L, 1);
+ final CommittableWithLineage<Integer> second = new CommittableWithLineage<>(2, 1L, 1);
+ final CommittableWithLineage<Integer> third = new CommittableWithLineage<>(3, 1L, 1);
assertThat(subtaskCommittableManager.getPendingRequests()).hasSize(0);
@@ -68,20 +68,8 @@ class SubtaskCommittableManagerTest {
final List<CommittableWithLineage<Integer>> committables =
subtaskCommittableManager.drainCommitted();
assertThat(committables).hasSize(2);
- assertThat(committables.get(0))
- .satisfies(
- c -> {
- assertThat(c.getSubtaskId()).isEqualTo(1);
- assertThat(c.getCommittable()).isEqualTo(1);
- assertThat(c.getCheckpointId()).hasValue(1L);
- });
- assertThat(committables.get(1))
- .satisfies(
- c -> {
- assertThat(c.getSubtaskId()).isEqualTo(1);
- assertThat(c.getCommittable()).isEqualTo(2);
- assertThat(c.getCheckpointId()).hasValue(1L);
- });
+ assertThat(committables.get(0)).hasSubtaskId(1).hasCommittable(1).hasCheckpointId(1);
+ assertThat(committables.get(1)).hasSubtaskId(1).hasCommittable(2).hasCheckpointId(1);
assertThat(subtaskCommittableManager.getNumFailed()).isEqualTo(0);
// Drain again should not yield anything