This is an automated email from the ASF dual-hosted git repository. arvid pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 41e8eb3cac5a3fc066a3234976a54da52b2cf2ec Author: Arvid Heise <[email protected]> AuthorDate: Mon Nov 25 14:37:48 2024 +0100 [FLINK-36788] Add InitContext to GlobalCommitter factory StandardSinkTopologies didn't expose yet the newly added ctx, such that the global committer couldn't access the metric groups --- .../connector/sink2/StandardSinkTopologies.java | 21 +++++++++- .../transformations/GlobalCommitterTransform.java | 8 ++-- .../operators/sink/CommitterInitContextImpl.java | 45 ++++++++++++++++++++++ .../runtime/operators/sink/CommitterOperator.java | 27 +------------ .../operators/sink/GlobalCommitterOperator.java | 17 +++++--- .../sink/GlobalCommitterOperatorTest.java | 2 +- 6 files changed, 85 insertions(+), 35 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/StandardSinkTopologies.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/StandardSinkTopologies.java index 691a797752e..e974bf97612 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/StandardSinkTopologies.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/StandardSinkTopologies.java @@ -20,9 +20,11 @@ package org.apache.flink.streaming.api.connector.sink2; import org.apache.flink.annotation.Experimental; import org.apache.flink.api.connector.sink2.Committer; +import org.apache.flink.api.connector.sink2.CommitterInitContext; import org.apache.flink.core.io.SimpleVersionedSerializer; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.transformations.GlobalCommitterTransform; +import org.apache.flink.util.function.SerializableFunction; import org.apache.flink.util.function.SerializableSupplier; /** This utility class provides building blocks for custom topologies. */ @@ -39,7 +41,7 @@ public class StandardSinkTopologies { */ public static <CommT> void addGlobalCommitter( DataStream<CommittableMessage<CommT>> committables, - SerializableSupplier<Committer<CommT>> committerFactory, + SerializableFunction<CommitterInitContext, Committer<CommT>> committerFactory, SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializer) { committables .getExecutionEnvironment() @@ -47,4 +49,21 @@ public class StandardSinkTopologies { new GlobalCommitterTransform<>( committables, committerFactory, committableSerializer)); } + + /** + * Adds a global committer to the pipeline that runs as final operator with a parallelism of + * one. + */ + public static <CommT> void addGlobalCommitter( + DataStream<CommittableMessage<CommT>> committables, + SerializableSupplier<Committer<CommT>> committerFactory, + SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializer) { + committables + .getExecutionEnvironment() + .addOperator( + new GlobalCommitterTransform<>( + committables, + ctx -> committerFactory.get(), + committableSerializer)); + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/transformations/GlobalCommitterTransform.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/transformations/GlobalCommitterTransform.java index 303c632259f..7b59fa8f0a7 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/transformations/GlobalCommitterTransform.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/transformations/GlobalCommitterTransform.java @@ -21,12 +21,14 @@ package org.apache.flink.streaming.api.transformations; import org.apache.flink.annotation.Internal; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.connector.sink2.Committer; +import org.apache.flink.api.connector.sink2.CommitterInitContext; import org.apache.flink.api.dag.Transformation; import org.apache.flink.core.io.SimpleVersionedSerializer; import org.apache.flink.streaming.api.connector.sink2.CommittableMessage; import org.apache.flink.streaming.api.connector.sink2.StandardSinkTopologies; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.operators.ChainingStrategy; +import org.apache.flink.util.function.SerializableFunction; import org.apache.flink.util.function.SerializableSupplier; import org.apache.flink.shaded.guava32.com.google.common.collect.Lists; @@ -45,12 +47,12 @@ import java.util.List; public class GlobalCommitterTransform<CommT> extends TransformationWithLineage<Void> { private final DataStream<CommittableMessage<CommT>> inputStream; - private final SerializableSupplier<Committer<CommT>> committerFactory; + private final SerializableFunction<CommitterInitContext, Committer<CommT>> committerFactory; private final SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializer; public GlobalCommitterTransform( DataStream<CommittableMessage<CommT>> inputStream, - SerializableSupplier<Committer<CommT>> committerFactory, + SerializableFunction<CommitterInitContext, Committer<CommT>> committerFactory, SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializer) { super(StandardSinkTopologies.GLOBAL_COMMITTER_TRANSFORMATION_NAME, Types.VOID, 1, true); this.inputStream = inputStream; @@ -78,7 +80,7 @@ public class GlobalCommitterTransform<CommT> extends TransformationWithLineage<V return inputStream; } - public SerializableSupplier<Committer<CommT>> getCommitterFactory() { + public SerializableFunction<CommitterInitContext, Committer<CommT>> getCommitterFactory() { return committerFactory; } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterInitContextImpl.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterInitContextImpl.java new file mode 100644 index 00000000000..f69fbd2fbd8 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterInitContextImpl.java @@ -0,0 +1,45 @@ +/* + * 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.connector.sink2.CommitterInitContext; +import org.apache.flink.metrics.groups.SinkCommitterMetricGroup; +import org.apache.flink.streaming.api.operators.StreamingRuntimeContext; + +import java.util.OptionalLong; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +class CommitterInitContextImpl extends InitContextBase implements CommitterInitContext { + + private final SinkCommitterMetricGroup metricGroup; + + public CommitterInitContextImpl( + StreamingRuntimeContext runtimeContext, + SinkCommitterMetricGroup metricGroup, + OptionalLong restoredCheckpointId) { + super(runtimeContext, restoredCheckpointId); + this.metricGroup = checkNotNull(metricGroup); + } + + @Override + public SinkCommitterMetricGroup metricGroup() { + return metricGroup; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java index 3a2c221a5ea..9cd85c4001a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java @@ -37,7 +37,6 @@ 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.StreamOperatorParameters; -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; @@ -123,7 +122,8 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage public void initializeState(StateInitializationContext context) throws Exception { super.initializeState(context); OptionalLong checkpointId = context.getRestoredCheckpointId(); - CommitterInitContext initContext = createInitContext(checkpointId); + CommitterInitContext initContext = + new CommitterInitContextImpl(getRuntimeContext(), metricGroup, checkpointId); committer = committerSupplier.apply(initContext); committableCollectorState = new SimpleVersionedListState<>( @@ -213,27 +213,4 @@ 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-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperator.java index 3b80a0870ab..9c229b0bb2c 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperator.java @@ -23,6 +23,7 @@ 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.Committer; +import org.apache.flink.api.connector.sink2.CommitterInitContext; import org.apache.flink.configuration.SinkOptions; import org.apache.flink.core.io.SimpleVersionedSerializer; import org.apache.flink.metrics.groups.SinkCommitterMetricGroup; @@ -40,6 +41,7 @@ import org.apache.flink.streaming.runtime.operators.sink.committables.Committabl import org.apache.flink.streaming.runtime.operators.sink.committables.CommittableCollectorSerializer; import org.apache.flink.streaming.runtime.streamrecord.StreamRecord; import org.apache.flink.streaming.runtime.tasks.StreamTask; +import org.apache.flink.util.function.SerializableFunction; import org.apache.flink.util.function.SerializableSupplier; import javax.annotation.Nullable; @@ -48,6 +50,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.OptionalLong; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -101,7 +104,7 @@ public class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamO new ListStateDescriptor<>( "streaming_committer_raw_states", BytePrimitiveArraySerializer.INSTANCE); - private final SerializableSupplier<Committer<CommT>> committerFactory; + private final SerializableFunction<CommitterInitContext, Committer<CommT>> committerFactory; private final SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializerFactory; /** @@ -122,7 +125,7 @@ public class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamO private List<GlobalCommT> sinkV1State = new ArrayList<>(); public GlobalCommitterOperator( - SerializableSupplier<Committer<CommT>> committerFactory, + SerializableFunction<CommitterInitContext, Committer<CommT>> committerFactory, SerializableSupplier<SimpleVersionedSerializer<CommT>> committableSerializerFactory, boolean commitOnInput) { this.committerFactory = checkNotNull(committerFactory); @@ -136,7 +139,6 @@ public class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamO StreamConfig config, Output<StreamRecord<Void>> output) { super.setup(containingTask, config, output); - committer = committerFactory.get(); metricGroup = InternalSinkCommitterMetricGroup.wrap(metrics); committableCollector = CommittableCollector.of(metricGroup); committableSerializer = committableSerializerFactory.get(); @@ -156,6 +158,11 @@ public class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamO @Override public void initializeState(StateInitializationContext context) throws Exception { super.initializeState(context); + OptionalLong restoredCheckpointId = context.getRestoredCheckpointId(); + committer = + committerFactory.apply( + new CommitterInitContextImpl( + getRuntimeContext(), metricGroup, restoredCheckpointId)); globalCommitterState = new SimpleVersionedListState<>( context.getOperatorStateStore() @@ -170,8 +177,8 @@ public class GlobalCommitterOperator<CommT, GlobalCommT> extends AbstractStreamO committableCollector.merge(cc.getCommittableCollector()); }); // try to re-commit recovered transactions as quickly as possible - if (context.getRestoredCheckpointId().isPresent()) { - commit(context.getRestoredCheckpointId().getAsLong()); + if (restoredCheckpointId.isPresent()) { + commit(restoredCheckpointId.getAsLong()); } } } diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperatorTest.java index b2bd505fcff..a73dcc24d01 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperatorTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/runtime/operators/sink/GlobalCommitterOperatorTest.java @@ -176,7 +176,7 @@ class GlobalCommitterOperatorTest { Committer<Integer> committer, boolean commitOnInput) throws Exception { return new OneInputStreamOperatorTestHarness<>( new GlobalCommitterOperator<>( - () -> committer, IntegerSerializer::new, commitOnInput)); + ctx -> committer, IntegerSerializer::new, commitOnInput)); } private static class MockCommitter implements Committer<Integer> {
