You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2023/11/12 16:31:42 UTC
(flink) branch master updated: [FLINK-25857] Add committer metrics to track the status of committables
This is an automated email from the ASF dual-hosted git repository.
mbalassi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new c3a07f98e5d [FLINK-25857] Add committer metrics to track the status of committables
c3a07f98e5d is described below
commit c3a07f98e5d1d7624adc967932f57d31355d9ddd
Author: pvary <pe...@gmail.com>
AuthorDate: Sun Nov 12 17:31:34 2023 +0100
[FLINK-25857] Add committer metrics to track the status of committables
Temporarily disables failing python connector tests which should be moved to external repositories.
---------
Co-authored-by: Peter Vary <pe...@apple.com>
---
.../f7a4e6fa-e7de-48c9-a61e-c13e83f0c72e | 30 ++--
.../base/sink/writer/AsyncSinkWriter.java | 6 +-
.../base/sink/writer/ElementConverter.java | 2 +-
.../connector/base/sink/ArrayListAsyncSink.java | 4 +-
.../base/sink/writer/AsyncSinkWriterTest.java | 8 +-
.../sink/writer/AsyncSinkWriterThrottlingTest.java | 2 +-
.../base/sink/writer/TestElementConverter.java | 2 +-
.../base/sink/writer/TestSinkInitContext.java | 4 +-
.../TestSinkInitContextAnyThreadMailbox.java | 2 +-
.../apache/flink/connector/file/sink/FileSink.java | 13 +-
.../org/apache/flink/api/connector/sink/Sink.java | 2 +-
.../flink/api/connector/sink2/InitContext.java | 62 +++++++
.../org/apache/flink/api/connector/sink2/Sink.java | 37 +----
.../flink/api/connector/sink2/StatefulSink.java | 7 +-
.../connector/sink2/TwoPhaseCommittingSink.java | 15 +-
flink-end-to-end-tests/run-nightly-tests.sh | 3 +-
.../metrics/groups/SinkCommitterMetricGroup.java | 49 ++++++
.../connectors/tests/test_elasticsearch.py | 3 +
.../datastream/connectors/tests/test_kafka.py | 4 +
.../datastream/connectors/tests/test_kinesis.py | 3 +
.../datastream/connectors/tests/test_pulsar.py | 4 +
.../apache/flink/runtime/metrics/MetricNames.java | 8 +
.../groups/InternalSinkCommitterMetricGroup.java | 96 +++++++++++
.../metrics/groups/MetricsGroupTestUtils.java | 10 +-
.../connector/sink2/GlobalCommitterOperator.java | 12 +-
.../connector/sink2/GlobalCommitterSerializer.java | 9 +-
.../streaming/api/functions/sink/PrintSink.java | 2 +-
.../api/functions/sink/v2/DiscardingSink.java | 2 +-
.../api/transformations/SinkV1Adapter.java | 19 +--
.../runtime/operators/sink/CommitterOperator.java | 49 +++++-
.../operators/sink/CommitterOperatorFactory.java | 2 +-
.../runtime/operators/sink/InitContextBase.java | 71 ++++++++
.../runtime/operators/sink/SinkWriterOperator.java | 59 ++-----
.../operators/sink/SinkWriterStateHandler.java | 6 +-
.../sink/StatefulSinkWriterStateHandler.java | 4 +-
.../sink/StatelessSinkWriterStateHandler.java | 4 +-
.../CheckpointCommittableManagerImpl.java | 23 ++-
.../sink/committables/CommitRequestImpl.java | 23 ++-
.../sink/committables/CommittableCollector.java | 33 ++--
.../CommittableCollectorSerializer.java | 27 +++-
.../committables/SubtaskCommittableManager.java | 26 ++-
.../sink2/GlobalCommitterSerializerTest.java | 14 +-
.../streaming/api/functions/PrintSinkTest.java | 2 +-
.../api/graph/StreamingJobGraphGeneratorTest.java | 4 +-
.../sink/SinkV2SinkWriterOperatorTest.java | 2 +-
.../operators/sink/SinkWriterOperatorTestBase.java | 12 +-
.../runtime/operators/sink/TestSinkV2.java | 14 +-
.../CheckpointCommittableManagerImplTest.java | 14 +-
.../CommittableCollectorSerializerTest.java | 22 +--
.../committables/CommittableCollectorTest.java | 10 +-
.../SubtaskCommittableManagerTest.java | 21 ++-
.../flink/streaming/util/TestExpandingSink.java | 4 +-
.../connector/upserttest/sink/UpsertTestSink.java | 2 +-
.../scheduling/SpeculativeSchedulerITCase.java | 5 +-
.../streaming/runtime/SinkV2MetricsITCase.java | 178 ++++++++++++++++++++-
55 files changed, 810 insertions(+), 241 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 5113acebcd8..0c69f24bc8b 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
@@ -33,15 +33,15 @@ Class <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperator
Class <org.apache.flink.connector.file.sink.compactor.operator.CompactorOperatorStateHandlerFactory> implements interface <org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory> in (CompactorOperatorStateHandlerFactory.java:0)
Class <org.apache.flink.connector.file.table.stream.AbstractStreamingWriter$1> implements interface <org.apache.flink.streaming.api.functions.sink.filesystem.BucketLifeCycleListener> in (AbstractStreamingWriter.java:0)
Constructor <org.apache.flink.connector.base.sink.AsyncSinkBase.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, int, int, int, long, long, long)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object, java.lang.String)> in (AsyncSinkBase.java:67)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:269)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:270)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:271)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:272)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:273)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AsyncSinkWriter.java:274)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AsyncSinkWriter.java:278)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AsyncSinkWriter.java:268)
-Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$InitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AsyncSinkWriter.java:282)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:269)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:270)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:271)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:272)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean)> in (AsyncSinkWriter.java:273)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AsyncSinkWriter.java:274)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AsyncSinkWriter.java:278)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AsyncSinkWriter.java:268)
+Constructor <org.apache.flink.connector.base.sink.writer.AsyncSinkWriter.<init>(org.apache.flink.connector.base.sink.writer.ElementConverter, org.apache.flink.api.connector.sink2.Sink$WriterInitContext, org.apache.flink.connector.base.sink.writer.config.AsyncSinkWriterConfiguration, java.util.Collection)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (AsyncSinkWriter.java:282)
Constructor <org.apache.flink.connector.base.sink.writer.strategy.AIMDScalingStrategy.<init>(int, double, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AIMDScalingStrategy.java:34)
Constructor <org.apache.flink.connector.base.sink.writer.strategy.AIMDScalingStrategy.<init>(int, double, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AIMDScalingStrategy.java:35)
Constructor <org.apache.flink.connector.base.sink.writer.strategy.AIMDScalingStrategy.<init>(int, double, int)> calls method <org.apache.flink.util.Preconditions.checkArgument(boolean, java.lang.Object)> in (AIMDScalingStrategy.java:38)
@@ -340,9 +340,9 @@ Method <org.apache.flink.connector.datagen.table.types.RowDataGenerator.snapshot
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()> 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:614)
-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:615)
-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:616)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:614)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartPrefix()> in (FileSink.java:615)
+Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartSuffix()> in (FileSink.java:616)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.enableCompact(org.apache.flink.connector.file.sink.compactor.FileCompactStrategy, org.apache.flink.connector.file.sink.compactor.FileCompactor)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FileSink.java:590)
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.enableCompact(org.apache.flink.connector.file.sink.compactor.FileCompactStrategy, org.apache.flink.connector.file.sink.compactor.FileCompactor)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FileSink.java:591)
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)
@@ -360,9 +360,9 @@ Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.withOutp
Method <org.apache.flink.connector.file.sink.FileSink$BulkFormatBuilder.withRollingPolicy(org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.CheckpointRollingPolicy)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FileSink.java:564)
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()> 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.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:420)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartPrefix()> in (FileSink.java:421)
-Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$InitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartSuffix()> in (FileSink.java:422)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.builder()> in (FileSink.java:420)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartPrefix()> in (FileSink.java:421)
+Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.createWriter(org.apache.flink.api.connector.sink2.Sink$WriterInitContext)> calls method <org.apache.flink.streaming.api.functions.sink.filesystem.OutputFileConfig.getPartSuffix()> in (FileSink.java:422)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.enableCompact(org.apache.flink.connector.file.sink.compactor.FileCompactStrategy, org.apache.flink.connector.file.sink.compactor.FileCompactor)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FileSink.java:396)
Method <org.apache.flink.connector.file.sink.FileSink$RowFormatBuilder.enableCompact(org.apache.flink.connector.file.sink.compactor.FileCompactStrategy, org.apache.flink.connector.file.sink.compactor.FileCompactor)> calls method <org.apache.flink.util.Preconditions.checkNotNull(java.lang.Object)> in (FileSink.java:397)
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)
diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java
index 7dea96df7b5..e295c356aef 100644
--- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java
+++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriter.java
@@ -208,7 +208,7 @@ public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable
@Deprecated
public AsyncSinkWriter(
ElementConverter<InputT, RequestEntryT> elementConverter,
- Sink.InitContext context,
+ Sink.WriterInitContext context,
int maxBatchSize,
int maxInFlightRequests,
int maxBufferedRequests,
@@ -234,7 +234,7 @@ public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable
@Deprecated
public AsyncSinkWriter(
ElementConverter<InputT, RequestEntryT> elementConverter,
- Sink.InitContext context,
+ Sink.WriterInitContext context,
int maxBatchSize,
int maxInFlightRequests,
int maxBufferedRequests,
@@ -258,7 +258,7 @@ public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable
public AsyncSinkWriter(
ElementConverter<InputT, RequestEntryT> elementConverter,
- Sink.InitContext context,
+ Sink.WriterInitContext context,
AsyncSinkWriterConfiguration configuration,
Collection<BufferedRequestState<RequestEntryT>> states) {
this.elementConverter = elementConverter;
diff --git a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java
index d70a27e43d9..31e3bef5a10 100644
--- a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java
+++ b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/sink/writer/ElementConverter.java
@@ -35,7 +35,7 @@ import java.io.Serializable;
public interface ElementConverter<InputT, RequestEntryT> extends Serializable {
RequestEntryT apply(InputT element, SinkWriter.Context context);
- default void open(Sink.InitContext context) {
+ default void open(Sink.WriterInitContext context) {
// No-op default implementation
}
}
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/ArrayListAsyncSink.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/ArrayListAsyncSink.java
index a07bb2793f6..1466dc36d9e 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/ArrayListAsyncSink.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/ArrayListAsyncSink.java
@@ -58,7 +58,7 @@ public class ArrayListAsyncSink extends AsyncSinkBase<String, Integer> {
@Override
public StatefulSinkWriter<String, BufferedRequestState<Integer>> createWriter(
- InitContext context) throws IOException {
+ WriterInitContext context) throws IOException {
return new AsyncSinkWriter<String, Integer>(
getElementConverter(),
context,
@@ -92,7 +92,7 @@ public class ArrayListAsyncSink extends AsyncSinkBase<String, Integer> {
@Override
public StatefulSinkWriter<String, BufferedRequestState<Integer>> restoreWriter(
- InitContext context, Collection<BufferedRequestState<Integer>> recoveredState)
+ WriterInitContext context, Collection<BufferedRequestState<Integer>> recoveredState)
throws IOException {
return createWriter(context);
}
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
index ac4d9447e15..d7a4ff7d05c 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterTest.java
@@ -1040,7 +1040,7 @@ public class AsyncSinkWriterTest {
private AsyncSinkWriterImpl(
ElementConverter<String, Integer> elementConverter,
- Sink.InitContext context,
+ Sink.WriterInitContext context,
int maxBatchSize,
int maxInFlightRequests,
int maxBufferedRequests,
@@ -1179,7 +1179,7 @@ public class AsyncSinkWriterTest {
(elem, ctx) -> Integer.parseInt(elem);
private boolean simulateFailures = false;
private int delay = 0;
- private Sink.InitContext context;
+ private Sink.WriterInitContext context;
private int maxBatchSize = 10;
private int maxInFlightRequests = 1;
private int maxBufferedRequests = 100;
@@ -1193,7 +1193,7 @@ public class AsyncSinkWriterTest {
return this;
}
- private AsyncSinkWriterImplBuilder context(Sink.InitContext context) {
+ private AsyncSinkWriterImplBuilder context(Sink.WriterInitContext context) {
this.context = context;
return this;
}
@@ -1281,7 +1281,7 @@ public class AsyncSinkWriterTest {
private final boolean blockForLimitedTime;
public AsyncSinkReleaseAndBlockWriterImpl(
- Sink.InitContext context,
+ Sink.WriterInitContext context,
int maxInFlightRequests,
CountDownLatch blockedThreadLatch,
CountDownLatch delayedStartLatch,
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterThrottlingTest.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterThrottlingTest.java
index 639cafe6d05..30b73bc6521 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterThrottlingTest.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/AsyncSinkWriterThrottlingTest.java
@@ -98,7 +98,7 @@ public class AsyncSinkWriterThrottlingTest {
public ThrottlingWriter(
ElementConverter<String, Long> elementConverter,
- Sink.InitContext context,
+ Sink.WriterInitContext context,
int maxBatchSize,
int maxInFlightRequests) {
super(
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestElementConverter.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestElementConverter.java
index fe9d2afaeab..14d006b73e6 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestElementConverter.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestElementConverter.java
@@ -28,7 +28,7 @@ public class TestElementConverter implements ElementConverter<String, Integer> {
private int openCallCount;
@Override
- public void open(Sink.InitContext context) {
+ public void open(Sink.WriterInitContext context) {
openCallCount++;
}
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java
index 1f70b03413c..2d052b008a1 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContext.java
@@ -43,8 +43,8 @@ import java.util.OptionalLong;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
-/** A mock implementation of a {@code Sink.InitContext} to be used in sink unit tests. */
-public class TestSinkInitContext implements Sink.InitContext {
+/** A mock implementation of a {@code Sink.WriterInitContext} to be used in sink unit tests. */
+public class TestSinkInitContext implements Sink.WriterInitContext {
private static final TestProcessingTimeService processingTimeService;
private final MetricListener metricListener = new MetricListener();
diff --git a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContextAnyThreadMailbox.java b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContextAnyThreadMailbox.java
index ed8c5ef4ba5..43cac59eada 100644
--- a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContextAnyThreadMailbox.java
+++ b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/sink/writer/TestSinkInitContextAnyThreadMailbox.java
@@ -22,7 +22,7 @@ import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxExecutorImpl;
import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
/**
- * A mock implementation of a {@code Sink.InitContext} to be used in sink unit tests.
+ * A mock implementation of a {@code Sink.WriterInitContext} to be used in sink unit tests.
*
* <p>The only difference between this and {@link TestSinkInitContext} is that the mailbox thread
* methods of this context may be accessed from any thread. This is useful for testing fine-grained
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
index b63e7799fb7..bc60ef5a690 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
@@ -141,13 +141,13 @@ public class FileSink<IN>
}
@Override
- public FileWriter<IN> createWriter(InitContext context) throws IOException {
+ public FileWriter<IN> createWriter(WriterInitContext context) throws IOException {
return restoreWriter(context, Collections.emptyList());
}
@Override
public FileWriter<IN> restoreWriter(
- InitContext context, Collection<FileWriterBucketState> recoveredState)
+ WriterInitContext context, Collection<FileWriterBucketState> recoveredState)
throws IOException {
FileWriter<IN> writer = bucketsBuilder.createWriter(context);
writer.initializeState(recoveredState);
@@ -167,7 +167,8 @@ public class FileSink<IN>
}
@Override
- public Committer<FileSinkCommittable> createCommitter() throws IOException {
+ public Committer<FileSinkCommittable> createCommitter(CommitterInitContext context)
+ throws IOException {
return bucketsBuilder.createCommitter();
}
@@ -291,7 +292,7 @@ public class FileSink<IN>
}
@Internal
- abstract FileWriter<IN> createWriter(final InitContext context) throws IOException;
+ abstract FileWriter<IN> createWriter(final WriterInitContext context) throws IOException;
@Internal
abstract FileCommitter createCommitter() throws IOException;
@@ -409,7 +410,7 @@ public class FileSink<IN>
}
@Override
- FileWriter<IN> createWriter(InitContext context) throws IOException {
+ FileWriter<IN> createWriter(WriterInitContext context) throws IOException {
OutputFileConfig writerFileConfig;
if (compactStrategy == null) {
writerFileConfig = outputFileConfig;
@@ -603,7 +604,7 @@ public class FileSink<IN>
}
@Override
- FileWriter<IN> createWriter(InitContext context) throws IOException {
+ FileWriter<IN> createWriter(WriterInitContext context) throws IOException {
OutputFileConfig writerFileConfig;
if (compactStrategy == null) {
writerFileConfig = outputFileConfig;
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
index d0a2568bf65..8ae60574616 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
@@ -135,7 +135,7 @@ public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializ
* The interface exposes some runtime info for creating a {@link SinkWriter}.
*
* @deprecated Please migrate to {@link org.apache.flink.api.connector.sink2.Sink} and use
- * {@link org.apache.flink.api.connector.sink2.Sink.InitContext}.
+ * {@link org.apache.flink.api.connector.sink2.Sink.WriterInitContext}.
*/
@Deprecated
@PublicEvolving
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/InitContext.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/InitContext.java
new file mode 100644
index 00000000000..ec2443aabfe
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/InitContext.java
@@ -0,0 +1,62 @@
+/*
+ * 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.api.connector.sink2;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
+
+import java.util.OptionalLong;
+
+/**
+ * Common interface which exposes runtime info for creating {@link SinkWriter} and {@link Committer}
+ * objects.
+ */
+@Internal
+public interface InitContext {
+ /**
+ * The first checkpoint id when an application is started and not recovered from a previously
+ * taken checkpoint or savepoint.
+ */
+ long INITIAL_CHECKPOINT_ID = 1;
+
+ /** @return The id of task where the committer is running. */
+ int getSubtaskId();
+
+ /** @return The number of parallel committer tasks. */
+ int getNumberOfParallelSubtasks();
+
+ /**
+ * Gets the attempt number of this parallel subtask. First attempt is numbered 0.
+ *
+ * @return Attempt number of the subtask.
+ */
+ int getAttemptNumber();
+
+ /**
+ * Returns id of the restored checkpoint, if state was restored from the snapshot of a previous
+ * execution.
+ */
+ OptionalLong getRestoredCheckpointId();
+
+ /**
+ * The ID of the current job. Note that Job ID can change in particular upon manual restart. The
+ * returned ID should NOT be used for any job management tasks.
+ */
+ JobID getJobId();
+}
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java
index 758a536119f..f22c460df3d 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/Sink.java
@@ -20,7 +20,6 @@ package org.apache.flink.api.connector.sink2;
import org.apache.flink.annotation.Experimental;
import org.apache.flink.annotation.PublicEvolving;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.operators.ProcessingTimeService;
import org.apache.flink.api.common.serialization.SerializationSchema;
@@ -31,7 +30,6 @@ import org.apache.flink.util.UserCodeClassLoader;
import java.io.IOException;
import java.io.Serializable;
import java.util.Optional;
-import java.util.OptionalLong;
import java.util.function.Consumer;
/**
@@ -55,17 +53,11 @@ public interface Sink<InputT> extends Serializable {
* @return A sink writer.
* @throws IOException for any failure during creation.
*/
- SinkWriter<InputT> createWriter(InitContext context) throws IOException;
+ SinkWriter<InputT> createWriter(WriterInitContext context) throws IOException;
/** The interface exposes some runtime info for creating a {@link SinkWriter}. */
@PublicEvolving
- interface InitContext {
- /**
- * The first checkpoint id when an application is started and not recovered from a
- * previously taken checkpoint or savepoint.
- */
- long INITIAL_CHECKPOINT_ID = 1;
-
+ interface WriterInitContext extends InitContext {
/**
* Gets the {@link UserCodeClassLoader} to load classes that are not in system's classpath,
* but are part of the jar file of a user job.
@@ -91,28 +83,9 @@ public interface Sink<InputT> extends Serializable {
*/
ProcessingTimeService getProcessingTimeService();
- /** @return The id of task where the writer is. */
- int getSubtaskId();
-
- /** @return The number of parallel Sink tasks. */
- int getNumberOfParallelSubtasks();
-
- /**
- * Gets the attempt number of this parallel subtask. First attempt is numbered 0.
- *
- * @return Attempt number of the subtask.
- */
- int getAttemptNumber();
-
/** @return The metric group this writer belongs to. */
SinkWriterMetricGroup metricGroup();
- /**
- * Returns id of the restored checkpoint, if state was restored from the snapshot of a
- * previous execution.
- */
- OptionalLong getRestoredCheckpointId();
-
/**
* Provides a view on this context as a {@link SerializationSchema.InitializationContext}.
*/
@@ -124,12 +97,6 @@ public interface Sink<InputT> extends Serializable {
/** Creates a serializer for the type of sink's input. */
<IN> TypeSerializer<IN> createInputSerializer();
- /**
- * The ID of the current job. Note that Job ID can change in particular upon manual restart.
- * The returned ID should NOT be used for any job management tasks.
- */
- JobID getJobId();
-
/**
* Returns a metadata consumer, the {@link SinkWriter} can publish metadata events of type
* {@link MetaT} to the consumer.
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java
index a1814669fbc..69180449f39 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/StatefulSink.java
@@ -45,7 +45,8 @@ public interface StatefulSink<InputT, WriterStateT> extends Sink<InputT> {
* @return A sink writer.
* @throws IOException for any failure during creation.
*/
- StatefulSinkWriter<InputT, WriterStateT> createWriter(InitContext context) throws IOException;
+ StatefulSinkWriter<InputT, WriterStateT> createWriter(WriterInitContext context)
+ throws IOException;
/**
* Create a {@link StatefulSinkWriter} from a recovered state.
@@ -55,12 +56,12 @@ public interface StatefulSink<InputT, WriterStateT> extends Sink<InputT> {
* @throws IOException for any failure during creation.
*/
StatefulSinkWriter<InputT, WriterStateT> restoreWriter(
- InitContext context, Collection<WriterStateT> recoveredState) throws IOException;
+ WriterInitContext context, Collection<WriterStateT> recoveredState) throws IOException;
/**
* Any stateful sink needs to provide this state serializer and implement {@link
* StatefulSinkWriter#snapshotState(long)} properly. The respective state is used in {@link
- * #restoreWriter(InitContext, Collection)} on recovery.
+ * #restoreWriter(WriterInitContext, Collection)} on recovery.
*
* @return the serializer of the writer's state type.
*/
diff --git a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java
index 7fcb15967bc..01977f786f5 100644
--- a/flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java
+++ b/flink-core/src/main/java/org/apache/flink/api/connector/sink2/TwoPhaseCommittingSink.java
@@ -21,6 +21,7 @@ package org.apache.flink.api.connector.sink2;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.connector.sink2.StatefulSink.StatefulSinkWriter;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import java.io.IOException;
import java.util.Collection;
@@ -49,16 +50,18 @@ public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
* @return A sink writer for the two-phase commit protocol.
* @throws IOException for any failure during creation.
*/
- PrecommittingSinkWriter<InputT, CommT> createWriter(InitContext context) throws IOException;
+ PrecommittingSinkWriter<InputT, CommT> createWriter(WriterInitContext context)
+ throws IOException;
/**
* Creates a {@link Committer} that permanently makes the previously written data visible
* through {@link Committer#commit(Collection)}.
*
+ * @param context The context information for the committer initialization.
* @return A committer for the two-phase commit protocol.
* @throws IOException for any failure during creation.
*/
- Committer<CommT> createCommitter() throws IOException;
+ Committer<CommT> createCommitter(CommitterInitContext context) throws IOException;
/** Returns the serializer of the committable type. */
SimpleVersionedSerializer<CommT> getCommittableSerializer();
@@ -77,4 +80,12 @@ public interface TwoPhaseCommittingSink<InputT, CommT> extends Sink<InputT> {
*/
Collection<CommT> prepareCommit() throws IOException, InterruptedException;
}
+
+ /** The interface exposes some runtime info for creating a {@link Committer}. */
+ @PublicEvolving
+ interface CommitterInitContext extends org.apache.flink.api.connector.sink2.InitContext {
+
+ /** @return The metric group this committer belongs to. */
+ SinkCommitterMetricGroup metricGroup();
+ }
}
diff --git a/flink-end-to-end-tests/run-nightly-tests.sh b/flink-end-to-end-tests/run-nightly-tests.sh
index 57f4fa22f7b..c1104fd0811 100755
--- a/flink-end-to-end-tests/run-nightly-tests.sh
+++ b/flink-end-to-end-tests/run-nightly-tests.sh
@@ -204,7 +204,8 @@ function run_group_2 {
run_test "Walkthrough DataStream Java nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_datastream_walkthroughs.sh java"
- run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_confluent_schema_registry.sh"
+ # Disabled and should be moved to Kafka connector repository as the test depends on the Kafka connector
+ # run_test "Avro Confluent Schema Registry nightly end-to-end test" "$END_TO_END_DIR/test-scripts/test_confluent_schema_registry.sh"
run_test "State TTL Heap backend end-to-end test" "$END_TO_END_DIR/test-scripts/test_stream_state_ttl.sh hashmap" "skip_check_exceptions"
run_test "State TTL RocksDb backend end-to-end test" "$END_TO_END_DIR/test-scripts/test_stream_state_ttl.sh rocks" "skip_check_exceptions"
diff --git a/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/SinkCommitterMetricGroup.java b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/SinkCommitterMetricGroup.java
new file mode 100644
index 00000000000..8a6cb454445
--- /dev/null
+++ b/flink-metrics/flink-metrics-core/src/main/java/org/apache/flink/metrics/groups/SinkCommitterMetricGroup.java
@@ -0,0 +1,49 @@
+/*
+ * 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.metrics.groups;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+
+/**
+ * Pre-defined metrics for sinks.
+ *
+ * <p>You should only update the metrics in the main operator thread.
+ */
+@PublicEvolving
+public interface SinkCommitterMetricGroup extends OperatorMetricGroup {
+
+ /** The total number of committables arrived. */
+ Counter getNumCommittablesTotalCounter();
+
+ /** The total number of committable failures. */
+ Counter getNumCommittablesFailureCounter();
+
+ /** The total number of committable retry. */
+ Counter getNumCommittablesRetryCounter();
+
+ /** The total number of successful committables. */
+ Counter getNumCommittablesSuccessCounter();
+
+ /** The total number of already committed committables. */
+ Counter getNumCommittablesAlreadyCommittedCounter();
+
+ /** The pending committables. */
+ void setCurrentPendingCommittablesGauge(Gauge<Integer> currentPendingCommittablesGauge);
+}
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_elasticsearch.py b/flink-python/pyflink/datastream/connectors/tests/test_elasticsearch.py
index b3007e995f6..0deb50045a0 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_elasticsearch.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_elasticsearch.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import unittest
+
from pyflink.common import Types
from pyflink.datastream.connectors import DeliveryGuarantee
from pyflink.datastream.connectors.elasticsearch import Elasticsearch7SinkBuilder, \
@@ -23,6 +25,7 @@ from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
from pyflink.util.java_utils import get_field_value, is_instance_of
+@unittest.skip("Should be moved to elasticsearch connector repo")
class FlinkElasticsearch7Test(PyFlinkStreamingTestCase):
def test_es_sink(self):
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
index cf9990af354..2c035d53f2f 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_kafka.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import unittest
+
import json
from typing import Dict
@@ -419,6 +421,7 @@ class KafkaSourceTests(PyFlinkStreamingTestCase):
return Configuration(j_configuration=j_configuration)
+@unittest.skip("Should be moved to kafka connector repo")
class KafkaSinkTests(PyFlinkStreamingTestCase):
def test_compile(self):
@@ -531,6 +534,7 @@ class KafkaRecordSerializationSchemaTests(PyFlinkTestCase):
self.assertIsNone(j_record.key())
self.assertEqual(j_record.value(), b'{"f0":"test"}')
+ @unittest.skip("Should be moved to kafka connector repo")
def test_set_topic_selector(self):
def _select(data):
data = data[0]
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_kinesis.py b/flink-python/pyflink/datastream/connectors/tests/test_kinesis.py
index 30077112219..4c28621d3ee 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_kinesis.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_kinesis.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import unittest
+
from pyflink.common import SimpleStringSchema, Types
from pyflink.datastream.connectors.kinesis import PartitionKeyGenerator, FlinkKinesisConsumer, \
KinesisStreamsSink, KinesisFirehoseSink
@@ -22,6 +24,7 @@ from pyflink.testing.test_case_utils import PyFlinkUTTestCase
from pyflink.util.java_utils import get_field_value
+@unittest.skip("Should be moved to kinesis connector repo")
class FlinkKinesisTest(PyFlinkUTTestCase):
def test_kinesis_source(self):
diff --git a/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py b/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
index 55f4659c28f..fbd21a0ba41 100644
--- a/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
+++ b/flink-python/pyflink/datastream/connectors/tests/test_pulsar.py
@@ -15,6 +15,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import unittest
+
from pyflink.common import WatermarkStrategy, SimpleStringSchema, Types, ConfigOptions, Duration
from pyflink.datastream.connectors import DeliveryGuarantee
from pyflink.datastream.connectors.pulsar import TopicRoutingMode, MessageDelayer, PulsarSink, \
@@ -176,6 +178,7 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
.set_authentication('test.class', {'k1': 'v1', 'k2': 'v2'}) \
.build()
+ @unittest.skip("Should be moved to pulsar connector repo")
def test_pulsar_sink(self):
ds = self.env.from_collection([('ab', 1), ('bdc', 2), ('cfgs', 3), ('deeefg', 4)],
type_info=Types.ROW([Types.STRING(), Types.INT()]))
@@ -250,6 +253,7 @@ class FlinkPulsarTest(PyFlinkUTTestCase):
.long_type()
.no_default_value()._j_config_option), 100)
+ @unittest.skip("Should be moved to pulsar connector repo")
def test_sink_set_topics_with_list(self):
PulsarSink.builder() \
.set_service_url('pulsar://localhost:6650') \
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java
index a404b187824..8f4e5a31bed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/MetricNames.java
@@ -127,4 +127,12 @@ public class MetricNames {
// FLIP-27 for split enumerator
public static final String UNASSIGNED_SPLITS = "unassignedSplits";
+
+ // FLIP-371 for sink committer
+ public static final String TOTAL_COMMITTABLES = "totalCommittables";
+ public static final String SUCCESSFUL_COMMITTABLES = "successfulCommittables";
+ public static final String ALREADY_COMMITTED_COMMITTABLES = "alreadyCommittedCommittables";
+ public static final String FAILED_COMMITTABLES = "failedCommittables";
+ public static final String RETRIED_COMMITTABLES = "retriedCommittables";
+ public static final String PENDING_COMMITTABLES = "pendingCommittables";
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSinkCommitterMetricGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSinkCommitterMetricGroup.java
new file mode 100644
index 00000000000..6c15862ee2b
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/metrics/groups/InternalSinkCommitterMetricGroup.java
@@ -0,0 +1,96 @@
+/*
+ * 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.runtime.metrics.groups;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
+import org.apache.flink.metrics.groups.OperatorMetricGroup;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.MetricNames;
+
+/** Special {@link MetricGroup} representing an Operator. */
+@Internal
+public class InternalSinkCommitterMetricGroup extends ProxyMetricGroup<MetricGroup>
+ implements SinkCommitterMetricGroup {
+
+ private final Counter numCommittablesTotal;
+ private final Counter numCommittablesFailure;
+ private final Counter numCommittablesRetry;
+ private final Counter numCommitatblesSuccess;
+ private final Counter numCommitatblesAlreadyCommitted;
+ private final OperatorIOMetricGroup operatorIOMetricGroup;
+
+ @VisibleForTesting
+ InternalSinkCommitterMetricGroup(
+ MetricGroup parentMetricGroup, OperatorIOMetricGroup operatorIOMetricGroup) {
+ super(parentMetricGroup);
+ numCommittablesTotal = parentMetricGroup.counter(MetricNames.TOTAL_COMMITTABLES);
+ numCommittablesFailure = parentMetricGroup.counter(MetricNames.FAILED_COMMITTABLES);
+ numCommittablesRetry = parentMetricGroup.counter(MetricNames.RETRIED_COMMITTABLES);
+ numCommitatblesSuccess = parentMetricGroup.counter(MetricNames.SUCCESSFUL_COMMITTABLES);
+ numCommitatblesAlreadyCommitted =
+ parentMetricGroup.counter(MetricNames.ALREADY_COMMITTED_COMMITTABLES);
+
+ this.operatorIOMetricGroup = operatorIOMetricGroup;
+ }
+
+ public static InternalSinkCommitterMetricGroup wrap(OperatorMetricGroup operatorMetricGroup) {
+ return new InternalSinkCommitterMetricGroup(
+ operatorMetricGroup, operatorMetricGroup.getIOMetricGroup());
+ }
+
+ @Override
+ public OperatorIOMetricGroup getIOMetricGroup() {
+ return operatorIOMetricGroup;
+ }
+
+ @Override
+ public Counter getNumCommittablesTotalCounter() {
+ return numCommittablesTotal;
+ }
+
+ @Override
+ public Counter getNumCommittablesFailureCounter() {
+ return numCommittablesFailure;
+ }
+
+ @Override
+ public Counter getNumCommittablesRetryCounter() {
+ return numCommittablesRetry;
+ }
+
+ @Override
+ public Counter getNumCommittablesSuccessCounter() {
+ return numCommitatblesSuccess;
+ }
+
+ @Override
+ public Counter getNumCommittablesAlreadyCommittedCounter() {
+ return numCommitatblesAlreadyCommitted;
+ }
+
+ @Override
+ public void setCurrentPendingCommittablesGauge(Gauge<Integer> currentPendingCommittablesGauge) {
+ parentMetricGroup.gauge(MetricNames.PENDING_COMMITTABLES, currentPendingCommittablesGauge);
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricsGroupTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricsGroupTestUtils.java
index fae09a39e9b..1308bde0b18 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricsGroupTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/metrics/groups/MetricsGroupTestUtils.java
@@ -18,7 +18,6 @@
package org.apache.flink.runtime.metrics.groups;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.metrics.groups.OperatorIOMetricGroup;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
@@ -26,22 +25,25 @@ import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
/** Util class to create metric groups for SinkV2 tests. */
public class MetricsGroupTestUtils {
- @VisibleForTesting
public static InternalSinkWriterMetricGroup mockWriterMetricGroup() {
return new InternalSinkWriterMetricGroup(
new UnregisteredMetricsGroup(),
UnregisteredMetricsGroup.createOperatorIOMetricGroup());
}
- @VisibleForTesting
public static InternalSinkWriterMetricGroup mockWriterMetricGroup(MetricGroup metricGroup) {
return new InternalSinkWriterMetricGroup(
metricGroup, UnregisteredMetricsGroup.createOperatorIOMetricGroup());
}
- @VisibleForTesting
public static InternalSinkWriterMetricGroup mockWriterMetricGroup(
MetricGroup metricGroup, OperatorIOMetricGroup operatorIOMetricGroup) {
return new InternalSinkWriterMetricGroup(metricGroup, operatorIOMetricGroup);
}
+
+ public static InternalSinkCommitterMetricGroup mockCommitterMetricGroup() {
+ return new InternalSinkCommitterMetricGroup(
+ new UnregisteredMetricsGroup(),
+ UnregisteredMetricsGroup.createOperatorIOMetricGroup());
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
index a57371e17eb..4bb861544bf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterOperator.java
@@ -24,6 +24,8 @@ import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerial
import org.apache.flink.api.connector.sink.GlobalCommitter;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.InternalSinkCommitterMetricGroup;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.graph.StreamConfig;
@@ -69,6 +71,7 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
private CommittableCollector<CommT> committableCollector;
private long lastCompletedCheckpointId = -1;
private SimpleVersionedSerializer<CommT> committableSerializer;
+ private SinkCommitterMetricGroup metricGroup;
@Nullable private GlobalCommitter<CommT, GlobalCommT> globalCommitter;
@Nullable private SimpleVersionedSerializer<GlobalCommT> globalCommittableSerializer;
@@ -88,7 +91,8 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
Output<StreamRecord<Void>> output) {
super.setup(containingTask, config, output);
committer = committerFactory.get();
- committableCollector = CommittableCollector.of(getRuntimeContext());
+ metricGroup = InternalSinkCommitterMetricGroup.wrap(metrics);
+ committableCollector = CommittableCollector.of(getRuntimeContext(), metricGroup);
committableSerializer = committableSerializerFactory.get();
if (committer instanceof SinkV1Adapter.GlobalCommitterAdapter) {
final SinkV1Adapter<?, CommT, ?, GlobalCommT>.GlobalCommitterAdapter gc =
@@ -115,13 +119,15 @@ class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamOperator
new CommittableCollectorSerializer<>(
committableSerializer,
getRuntimeContext().getIndexOfThisSubtask(),
- getRuntimeContext().getMaxNumberOfParallelSubtasks());
+ getRuntimeContext().getMaxNumberOfParallelSubtasks(),
+ metricGroup);
final SimpleVersionedSerializer<GlobalCommittableWrapper<CommT, GlobalCommT>> serializer =
new GlobalCommitterSerializer<>(
committableCollectorSerializer,
globalCommittableSerializer,
getRuntimeContext().getIndexOfThisSubtask(),
- getRuntimeContext().getMaxNumberOfParallelSubtasks());
+ getRuntimeContext().getMaxNumberOfParallelSubtasks(),
+ metricGroup);
globalCommitterState =
new SimpleVersionedListState<>(
context.getOperatorStateStore()
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializer.java
index 3c140f2fe33..36b38583517 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializer.java
@@ -24,6 +24,7 @@ import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollector;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer;
import org.apache.flink.streaming.runtime.operators.sink.committables.SinkV1CommittableDeserializer;
@@ -49,16 +50,19 @@ class GlobalCommitterSerializer<CommT, GlobalCommT>
@Nullable private final SimpleVersionedSerializer<GlobalCommT> globalCommittableSerializer;
private final int subtaskId;
private final int numberOfSubtasks;
+ private final SinkCommitterMetricGroup metricGroup;
GlobalCommitterSerializer(
CommittableCollectorSerializer<CommT> committableCollectorSerializer,
@Nullable SimpleVersionedSerializer<GlobalCommT> globalCommittableSerializer,
int subtaskId,
- int numberOfSubtasks) {
+ int numberOfSubtasks,
+ SinkCommitterMetricGroup metricGroup) {
this.committableCollectorSerializer = checkNotNull(committableCollectorSerializer);
this.globalCommittableSerializer = globalCommittableSerializer;
this.subtaskId = subtaskId;
this.numberOfSubtasks = numberOfSubtasks;
+ this.metricGroup = metricGroup;
}
@Override
@@ -107,7 +111,8 @@ class GlobalCommitterSerializer<CommT, GlobalCommT>
SinkV1CommittableDeserializer.readVersionAndDeserializeList(
globalCommittableSerializer, in);
return new GlobalCommittableWrapper<>(
- new CommittableCollector<>(subtaskId, numberOfSubtasks), globalCommittables);
+ new CommittableCollector<>(subtaskId, numberOfSubtasks, metricGroup),
+ globalCommittables);
}
private GlobalCommittableWrapper<CommT, GlobalCommT> deserializeV2(DataInputView in)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSink.java
index d510d58e7d1..58f92f8a5bf 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/PrintSink.java
@@ -84,7 +84,7 @@ public class PrintSink<IN> implements Sink<IN>, SupportsConcurrentExecutionAttem
}
@Override
- public SinkWriter<IN> createWriter(InitContext context) throws IOException {
+ public SinkWriter<IN> createWriter(WriterInitContext context) throws IOException {
final PrintSinkOutputWriter<IN> writer =
new PrintSinkOutputWriter<>(sinkIdentifier, stdErr);
writer.open(context.getSubtaskId(), context.getNumberOfParallelSubtasks());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/v2/DiscardingSink.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/v2/DiscardingSink.java
index fc7eb0d4835..47506408ea1 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/v2/DiscardingSink.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/sink/v2/DiscardingSink.java
@@ -35,7 +35,7 @@ public class DiscardingSink<IN> implements Sink<IN>, SupportsConcurrentExecution
private static final long serialVersionUID = 1L;
@Override
- public SinkWriter<IN> createWriter(InitContext context) throws IOException {
+ public SinkWriter<IN> createWriter(WriterInitContext context) throws IOException {
return new DiscardingElementWriter();
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java
index ddd421f4309..1eb427129e0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SinkV1Adapter.java
@@ -71,7 +71,7 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
}
@Override
- public SinkWriterV1Adapter<InputT, CommT, WriterStateT> createWriter(InitContext context)
+ public SinkWriterV1Adapter<InputT, CommT, WriterStateT> createWriter(WriterInitContext context)
throws IOException {
org.apache.flink.api.connector.sink.SinkWriter<InputT, CommT, WriterStateT> writer =
sink.createWriter(new InitContextAdapter(context), Collections.emptyList());
@@ -183,9 +183,9 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
private static class InitContextAdapter
implements org.apache.flink.api.connector.sink.Sink.InitContext {
- private final InitContext context;
+ private final WriterInitContext context;
- public InitContextAdapter(InitContext context) {
+ public InitContextAdapter(WriterInitContext context) {
this.context = context;
}
@@ -301,8 +301,8 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
/** Main class to simulate SinkV1 with SinkV2. */
class PlainSinkAdapter implements Sink<InputT> {
@Override
- public SinkWriterV1Adapter<InputT, CommT, WriterStateT> createWriter(InitContext context)
- throws IOException {
+ public SinkWriterV1Adapter<InputT, CommT, WriterStateT> createWriter(
+ WriterInitContext context) throws IOException {
return SinkV1Adapter.this.createWriter(context);
}
@@ -316,7 +316,8 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
implements StatefulSink<InputT, WriterStateT> {
@Override
public StatefulSinkWriter<InputT, WriterStateT> restoreWriter(
- InitContext context, Collection<WriterStateT> recoveredState) throws IOException {
+ WriterInitContext context, Collection<WriterStateT> recoveredState)
+ throws IOException {
org.apache.flink.api.connector.sink.SinkWriter<InputT, CommT, WriterStateT> writer =
sink.createWriter(
new InitContextAdapter(context), new ArrayList<>(recoveredState));
@@ -336,7 +337,7 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
private class TwoPhaseCommittingSinkAdapter extends PlainSinkAdapter
implements TwoPhaseCommittingSink<InputT, CommT>, WithCompatibleState {
@Override
- public Committer<CommT> createCommitter() throws IOException {
+ public Committer<CommT> createCommitter(CommitterInitContext context) throws IOException {
return new CommitterAdapter<>(
sink.createCommitter().orElse(new SinkV1Adapter.NoopCommitter<>()));
}
@@ -374,8 +375,8 @@ public class SinkV1Adapter<InputT, CommT, WriterStateT, GlobalCommT> implements
TwoPhaseCommittingSinkAdapter adapter = new TwoPhaseCommittingSinkAdapter();
@Override
- public Committer<CommT> createCommitter() throws IOException {
- return adapter.createCommitter();
+ public Committer<CommT> createCommitter(CommitterInitContext context) throws IOException {
+ return adapter.createCommitter(context);
}
@Override
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
index ed74daa5bf3..fc86eaeb924 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -22,6 +22,8 @@ import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.connector.sink2.Committer;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.InternalSinkCommitterMetricGroup;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
@@ -31,6 +33,7 @@ import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.BoundedOneInput;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.Output;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
import org.apache.flink.streaming.api.operators.util.SimpleVersionedListState;
import org.apache.flink.streaming.runtime.operators.sink.committables.CheckpointCommittableManager;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollector;
@@ -39,12 +42,14 @@ import org.apache.flink.streaming.runtime.operators.sink.committables.Committabl
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.util.function.FunctionWithException;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.OptionalLong;
+import static org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink.CommitterInitContext;
import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -62,10 +67,13 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
private static final long RETRY_DELAY = 1000;
private final SimpleVersionedSerializer<CommT> committableSerializer;
- private final Committer<CommT> committer;
+ private final FunctionWithException<CommitterInitContext, Committer<CommT>, IOException>
+ committerSupplier;
private final boolean emitDownstream;
private final boolean isBatchMode;
private final boolean isCheckpointingEnabled;
+ private SinkCommitterMetricGroup metricGroup;
+ private Committer<CommT> committer;
private CommittableCollector<CommT> committableCollector;
private long lastCompletedCheckpointId = -1;
@@ -82,7 +90,8 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
public CommitterOperator(
ProcessingTimeService processingTimeService,
SimpleVersionedSerializer<CommT> committableSerializer,
- Committer<CommT> committer,
+ FunctionWithException<CommitterInitContext, Committer<CommT>, IOException>
+ committerSupplier,
boolean emitDownstream,
boolean isBatchMode,
boolean isCheckpointingEnabled) {
@@ -91,7 +100,7 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
this.isCheckpointingEnabled = isCheckpointingEnabled;
this.processingTimeService = checkNotNull(processingTimeService);
this.committableSerializer = checkNotNull(committableSerializer);
- this.committer = checkNotNull(committer);
+ this.committerSupplier = checkNotNull(committerSupplier);
}
@Override
@@ -100,12 +109,16 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
StreamConfig config,
Output<StreamRecord<CommittableMessage<CommT>>> output) {
super.setup(containingTask, config, output);
- committableCollector = CommittableCollector.of(getRuntimeContext());
+ metricGroup = InternalSinkCommitterMetricGroup.wrap(getMetricGroup());
+ committableCollector = CommittableCollector.of(getRuntimeContext(), metricGroup);
}
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
+ OptionalLong checkpointId = context.getRestoredCheckpointId();
+ CommitterInitContext initContext = createInitContext(checkpointId);
+ committer = committerSupplier.apply(initContext);
committableCollectorState =
new SimpleVersionedListState<>(
context.getOperatorStateStore()
@@ -113,10 +126,11 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
new CommittableCollectorSerializer<>(
committableSerializer,
getRuntimeContext().getIndexOfThisSubtask(),
- getRuntimeContext().getNumberOfParallelSubtasks()));
+ getRuntimeContext().getNumberOfParallelSubtasks(),
+ metricGroup));
if (context.isRestored()) {
committableCollectorState.get().forEach(cc -> committableCollector.merge(cc));
- lastCompletedCheckpointId = context.getRestoredCheckpointId().getAsLong();
+ lastCompletedCheckpointId = checkpointId.getAsLong();
// try to re-commit recovered transactions as quickly as possible
commitAndEmitCheckpoints();
}
@@ -204,4 +218,27 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
public void close() throws Exception {
closeAll(committer, super::close);
}
+
+ private CommitterInitContext createInitContext(OptionalLong restoredCheckpointId) {
+ return new CommitterInitContextImp(getRuntimeContext(), metricGroup, restoredCheckpointId);
+ }
+
+ private static class CommitterInitContextImp extends InitContextBase
+ implements CommitterInitContext {
+
+ private final SinkCommitterMetricGroup metricGroup;
+
+ public CommitterInitContextImp(
+ StreamingRuntimeContext runtimeContext,
+ SinkCommitterMetricGroup metricGroup,
+ OptionalLong restoredCheckpointId) {
+ super(runtimeContext, restoredCheckpointId);
+ this.metricGroup = checkNotNull(metricGroup);
+ }
+
+ @Override
+ public SinkCommitterMetricGroup metricGroup() {
+ return metricGroup;
+ }
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
index 8f2056f7fb3..1f013ca0580 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
@@ -64,7 +64,7 @@ public final class CommitterOperatorFactory<CommT>
new CommitterOperator<>(
processingTimeService,
sink.getCommittableSerializer(),
- sink.createCommitter(),
+ context -> sink.createCommitter(context),
sink instanceof WithPostCommitTopology,
isBatchMode,
isCheckpointingEnabled);
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/InitContextBase.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/InitContextBase.java
new file mode 100644
index 00000000000..5207a6c8421
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/InitContextBase.java
@@ -0,0 +1,71 @@
+/*
+ * 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.streaming.runtime.operators.sink;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.functions.RuntimeContext;
+import org.apache.flink.api.connector.sink2.InitContext;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+
+import java.util.OptionalLong;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/** Base implementation for the {@link InitContext}. */
+public class InitContextBase implements InitContext {
+
+ private final OptionalLong restoredCheckpointId;
+
+ private final StreamingRuntimeContext runtimeContext;
+
+ public InitContextBase(
+ StreamingRuntimeContext runtimeContext, OptionalLong restoredCheckpointId) {
+ this.runtimeContext = checkNotNull(runtimeContext);
+ this.restoredCheckpointId = restoredCheckpointId;
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return runtimeContext.getNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public int getAttemptNumber() {
+ return runtimeContext.getAttemptNumber();
+ }
+
+ @Override
+ public int getSubtaskId() {
+ return runtimeContext.getIndexOfThisSubtask();
+ }
+
+ @Override
+ public OptionalLong getRestoredCheckpointId() {
+ return restoredCheckpointId;
+ }
+
+ @Override
+ public JobID getJobId() {
+ return runtimeContext.getJobId();
+ }
+
+ RuntimeContext getRuntimeContext() {
+ return runtimeContext;
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
index 68bb8cd40e7..c5657e7f374 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
@@ -17,7 +17,6 @@
package org.apache.flink.streaming.runtime.operators.sink;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.operators.MailboxExecutor;
import org.apache.flink.api.common.serialization.SerializationSchema.InitializationContext;
@@ -25,8 +24,9 @@ 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.array.BytePrimitiveArraySerializer;
+import org.apache.flink.api.connector.sink2.InitContext;
import org.apache.flink.api.connector.sink2.Sink;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.Sink.WriterInitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.StatefulSink;
import org.apache.flink.api.connector.sink2.TwoPhaseCommittingSink;
@@ -133,9 +133,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
- OptionalLong checkpointId = context.getRestoredCheckpointId();
- InitContext initContext =
- createInitContext(checkpointId.isPresent() ? checkpointId.getAsLong() : null);
+ WriterInitContext initContext = createInitContext(context.getRestoredCheckpointId());
if (context.isRestored()) {
if (committableSerializer != null) {
final ListState<List<CommT>> legacyCommitterState =
@@ -241,7 +239,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
}
}
- private Sink.InitContext createInitContext(@Nullable Long restoredCheckpointId) {
+ private WriterInitContext createInitContext(OptionalLong restoredCheckpointId) {
return new InitContextImpl(
getRuntimeContext(),
processingTimeService,
@@ -270,7 +268,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
}
}
- private static class InitContextImpl implements Sink.InitContext {
+ private static class InitContextImpl extends InitContextBase implements WriterInitContext {
private final ProcessingTimeService processingTimeService;
@@ -280,22 +278,17 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
private final StreamConfig operatorConfig;
- @Nullable private final Long restoredCheckpointId;
-
- private final StreamingRuntimeContext runtimeContext;
-
public InitContextImpl(
StreamingRuntimeContext runtimeContext,
ProcessingTimeService processingTimeService,
MailboxExecutor mailboxExecutor,
SinkWriterMetricGroup metricGroup,
StreamConfig operatorConfig,
- @Nullable Long restoredCheckpointId) {
- this.runtimeContext = checkNotNull(runtimeContext);
+ OptionalLong restoredCheckpointId) {
+ super(runtimeContext, restoredCheckpointId);
this.mailboxExecutor = checkNotNull(mailboxExecutor);
this.processingTimeService = checkNotNull(processingTimeService);
this.metricGroup = checkNotNull(metricGroup);
- this.restoredCheckpointId = restoredCheckpointId;
this.operatorConfig = checkNotNull(operatorConfig);
}
@@ -304,28 +297,19 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
return new UserCodeClassLoader() {
@Override
public ClassLoader asClassLoader() {
- return runtimeContext.getUserCodeClassLoader();
+ return getRuntimeContext().getUserCodeClassLoader();
}
@Override
public void registerReleaseHookIfAbsent(
String releaseHookName, Runnable releaseHook) {
- runtimeContext.registerUserCodeClassLoaderReleaseHookIfAbsent(
- releaseHookName, releaseHook);
+ getRuntimeContext()
+ .registerUserCodeClassLoaderReleaseHookIfAbsent(
+ releaseHookName, releaseHook);
}
};
}
- @Override
- public int getNumberOfParallelSubtasks() {
- return runtimeContext.getNumberOfParallelSubtasks();
- }
-
- @Override
- public int getAttemptNumber() {
- return runtimeContext.getAttemptNumber();
- }
-
@Override
public MailboxExecutor getMailboxExecutor() {
return mailboxExecutor;
@@ -337,23 +321,11 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
return processingTimeService;
}
- @Override
- public int getSubtaskId() {
- return runtimeContext.getIndexOfThisSubtask();
- }
-
@Override
public SinkWriterMetricGroup metricGroup() {
return metricGroup;
}
- @Override
- public OptionalLong getRestoredCheckpointId() {
- return restoredCheckpointId == null
- ? OptionalLong.empty()
- : OptionalLong.of(restoredCheckpointId);
- }
-
@Override
public InitializationContext asSerializationSchemaInitializationContext() {
return new InitContextInitializationContextAdapter(
@@ -362,19 +334,14 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Override
public boolean isObjectReuseEnabled() {
- return runtimeContext.getExecutionConfig().isObjectReuseEnabled();
+ return getRuntimeContext().getExecutionConfig().isObjectReuseEnabled();
}
@Override
public <IN> TypeSerializer<IN> createInputSerializer() {
return operatorConfig
- .<IN>getTypeSerializerIn(0, runtimeContext.getUserCodeClassLoader())
+ .<IN>getTypeSerializerIn(0, getRuntimeContext().getUserCodeClassLoader())
.duplicate();
}
-
- @Override
- public JobID getJobId() {
- return runtimeContext.getJobId();
- }
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterStateHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterStateHandler.java
index 0caf7bdb64b..50a5ef5d06b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterStateHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterStateHandler.java
@@ -17,7 +17,7 @@
package org.apache.flink.streaming.runtime.operators.sink;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.Sink.WriterInitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.runtime.state.StateInitializationContext;
@@ -38,6 +38,6 @@ interface SinkWriterStateHandler<InputT> {
void snapshotState(long checkpointId) throws Exception;
/** Creates a writer, potentially using state from {@link StateInitializationContext}. */
- SinkWriter<InputT> createWriter(InitContext initContext, StateInitializationContext context)
- throws Exception;
+ SinkWriter<InputT> createWriter(
+ WriterInitContext initContext, StateInitializationContext context) throws Exception;
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterStateHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterStateHandler.java
index b2b5a5d5108..9cadc98ae32 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterStateHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatefulSinkWriterStateHandler.java
@@ -22,7 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.Sink.WriterInitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.StatefulSink;
import org.apache.flink.api.connector.sink2.StatefulSink.StatefulSinkWriter;
@@ -88,7 +88,7 @@ final class StatefulSinkWriterStateHandler<InputT, WriterStateT>
@Override
public SinkWriter<InputT> createWriter(
- InitContext initContext, StateInitializationContext context) throws Exception {
+ WriterInitContext initContext, StateInitializationContext context) throws Exception {
final ListState<byte[]> rawState =
context.getOperatorStateStore().getListState(WRITER_RAW_STATES_DESC);
writerState =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatelessSinkWriterStateHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatelessSinkWriterStateHandler.java
index f49da996224..312b13fc15f 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatelessSinkWriterStateHandler.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatelessSinkWriterStateHandler.java
@@ -18,7 +18,7 @@
package org.apache.flink.streaming.runtime.operators.sink;
import org.apache.flink.api.connector.sink2.Sink;
-import org.apache.flink.api.connector.sink2.Sink.InitContext;
+import org.apache.flink.api.connector.sink2.Sink.WriterInitContext;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.runtime.state.StateInitializationContext;
@@ -35,7 +35,7 @@ final class StatelessSinkWriterStateHandler<InputT> implements SinkWriterStateHa
@Override
public SinkWriter<InputT> createWriter(
- InitContext initContext, StateInitializationContext context) throws Exception {
+ WriterInitContext initContext, StateInitializationContext context) throws Exception {
return sink.createWriter(initContext);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
index b2a703c8db7..38abd2d3a33 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
@@ -19,6 +19,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.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
@@ -42,21 +43,27 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
@Nullable private final Long checkpointId;
private final int subtaskId;
private final int numberOfSubtasks;
+ private final SinkCommitterMetricGroup metricGroup;
CheckpointCommittableManagerImpl(
- int subtaskId, int numberOfSubtasks, @Nullable Long checkpointId) {
- this(new HashMap<>(), subtaskId, numberOfSubtasks, checkpointId);
+ int subtaskId,
+ int numberOfSubtasks,
+ @Nullable Long checkpointId,
+ SinkCommitterMetricGroup metricGroup) {
+ this(new HashMap<>(), subtaskId, numberOfSubtasks, checkpointId, metricGroup);
}
CheckpointCommittableManagerImpl(
Map<Integer, SubtaskCommittableManager<CommT>> subtasksCommittableManagers,
int subtaskId,
int numberOfSubtasks,
- @Nullable Long checkpointId) {
+ @Nullable Long checkpointId,
+ SinkCommitterMetricGroup metricGroup) {
this.subtasksCommittableManagers = checkNotNull(subtasksCommittableManagers);
this.subtaskId = subtaskId;
this.numberOfSubtasks = numberOfSubtasks;
this.checkpointId = checkpointId;
+ this.metricGroup = metricGroup;
}
@Override
@@ -78,7 +85,8 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
subtaskId,
summary.getCheckpointId().isPresent()
? summary.getCheckpointId().getAsLong()
- : null));
+ : null,
+ metricGroup));
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");
@@ -125,7 +133,9 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
requests.forEach(CommitRequestImpl::setSelected);
committer.commit(new ArrayList<>(requests));
requests.forEach(CommitRequestImpl::setCommittedIfNoError);
- return drainFinished();
+ Collection<CommittableWithLineage<CommT>> committed = drainFinished();
+ metricGroup.setCurrentPendingCommittablesGauge(() -> getPendingRequests(false).size());
+ return committed;
}
Collection<CommitRequestImpl<CommT>> getPendingRequests(boolean fullyReceived) {
@@ -159,6 +169,7 @@ class CheckpointCommittableManagerImpl<CommT> implements CheckpointCommittableMa
.collect(Collectors.toMap(Map.Entry::getKey, (e) -> e.getValue().copy())),
subtaskId,
numberOfSubtasks,
- checkpointId);
+ checkpointId,
+ metricGroup);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommitRequestImpl.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommitRequestImpl.java
index e8c1894a199..8837f2afb32 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommitRequestImpl.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommitRequestImpl.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.operators.sink.committables;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.connector.sink2.Committer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
/**
* Internal implementation to commit a specific committable and handle the response.
@@ -32,16 +33,23 @@ public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT>
private CommT committable;
private int numRetries;
private CommitRequestState state;
+ private SinkCommitterMetricGroup metricGroup;
- protected CommitRequestImpl(CommT committable) {
+ protected CommitRequestImpl(CommT committable, SinkCommitterMetricGroup metricGroup) {
this.committable = committable;
+ this.metricGroup = metricGroup;
state = CommitRequestState.RECEIVED;
}
- protected CommitRequestImpl(CommT committable, int numRetries, CommitRequestState state) {
+ protected CommitRequestImpl(
+ CommT committable,
+ int numRetries,
+ CommitRequestState state,
+ SinkCommitterMetricGroup metricGroup) {
this.committable = committable;
this.numRetries = numRetries;
this.state = state;
+ this.metricGroup = metricGroup;
}
boolean isFinished() {
@@ -65,14 +73,14 @@ public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT>
@Override
public void signalFailedWithKnownReason(Throwable t) {
state = CommitRequestState.FAILED;
- // TODO: FLINK-25857 add metric later
+ metricGroup.getNumCommittablesFailureCounter().inc();
// let the user configure a strategy for failing and apply it here
}
@Override
public void signalFailedWithUnknownReason(Throwable t) {
state = CommitRequestState.FAILED;
- // TODO: FLINK-25857 add metric later
+ metricGroup.getNumCommittablesFailureCounter().inc();
// let the user configure a strategy for failing and apply it here
throw new IllegalStateException("Failed to commit " + committable, t);
}
@@ -81,7 +89,7 @@ public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT>
public void retryLater() {
state = CommitRequestState.RETRY;
numRetries++;
- // TODO: FLINK-25857 add metric later
+ metricGroup.getNumCommittablesRetryCounter().inc();
}
@Override
@@ -92,8 +100,8 @@ public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT>
@Override
public void signalAlreadyCommitted() {
- // TODO: FLINK-25857 add metric later
state = CommitRequestState.COMMITTED;
+ metricGroup.getNumCommittablesAlreadyCommittedCounter().inc();
}
void setSelected() {
@@ -103,10 +111,11 @@ public class CommitRequestImpl<CommT> implements Committer.CommitRequest<CommT>
void setCommittedIfNoError() {
if (state == CommitRequestState.RECEIVED) {
state = CommitRequestState.COMMITTED;
+ metricGroup.getNumCommittablesSuccessCounter().inc();
}
}
CommitRequestImpl<CommT> copy() {
- return new CommitRequestImpl<>(committable, numRetries, state);
+ return new CommitRequestImpl<>(committable, numRetries, state, metricGroup);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
index efb2ff1404c..17c7d634280 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
@@ -20,8 +20,9 @@ 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;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
@@ -54,21 +55,26 @@ public class CommittableCollector<CommT> {
private final int subtaskId;
private final int numberOfSubtasks;
+ private final SinkCommitterMetricGroup metricGroup;
- public CommittableCollector(int subtaskId, int numberOfSubtasks) {
+ public CommittableCollector(
+ int subtaskId, int numberOfSubtasks, SinkCommitterMetricGroup metricGroup) {
this.subtaskId = subtaskId;
this.numberOfSubtasks = numberOfSubtasks;
this.checkpointCommittables = new TreeMap<>();
+ this.metricGroup = metricGroup;
}
/** For deep-copy. */
CommittableCollector(
Map<Long, CheckpointCommittableManagerImpl<CommT>> checkpointCommittables,
int subtaskId,
- int numberOfSubtasks) {
+ int numberOfSubtasks,
+ SinkCommitterMetricGroup metricGroup) {
this.checkpointCommittables = new TreeMap<>(checkNotNull(checkpointCommittables));
this.subtaskId = subtaskId;
this.numberOfSubtasks = numberOfSubtasks;
+ this.metricGroup = metricGroup;
}
/**
@@ -76,12 +82,16 @@ public class CommittableCollector<CommT> {
* should be used for to instantiate a collector for all Sink V2.
*
* @param context holding runtime of information
+ * @param metricGroup storing the committable metrics
* @param <CommT> type of the committable
* @return {@link CommittableCollector}
*/
- public static <CommT> CommittableCollector<CommT> of(RuntimeContext context) {
+ public static <CommT> CommittableCollector<CommT> of(
+ RuntimeContext context, SinkCommitterMetricGroup metricGroup) {
return new CommittableCollector<>(
- context.getIndexOfThisSubtask(), context.getNumberOfParallelSubtasks());
+ context.getIndexOfThisSubtask(),
+ context.getNumberOfParallelSubtasks(),
+ metricGroup);
}
/**
@@ -89,11 +99,14 @@ public class CommittableCollector<CommT> {
* to create a collector from the state of Sink V1.
*
* @param committables list of committables
+ * @param metricGroup storing the committable metrics
* @param <CommT> type of committables
* @return {@link CommittableCollector}
*/
- static <CommT> CommittableCollector<CommT> ofLegacy(List<CommT> committables) {
- CommittableCollector<CommT> committableCollector = new CommittableCollector<>(0, 1);
+ static <CommT> CommittableCollector<CommT> ofLegacy(
+ List<CommT> committables, SinkCommitterMetricGroup metricGroup) {
+ CommittableCollector<CommT> committableCollector =
+ new CommittableCollector<>(0, 1, metricGroup);
// add a checkpoint with the lowest checkpoint id, this will be merged into the next
// checkpoint data, subtask id is arbitrary
CommittableSummary<CommT> summary =
@@ -211,7 +224,8 @@ public class CommittableCollector<CommT> {
.map(e -> Tuple2.of(e.getKey(), e.getValue().copy()))
.collect(Collectors.toMap((t) -> t.f0, (t) -> t.f1)),
subtaskId,
- numberOfSubtasks);
+ numberOfSubtasks,
+ metricGroup);
}
Collection<CheckpointCommittableManagerImpl<CommT>> getCheckpointCommittables() {
@@ -226,7 +240,8 @@ public class CommittableCollector<CommT> {
new CheckpointCommittableManagerImpl<>(
subtaskId,
numberOfSubtasks,
- summary.getCheckpointId().orElse(EOI)))
+ summary.getCheckpointId().orElse(EOI),
+ metricGroup))
.upsertSummary(summary);
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
index 80c35902e9e..0cd59305786 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializer.java
@@ -25,6 +25,7 @@ import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.core.memory.DataOutputView;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.util.CollectionUtil;
import javax.annotation.Nullable;
@@ -49,14 +50,17 @@ public final class CommittableCollectorSerializer<CommT>
private final SimpleVersionedSerializer<CommT> committableSerializer;
private final int subtaskId;
private final int numberOfSubtasks;
+ private final SinkCommitterMetricGroup metricGroup;
public CommittableCollectorSerializer(
SimpleVersionedSerializer<CommT> committableSerializer,
int subtaskId,
- int numberOfSubtasks) {
+ int numberOfSubtasks,
+ SinkCommitterMetricGroup metricGroup) {
this.committableSerializer = checkNotNull(committableSerializer);
this.subtaskId = subtaskId;
this.numberOfSubtasks = numberOfSubtasks;
+ this.metricGroup = metricGroup;
}
@Override
@@ -89,7 +93,8 @@ public final class CommittableCollectorSerializer<CommT>
private CommittableCollector<CommT> deserializeV1(DataInputView in) throws IOException {
return CommittableCollector.ofLegacy(
SinkV1CommittableDeserializer.readVersionAndDeserializeList(
- committableSerializer, in));
+ committableSerializer, in),
+ metricGroup);
}
private void serializeV2(
@@ -112,7 +117,8 @@ public final class CommittableCollectorSerializer<CommT>
Collectors.toMap(
CheckpointCommittableManagerImpl::getCheckpointId, e -> e)),
subtaskId,
- numberOfSubtasks);
+ numberOfSubtasks,
+ metricGroup);
}
private static void validateMagicNumber(DataInputView in) throws IOException {
@@ -125,6 +131,7 @@ public final class CommittableCollectorSerializer<CommT>
private class CheckpointSimpleVersionedSerializer
implements SimpleVersionedSerializer<CheckpointCommittableManagerImpl<CommT>> {
+
@Override
public int getVersion() {
return 0;
@@ -175,7 +182,11 @@ public final class CommittableCollectorSerializer<CommT>
}
return new CheckpointCommittableManagerImpl<>(
- subtasksCommittableManagers, subtaskId, numberOfSubtasks, checkpointId);
+ subtasksCommittableManagers,
+ subtaskId,
+ numberOfSubtasks,
+ checkpointId,
+ metricGroup);
}
}
@@ -236,7 +247,8 @@ public final class CommittableCollectorSerializer<CommT>
subtaskId,
checkNotNull(
checkpointId,
- "CheckpointId must be set to align the SubtaskCommittableManager with holding CheckpointCommittableManager."));
+ "CheckpointId must be set to align the SubtaskCommittableManager with holding CheckpointCommittableManager."),
+ metricGroup);
}
private class RequestSimpleVersionedSerializer
@@ -264,7 +276,10 @@ public final class CommittableCollectorSerializer<CommT>
SimpleVersionedSerialization.readVersionAndDeSerialize(
committableSerializer, in);
return new CommitRequestImpl<>(
- committable, in.readInt(), CommitRequestState.values()[in.readInt()]);
+ committable,
+ in.readInt(),
+ CommitRequestState.values()[in.readInt()],
+ metricGroup);
}
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
index 3414a8056c3..1baadeb40a8 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManager.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
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;
@@ -45,10 +46,21 @@ class SubtaskCommittableManager<CommT> {
private final int subtaskId;
private int numDrained;
private int numFailed;
+ private SinkCommitterMetricGroup metricGroup;
SubtaskCommittableManager(
- int numExpectedCommittables, int subtaskId, @Nullable Long checkpointId) {
- this(Collections.emptyList(), numExpectedCommittables, 0, 0, subtaskId, checkpointId);
+ int numExpectedCommittables,
+ int subtaskId,
+ @Nullable Long checkpointId,
+ SinkCommitterMetricGroup metricGroup) {
+ this(
+ Collections.emptyList(),
+ numExpectedCommittables,
+ 0,
+ 0,
+ subtaskId,
+ checkpointId,
+ metricGroup);
}
SubtaskCommittableManager(
@@ -57,13 +69,15 @@ class SubtaskCommittableManager<CommT> {
int numDrained,
int numFailed,
int subtaskId,
- @Nullable Long checkpointId) {
+ @Nullable Long checkpointId,
+ SinkCommitterMetricGroup metricGroup) {
this.checkpointId = checkpointId;
this.subtaskId = subtaskId;
this.numExpectedCommittables = numExpectedCommittables;
this.requests = new ArrayDeque<>(checkNotNull(requests));
this.numDrained = numDrained;
this.numFailed = numFailed;
+ this.metricGroup = metricGroup;
}
void add(CommittableWithLineage<CommT> committable) {
@@ -72,7 +86,8 @@ class SubtaskCommittableManager<CommT> {
void add(CommT committable) {
checkState(requests.size() < numExpectedCommittables, "Already received all committables.");
- requests.add(new CommitRequestImpl<>(committable));
+ requests.add(new CommitRequestImpl<>(committable, metricGroup));
+ metricGroup.getNumCommittablesTotalCounter().inc();
}
/**
@@ -189,6 +204,7 @@ class SubtaskCommittableManager<CommT> {
numDrained,
numFailed,
subtaskId,
- checkpointId);
+ checkpointId,
+ metricGroup);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializerTest.java
index b21abebce4a..0df8f1674e1 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/connector/sink2/GlobalCommitterSerializerTest.java
@@ -21,6 +21,8 @@ package org.apache.flink.streaming.api.connector.sink2;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.MetricsGroupTestUtils;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollector;
import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer;
import org.apache.flink.streaming.runtime.operators.sink.committables.SinkV1CommittableDeserializer;
@@ -40,15 +42,18 @@ class GlobalCommitterSerializerTest {
private static final int SUBTASK_ID = 0;
private static final int NUMBER_OF_SUBTASKS = 1;
+ private static final SinkCommitterMetricGroup METRIC_GROUP =
+ MetricsGroupTestUtils.mockCommitterMetricGroup();
private static final CommittableCollectorSerializer<Integer> COMMITTABLE_COLLECTOR_SERIALIZER =
new CommittableCollectorSerializer<>(
- new IntegerSerializer(), SUBTASK_ID, NUMBER_OF_SUBTASKS);
+ new IntegerSerializer(), SUBTASK_ID, NUMBER_OF_SUBTASKS, METRIC_GROUP);
private static final GlobalCommitterSerializer<Integer, String> SERIALIZER =
new GlobalCommitterSerializer<>(
COMMITTABLE_COLLECTOR_SERIALIZER,
new StringSerializer(),
SUBTASK_ID,
- NUMBER_OF_SUBTASKS);
+ NUMBER_OF_SUBTASKS,
+ METRIC_GROUP);
@ParameterizedTest
@ValueSource(booleans = {true, false})
@@ -58,9 +63,10 @@ class GlobalCommitterSerializerTest {
COMMITTABLE_COLLECTOR_SERIALIZER,
withSinkV1State ? new StringSerializer() : null,
SUBTASK_ID,
- NUMBER_OF_SUBTASKS);
+ NUMBER_OF_SUBTASKS,
+ METRIC_GROUP);
final CommittableCollector<Integer> collector =
- new CommittableCollector<>(SUBTASK_ID, NUMBER_OF_SUBTASKS);
+ new CommittableCollector<>(SUBTASK_ID, NUMBER_OF_SUBTASKS, METRIC_GROUP);
collector.addMessage(new CommittableSummary<>(2, 3, 1L, 1, 1, 0));
collector.addMessage(new CommittableWithLineage<>(1, 1L, 2));
final List<String> v1State =
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java
index 352118db5d4..d82228cfaef 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/functions/PrintSinkTest.java
@@ -159,7 +159,7 @@ class PrintSinkTest {
}
private static class MockInitContext
- implements Sink.InitContext, SerializationSchema.InitializationContext {
+ implements Sink.WriterInitContext, SerializationSchema.InitializationContext {
private final int numSubtasks;
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 8bc34767646..e4c3bdf133e 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
@@ -2324,7 +2324,7 @@ class StreamingJobGraphGeneratorTest {
WithPostCommitTopology<Integer, Void> {
@Override
- public PrecommittingSinkWriter<Integer, Void> createWriter(InitContext context)
+ public PrecommittingSinkWriter<Integer, Void> createWriter(WriterInitContext context)
throws IOException {
return new PrecommittingSinkWriter<Integer, Void>() {
@Override
@@ -2345,7 +2345,7 @@ class StreamingJobGraphGeneratorTest {
}
@Override
- public Committer<Void> createCommitter() throws IOException {
+ public Committer<Void> createCommitter(CommitterInitContext context) throws IOException {
return new Committer<Void>() {
@Override
public void commit(Collection<CommitRequest<Void>> committables)
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 5bb0135d69b..e8f50ebf2a0 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
@@ -113,7 +113,7 @@ class SinkV2SinkWriterOperatorTest extends SinkWriterOperatorTestBase {
}
@Override
- public void init(org.apache.flink.api.connector.sink2.Sink.InitContext context) {
+ public void init(org.apache.flink.api.connector.sink2.Sink.WriterInitContext context) {
this.processingTimeService = context.getProcessingTimeService();
this.processingTimeService.registerTimer(1000, this);
}
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 464b08c8f67..1810d311cb0 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
@@ -335,8 +335,8 @@ abstract class SinkWriterOperatorTestBase {
SinkV2Assertions.assertThat(((CommittableSummary<?>) cs))
.hasPendingCommittables(committables.size())
.hasCheckpointId(
- org.apache.flink.api.connector.sink2.Sink
- .InitContext.INITIAL_CHECKPOINT_ID)
+ org.apache.flink.api.connector.sink2.InitContext
+ .INITIAL_CHECKPOINT_ID)
.hasOverallCommittables(committables.size())
.hasFailedCommittables(0));
assertRestoredCommitterCommittable(
@@ -382,8 +382,8 @@ abstract class SinkWriterOperatorTestBase {
@Test
void testInitContext() throws Exception {
- final AtomicReference<org.apache.flink.api.connector.sink2.Sink.InitContext> initContext =
- new AtomicReference<>();
+ final AtomicReference<org.apache.flink.api.connector.sink2.Sink.WriterInitContext>
+ initContext = new AtomicReference<>();
final org.apache.flink.api.connector.sink2.Sink<String> sink =
context -> {
initContext.set(context);
@@ -433,8 +433,8 @@ abstract class SinkWriterOperatorTestBase {
SinkV2Assertions.assertThat((CommittableWithLineage<String>) cl)
.hasCommittable(committable)
.hasCheckpointId(
- org.apache.flink.api.connector.sink2.Sink
- .InitContext.INITIAL_CHECKPOINT_ID)
+ org.apache.flink.api.connector.sink2.InitContext
+ .INITIAL_CHECKPOINT_ID)
.hasSubtaskId(0));
}
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 bb89bf0fe7c..fad06a44e94 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
@@ -61,7 +61,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public SinkWriter<InputT> createWriter(InitContext context) {
+ public SinkWriter<InputT> createWriter(WriterInitContext context) {
writer.init(context);
return writer;
}
@@ -185,7 +185,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public Committer<String> createCommitter() {
+ public Committer<String> createCommitter(CommitterInitContext context) {
committer.init();
return committer;
}
@@ -196,7 +196,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public PrecommittingSinkWriter<InputT, String> createWriter(InitContext context) {
+ public PrecommittingSinkWriter<InputT, String> createWriter(WriterInitContext context) {
return (PrecommittingSinkWriter<InputT, String>) super.createWriter(context);
}
}
@@ -233,13 +233,13 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
}
@Override
- public DefaultStatefulSinkWriter<InputT> createWriter(InitContext context) {
+ public DefaultStatefulSinkWriter<InputT> createWriter(WriterInitContext context) {
return (DefaultStatefulSinkWriter<InputT>) super.createWriter(context);
}
@Override
public StatefulSinkWriter<InputT, String> restoreWriter(
- InitContext context, Collection<String> recoveredState) {
+ WriterInitContext context, Collection<String> recoveredState) {
DefaultStatefulSinkWriter<InputT> statefulWriter =
(DefaultStatefulSinkWriter) getWriter();
@@ -293,7 +293,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
// noting to do here
}
- public void init(InitContext context) {
+ public void init(WriterInitContext context) {
// context is not used in default case
}
}
@@ -337,7 +337,7 @@ public class TestSinkV2<InputT> implements Sink<InputT> {
// -------------------------------------- Sink Committer ---------------------------------------
/** Base class for testing {@link Committer}. */
- static class DefaultCommitter implements Committer<String>, Serializable {
+ public static class DefaultCommitter implements Committer<String>, Serializable {
@Nullable protected Queue<CommitRequest<String>> committedData;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
index a4c8c0f3c44..0ae88631a58 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImplTest.java
@@ -19,6 +19,8 @@
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.runtime.metrics.groups.MetricsGroupTestUtils;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
import org.apache.flink.streaming.api.connector.sink2.SinkV2Assertions;
@@ -35,10 +37,13 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
class CheckpointCommittableManagerImplTest {
+ private static final SinkCommitterMetricGroup METRIC_GROUP =
+ MetricsGroupTestUtils.mockCommitterMetricGroup();
+
@Test
void testAddSummary() {
final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
- new CheckpointCommittableManagerImpl<>(2, 1, 1L);
+ new CheckpointCommittableManagerImpl<>(2, 1, 1L, METRIC_GROUP);
assertThat(checkpointCommittables.getSubtaskCommittableManagers()).isEmpty();
final CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, 1L, 1, 0, 0);
@@ -62,7 +67,7 @@ class CheckpointCommittableManagerImplTest {
@Test
void testCommit() throws IOException, InterruptedException {
final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
- new CheckpointCommittableManagerImpl<>(1, 1, 1L);
+ 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.addCommittable(new CommittableWithLineage<>(3, 1L, 1));
@@ -86,7 +91,7 @@ class CheckpointCommittableManagerImplTest {
@Test
void testUpdateCommittableSummary() {
final CheckpointCommittableManagerImpl<Integer> checkpointCommittables =
- new CheckpointCommittableManagerImpl<>(1, 1, 1L);
+ new CheckpointCommittableManagerImpl<>(1, 1, 1L, METRIC_GROUP);
checkpointCommittables.upsertSummary(new CommittableSummary<>(1, 1, 1L, 1, 0, 0));
assertThatThrownBy(
() ->
@@ -103,7 +108,8 @@ class CheckpointCommittableManagerImplTest {
public void testCopy(int subtaskId, int numberOfSubtasks, long checkpointId) {
final CheckpointCommittableManagerImpl<Integer> original =
- new CheckpointCommittableManagerImpl<>(subtaskId, numberOfSubtasks, checkpointId);
+ new CheckpointCommittableManagerImpl<>(
+ subtaskId, numberOfSubtasks, checkpointId, METRIC_GROUP);
original.upsertSummary(
new CommittableSummary<>(subtaskId, numberOfSubtasks, checkpointId, 1, 0, 0));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
index 6ca121ac9e2..a460e855861 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorSerializerTest.java
@@ -18,10 +18,12 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
-import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.InitContext;
import org.apache.flink.core.io.SimpleVersionedSerialization;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.MetricsGroupTestUtils;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
import org.apache.flink.streaming.api.connector.sink2.IntegerSerializer;
@@ -48,9 +50,11 @@ class CommittableCollectorSerializerTest {
new IntegerSerializer();
private static final int SUBTASK_ID = 1;
private static final int NUMBER_OF_SUBTASKS = 1;
+ private static final SinkCommitterMetricGroup METRIC_GROUP =
+ MetricsGroupTestUtils.mockCommitterMetricGroup();
private static final CommittableCollectorSerializer<Integer> SERIALIZER =
new CommittableCollectorSerializer<>(
- COMMITTABLE_SERIALIZER, SUBTASK_ID, NUMBER_OF_SUBTASKS);
+ COMMITTABLE_SERIALIZER, SUBTASK_ID, NUMBER_OF_SUBTASKS, METRIC_GROUP);
@Test
void testCommittableCollectorV1SerDe() throws IOException {
@@ -85,10 +89,10 @@ class CommittableCollectorSerializerTest {
final CommittableCollectorSerializer<Integer> ccSerializer =
new CommittableCollectorSerializer<>(
- COMMITTABLE_SERIALIZER, subtaskId, numberOfSubtasks);
+ COMMITTABLE_SERIALIZER, subtaskId, numberOfSubtasks, METRIC_GROUP);
final CommittableCollector<Integer> committableCollector =
- new CommittableCollector<>(subtaskId, numberOfSubtasks);
+ new CommittableCollector<>(subtaskId, numberOfSubtasks, METRIC_GROUP);
committableCollector.addMessage(
new CommittableSummary<>(subtaskId, numberOfSubtasks, 1L, 1, 1, 0));
committableCollector.addMessage(
@@ -129,10 +133,10 @@ class CommittableCollectorSerializerTest {
final CommittableCollectorSerializer<Integer> ccSerializer =
new CommittableCollectorSerializer<>(
- COMMITTABLE_SERIALIZER, subtaskId, numberOfSubtasks);
+ COMMITTABLE_SERIALIZER, subtaskId, numberOfSubtasks, METRIC_GROUP);
final CommittableCollector<Integer> committableCollector =
- new CommittableCollector<>(subtaskId, numberOfSubtasks);
+ new CommittableCollector<>(subtaskId, numberOfSubtasks, METRIC_GROUP);
committableCollector.addMessage(
new CommittableSummary<>(subtaskId, numberOfSubtasks, 1L, 1, 1, 0));
committableCollector.addMessage(
@@ -169,10 +173,10 @@ class CommittableCollectorSerializerTest {
void testAlignSubtaskCommittableManagerCheckpointWithCheckpointCommittableManagerCheckpointId()
throws IOException {
// Create CommittableCollector holding a higher checkpointId than
- // Sink.InitContext#INITIAL_CHECKPOINT_ID
- long checkpointId = Sink.InitContext.INITIAL_CHECKPOINT_ID + 1;
+ // InitContext#INITIAL_CHECKPOINT_ID
+ long checkpointId = InitContext.INITIAL_CHECKPOINT_ID + 1;
final CommittableCollector<Integer> committableCollector =
- new CommittableCollector<>(SUBTASK_ID, NUMBER_OF_SUBTASKS);
+ new CommittableCollector<>(SUBTASK_ID, NUMBER_OF_SUBTASKS, METRIC_GROUP);
committableCollector.addMessage(
new CommittableSummary<>(SUBTASK_ID, NUMBER_OF_SUBTASKS, checkpointId, 1, 1, 0));
committableCollector.addMessage(new CommittableWithLineage<>(1, checkpointId, SUBTASK_ID));
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
index f8a1e90bd88..5af5efed1a9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollectorTest.java
@@ -18,6 +18,8 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.MetricsGroupTestUtils;
import org.apache.flink.streaming.api.connector.sink2.CommittableSummary;
import org.apache.flink.streaming.api.connector.sink2.SinkV2Assertions;
@@ -26,10 +28,13 @@ import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
class CommittableCollectorTest {
+ private static final SinkCommitterMetricGroup METRIC_GROUP =
+ MetricsGroupTestUtils.mockCommitterMetricGroup();
@Test
void testGetCheckpointCommittablesUpTo() {
- final CommittableCollector<Integer> committableCollector = new CommittableCollector<>(1, 1);
+ final CommittableCollector<Integer> committableCollector =
+ new CommittableCollector<>(1, 1, METRIC_GROUP);
CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, 1L, 1, 0, 0);
committableCollector.addMessage(first);
CommittableSummary<Integer> second = new CommittableSummary<>(1, 1, 2L, 1, 0, 0);
@@ -43,7 +48,8 @@ class CommittableCollectorTest {
@Test
void testGetEndOfInputCommittable() {
- final CommittableCollector<Integer> committableCollector = new CommittableCollector<>(1, 1);
+ final CommittableCollector<Integer> committableCollector =
+ new CommittableCollector<>(1, 1, METRIC_GROUP);
CommittableSummary<Integer> first = new CommittableSummary<>(1, 1, null, 1, 0, 0);
committableCollector.addMessage(first);
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
index c2ab6c1896b..0bb4f0d3db0 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/sink/committables/SubtaskCommittableManagerTest.java
@@ -18,6 +18,8 @@
package org.apache.flink.streaming.runtime.operators.sink.committables;
+import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
+import org.apache.flink.runtime.metrics.groups.MetricsGroupTestUtils;
import org.apache.flink.streaming.api.connector.sink2.CommittableWithLineage;
import org.junit.jupiter.api.Test;
@@ -31,11 +33,13 @@ import java.util.stream.IntStream;
import static org.assertj.core.api.Assertions.assertThat;
class SubtaskCommittableManagerTest {
+ private static final SinkCommitterMetricGroup METRIC_GROUP =
+ MetricsGroupTestUtils.mockCommitterMetricGroup();
@Test
void testDrainCommittables() {
final SubtaskCommittableManager<Integer> subtaskCommittableManager =
- new SubtaskCommittableManager<>(3, 1, 1L);
+ 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);
@@ -100,15 +104,24 @@ class SubtaskCommittableManagerTest {
void testMerge() {
final SubtaskCommittableManager<Integer> subtaskCommittableManager =
new SubtaskCommittableManager<>(
- Collections.singletonList(new CommitRequestImpl<>(1)), 5, 1, 2, 1, 2L);
+ Collections.singletonList(new CommitRequestImpl<>(1, METRIC_GROUP)),
+ 5,
+ 1,
+ 2,
+ 1,
+ 2L,
+ METRIC_GROUP);
subtaskCommittableManager.merge(
new SubtaskCommittableManager<>(
- Arrays.asList(new CommitRequestImpl<>(2), new CommitRequestImpl<>(3)),
+ Arrays.asList(
+ new CommitRequestImpl<>(2, METRIC_GROUP),
+ new CommitRequestImpl<>(3, METRIC_GROUP)),
10,
2,
3,
1,
- 2L));
+ 2L,
+ METRIC_GROUP));
assertThat(subtaskCommittableManager.getNumCommittables()).isEqualTo(11);
assertThat(subtaskCommittableManager.getNumDrained()).isEqualTo(3);
assertThat(subtaskCommittableManager.isFinished()).isFalse();
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestExpandingSink.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestExpandingSink.java
index 532658890de..4f93d9d923f 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestExpandingSink.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestExpandingSink.java
@@ -54,13 +54,13 @@ public class TestExpandingSink
}
@Override
- public PrecommittingSinkWriter<Integer, Integer> createWriter(InitContext context)
+ public PrecommittingSinkWriter<Integer, Integer> createWriter(WriterInitContext context)
throws IOException {
return null;
}
@Override
- public Committer<Integer> createCommitter() {
+ public Committer<Integer> createCommitter(CommitterInitContext context) {
return null;
}
diff --git a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java
index b628ccbc55a..02ec0be20a3 100644
--- a/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java
+++ b/flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/connector/upserttest/sink/UpsertTestSink.java
@@ -67,7 +67,7 @@ public class UpsertTestSink<IN> implements Sink<IN> {
@Internal
@Override
- public SinkWriter<IN> createWriter(InitContext context) {
+ public SinkWriter<IN> createWriter(WriterInitContext context) {
try {
keySerializationSchema.open(context.asSerializationSchemaInitializationContext());
valueSerializationSchema.open(context.asSerializationSchemaInitializationContext());
diff --git a/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java b/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java
index fd493575bfc..044156d817c 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeSchedulerITCase.java
@@ -527,13 +527,14 @@ class SpeculativeSchedulerITCase {
@Override
public PrecommittingSinkWriter<Long, Tuple3<Integer, Integer, Map<Long, Long>>>
- createWriter(InitContext context) {
+ createWriter(WriterInitContext context) {
return new DummyPrecommittingSinkWriter(
context.getSubtaskId(), context.getAttemptNumber());
}
@Override
- public Committer<Tuple3<Integer, Integer, Map<Long, Long>>> createCommitter() {
+ public Committer<Tuple3<Integer, Integer, Map<Long, Long>>> createCommitter(
+ CommitterInitContext context) {
return new DummyCommitter();
}
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 76fcc7b66cc..37f1f0dc214 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
@@ -22,6 +22,8 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.Metric;
import org.apache.flink.metrics.groups.OperatorMetricGroup;
import org.apache.flink.metrics.groups.SinkWriterMetricGroup;
@@ -35,19 +37,25 @@ import org.apache.flink.testutils.junit.SharedObjects;
import org.apache.flink.testutils.junit.SharedReference;
import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
import org.junit.Rule;
import org.junit.Test;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
+import java.util.stream.Collectors;
import java.util.stream.LongStream;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatCounter;
import static org.apache.flink.metrics.testutils.MetricAssertions.assertThatGauge;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasEntry;
import static org.hamcrest.Matchers.hasSize;
/** Tests whether all provided metrics of a {@link Sink} are of the expected values (FLIP-33). */
@@ -56,13 +64,14 @@ public class SinkV2MetricsITCase extends TestLogger {
private static final String TEST_SINK_NAME = "MetricTestSink";
// please refer to SinkTransformationTranslator#WRITER_NAME
private static final String DEFAULT_WRITER_NAME = "Writer";
+ private static final String DEFAULT_COMMITTER_NAME = "Committer";
private static final int DEFAULT_PARALLELISM = 4;
@Rule public final SharedObjects sharedObjects = SharedObjects.create();
- private static final InMemoryReporter reporter = InMemoryReporter.createWithRetainedMetrics();
+ private final InMemoryReporter reporter = InMemoryReporter.createWithRetainedMetrics();
- @ClassRule
- public static final MiniClusterWithClientResource MINI_CLUSTER_RESOURCE =
+ @Rule
+ public final MiniClusterWithClientResource miniClusterResource =
new MiniClusterWithClientResource(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(1)
@@ -116,6 +125,57 @@ public class SinkV2MetricsITCase extends TestLogger {
jobClient.getJobExecutionResult().get();
}
+ @Test
+ public void testCommitterMetrics() throws Exception {
+ final int numCommittables = 7;
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ // make sure all parallel instances have processed the records once before validating
+ // metrics
+ SharedReference<CountDownLatch> beforeLatch =
+ sharedObjects.add(new CountDownLatch(numCommittables));
+ SharedReference<CountDownLatch> afterLatch = sharedObjects.add(new CountDownLatch(1));
+
+ env.fromSequence(0, numCommittables - 1)
+ .returns(BasicTypeInfo.LONG_TYPE_INFO)
+ .sinkTo(
+ TestSinkV2.<Long>newBuilder()
+ .setCommitter(new MetricCommitter(beforeLatch, afterLatch))
+ .setCommittableSerializer(TestSinkV2.StringSerializer.INSTANCE)
+ .build())
+ .name(TEST_SINK_NAME);
+ JobClient jobClient = env.executeAsync();
+ final JobID jobId = jobClient.getJobID();
+
+ // Run until every committer finished with 1 commit round - everything should be retried and
+ // pending
+ beforeLatch.get().await();
+ assertSinkCommitterMetrics(
+ jobId,
+ env.getParallelism(),
+ ImmutableMap.of(
+ MetricNames.ALREADY_COMMITTED_COMMITTABLES, 0L,
+ MetricNames.FAILED_COMMITTABLES, 0L,
+ MetricNames.RETRIED_COMMITTABLES, 7L,
+ MetricNames.SUCCESSFUL_COMMITTABLES, 0L,
+ MetricNames.TOTAL_COMMITTABLES, 7L,
+ MetricNames.PENDING_COMMITTABLES, 7L));
+ afterLatch.get().countDown();
+
+ // Run until finished
+ jobClient.getJobExecutionResult().get();
+ assertSinkCommitterMetrics(
+ jobId,
+ env.getParallelism(),
+ ImmutableMap.of(
+ MetricNames.ALREADY_COMMITTED_COMMITTABLES, 1L,
+ MetricNames.FAILED_COMMITTABLES, 2L,
+ MetricNames.RETRIED_COMMITTABLES, 10L,
+ MetricNames.SUCCESSFUL_COMMITTABLES, 4L,
+ MetricNames.TOTAL_COMMITTABLES, 7L,
+ MetricNames.PENDING_COMMITTABLES, 0L));
+ }
+
@SuppressWarnings("checkstyle:WhitespaceAfter")
private void assertSinkMetrics(
JobID jobId, long processedRecordsPerSubtask, int parallelism, int numSplits) {
@@ -157,6 +217,49 @@ public class SinkV2MetricsITCase extends TestLogger {
assertThat(subtaskWithMetrics, equalTo(numSplits));
}
+ private void assertSinkCommitterMetrics(
+ JobID jobId, int parallelism, Map<String, Long> expected) {
+ List<OperatorMetricGroup> groups =
+ reporter.findOperatorMetricGroups(
+ jobId, TEST_SINK_NAME + ": " + DEFAULT_COMMITTER_NAME);
+ assertThat(groups, hasSize(parallelism));
+
+ Map<String, Long> aggregated = new HashMap<>(6);
+ for (OperatorMetricGroup group : groups) {
+ Map<String, Metric> metrics = reporter.getMetricsByGroup(group);
+
+ aggregated.merge(
+ MetricNames.SUCCESSFUL_COMMITTABLES,
+ ((Counter) metrics.get(MetricNames.SUCCESSFUL_COMMITTABLES)).getCount(),
+ Long::sum);
+ aggregated.merge(
+ MetricNames.ALREADY_COMMITTED_COMMITTABLES,
+ ((Counter) metrics.get(MetricNames.ALREADY_COMMITTED_COMMITTABLES)).getCount(),
+ Long::sum);
+ aggregated.merge(
+ MetricNames.RETRIED_COMMITTABLES,
+ ((Counter) metrics.get(MetricNames.RETRIED_COMMITTABLES)).getCount(),
+ Long::sum);
+ aggregated.merge(
+ MetricNames.FAILED_COMMITTABLES,
+ ((Counter) metrics.get(MetricNames.FAILED_COMMITTABLES)).getCount(),
+ Long::sum);
+ aggregated.merge(
+ MetricNames.TOTAL_COMMITTABLES,
+ ((Counter) metrics.get(MetricNames.TOTAL_COMMITTABLES)).getCount(),
+ Long::sum);
+ aggregated.merge(
+ MetricNames.PENDING_COMMITTABLES,
+ ((Gauge<Integer>) metrics.get(MetricNames.PENDING_COMMITTABLES))
+ .getValue()
+ .longValue(),
+ Long::sum);
+ }
+
+ expected.entrySet()
+ .forEach(e -> assertThat(aggregated, hasEntry(e.getKey(), e.getValue())));
+ }
+
private static class MetricWriter extends TestSinkV2.DefaultSinkWriter<Long> {
static final long BASE_SEND_TIME = 100;
static final long RECORD_SIZE_IN_BYTES = 10;
@@ -164,7 +267,7 @@ public class SinkV2MetricsITCase extends TestLogger {
private long sendTime;
@Override
- public void init(Sink.InitContext context) {
+ public void init(Sink.WriterInitContext context) {
this.metricGroup = context.metricGroup();
metricGroup.setCurrentSendTimeGauge(() -> sendTime);
}
@@ -180,4 +283,69 @@ public class SinkV2MetricsITCase extends TestLogger {
metricGroup.getIOMetricGroup().getNumBytesOutCounter().inc(RECORD_SIZE_IN_BYTES);
}
}
+
+ private static class MetricCommitter extends TestSinkV2.DefaultCommitter {
+ private int counter = 0;
+ private SharedReference<CountDownLatch> beforeLatch;
+ private SharedReference<CountDownLatch> afterLatch;
+
+ MetricCommitter(
+ SharedReference<CountDownLatch> beforeLatch,
+ SharedReference<CountDownLatch> afterLatch) {
+ this.beforeLatch = beforeLatch;
+ this.afterLatch = afterLatch;
+ this.counter = 0;
+ }
+
+ @Override
+ public void commit(Collection<CommitRequest<String>> committables) {
+ if (counter == 0) {
+ System.err.println(
+ "Committables arrived "
+ + Thread.currentThread().getName()
+ + " "
+ + committables.stream()
+ .map(c -> c.getCommittable())
+ .collect(Collectors.toList()));
+ committables.forEach(c -> c.retryLater());
+ } else {
+ if (counter == 1) {
+ // Wait for metrics check before continue
+ try {
+ committables.forEach(any -> beforeLatch.get().countDown());
+ afterLatch.get().await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new RuntimeException(e);
+ }
+ }
+
+ committables.forEach(
+ c -> {
+ switch (c.getCommittable().charAt(1)) {
+ case '0':
+ c.signalAlreadyCommitted();
+ // 1 already committed
+ break;
+ case '1':
+ case '2':
+ // 2 failed
+ c.signalFailedWithKnownReason(new RuntimeException());
+ break;
+ case '3':
+ // Retry without change
+ if (counter == 1) {
+ c.retryLater();
+ }
+ break;
+ case '4':
+ case '5':
+ // Retry with change
+ c.updateAndRetryLater("Retry-" + c.getCommittable());
+ }
+ });
+ }
+ counter++;
+ }
+ }
}