You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ar...@apache.org on 2021/08/06 05:54:52 UTC
[flink] 02/03: [FLINK-23617][datastream] Add SinkOperator and
CommitterOperator and introducing write state and commit handler.
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 71aa777431f1026222b1abc0c8c0b12a96acf21f
Author: Arvid Heise <ar...@ververica.com>
AuthorDate: Thu Aug 5 20:15:53 2021 +0200
[FLINK-23617][datastream] Add SinkOperator and CommitterOperator and introducing write state and commit handler.
The SinkOperator is writing data and can directly use a co-located commit handler. It optionally forwards committables a CommitterOperator with a different parallelism where another commit handler is applied.
Note that both operators exchange the committables as byte[] using the committable serializer to avoid falling back to Kryo.
---
.../operators/sink/AbstractCommitterHandler.java | 51 ++++
.../sink/AbstractStreamingCommitterOperator.java | 6 +
.../runtime/operators/sink/CommitterHandler.java | 69 +++++
.../runtime/operators/sink/CommitterOperator.java | 127 +++++++++
.../operators/sink/ForwardCommittingHandler.java | 35 +++
.../operators/sink/NoopCommitterHandler.java | 42 +++
.../runtime/operators/sink/SinkOperator.java | 301 +++++++++++++++++++++
.../operators/sink/SinkWriterStateHandler.java | 43 +++
.../sink/StatelessSinkWriterStateHandler.java | 43 +++
.../operators/sink/StreamingCommitterOperator.java | 1 +
.../sink/StreamingGlobalCommitterOperator.java | 1 +
11 files changed, 719 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java
new file mode 100644
index 0000000..c7548a9
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java
@@ -0,0 +1,51 @@
+/*
+ * 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.util.function.SupplierWithException;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.List;
+
+abstract class AbstractCommitterHandler<InputT, OutputT>
+ implements CommitterHandler<InputT, OutputT> {
+
+ /** Record all the committables until commit. */
+ private final Deque<InputT> committables = new ArrayDeque<>();
+
+ @Override
+ public List<OutputT> processCommittables(
+ SupplierWithException<List<InputT>, Exception> committableSupplier) throws Exception {
+ this.committables.addAll(committableSupplier.get());
+ return Collections.emptyList();
+ }
+
+ protected List<InputT> pollCommittables() {
+ List<InputT> committables = new ArrayList<>(this.committables);
+ this.committables.clear();
+ return committables;
+ }
+
+ @Override
+ public void close() throws Exception {
+ committables.clear();
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterOperator.java
index 712f6a6..d9c2309 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterOperator.java
@@ -162,4 +162,10 @@ abstract class AbstractStreamingCommitterOperator<InputT, CommT>
output.collect(new StreamRecord<>(committable));
}
}
+
+ @Override
+ public void close() throws Exception {
+ committablesPerCheckpoint.clear();
+ super.close();
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
new file mode 100644
index 0000000..e642944
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
@@ -0,0 +1,69 @@
+/*
+ * 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.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.util.function.SupplierWithException;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A wrapper around a {@link org.apache.flink.api.connector.sink.Committer} or {@link
+ * org.apache.flink.api.connector.sink.GlobalCommitter} that manages states.
+ *
+ * @param <InputT>
+ * @param <OutputT>
+ */
+interface CommitterHandler<InputT, OutputT> extends AutoCloseable, Serializable {
+
+ /** Initializes the state of the committer and this handler. */
+ default void initializeState(StateInitializationContext context) throws Exception {}
+
+ /** Snapshots the state of the committer and this handler. */
+ default void snapshotState(StateSnapshotContext context) throws Exception {}
+
+ /**
+ * Processes the committables by either directly transforming them or by adding them to the
+ * internal state of this handler. The supplier should only be queried once.
+ *
+ * @return a list of output committables that is send downstream.
+ */
+ List<OutputT> processCommittables(
+ SupplierWithException<List<InputT>, Exception> committableSupplier) throws Exception;
+
+ /**
+ * Called when no more committables are going to be added through {@link
+ * #processCommittables(SupplierWithException)}.
+ *
+ * @return a list of output committables that is send downstream.
+ */
+ default List<OutputT> endOfInput() throws IOException, InterruptedException {
+ return Collections.emptyList();
+ }
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId)
+ throws IOException, InterruptedException {
+ return Collections.emptyList();
+ }
+}
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
new file mode 100644
index 0000000..d8e1653
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
@@ -0,0 +1,127 @@
+/*
+ * 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.core.io.SimpleVersionedSerialization;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+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.runtime.streamrecord.StreamRecord;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * An operator that processes committables of a {@link org.apache.flink.api.connector.sink.Sink}.
+ *
+ * <p>The operator may be part of a sink pipeline but usually is the last operator. There are
+ * currently two ways this operator is used:
+ *
+ * <ul>
+ * <li>In streaming mode, there is a {@link SinkOperator} with parallelism p containing {@link
+ * org.apache.flink.api.connector.sink.SinkWriter} and {@link
+ * org.apache.flink.api.connector.sink.Committer} and this operator containing the {@link
+ * org.apache.flink.api.connector.sink.GlobalCommitter} with parallelism 1.
+ * <li>In batch mode, there is a {@link SinkOperator} with parallelism p containing {@link
+ * org.apache.flink.api.connector.sink.SinkWriter} and this operator containing the {@link
+ * org.apache.flink.api.connector.sink.Committer} and {@link
+ * org.apache.flink.api.connector.sink.GlobalCommitter} with parallelism 1.
+ * </ul>
+ *
+ * @param <InputT> the type of the committable
+ * @param <OutputT> the type of the committable to send to downstream operators
+ */
+class CommitterOperator<InputT, OutputT> extends AbstractStreamOperator<byte[]>
+ implements OneInputStreamOperator<byte[], byte[]>, BoundedOneInput {
+
+ private final SimpleVersionedSerializer<InputT> inputSerializer;
+ private final CommitterHandler<InputT, OutputT> committerHandler;
+ @Nullable private final SimpleVersionedSerializer<OutputT> outputSerializer;
+
+ public CommitterOperator(
+ SimpleVersionedSerializer<InputT> inputSerializer,
+ CommitterHandler<InputT, OutputT> committerHandler,
+ SimpleVersionedSerializer<OutputT> outputSerializer) {
+ this.inputSerializer = inputSerializer;
+ this.committerHandler = committerHandler;
+ this.outputSerializer = outputSerializer;
+ }
+
+ public CommitterOperator(
+ SimpleVersionedSerializer<InputT> inputSerializer,
+ CommitterHandler<InputT, OutputT> committerHandler) {
+ this.inputSerializer = inputSerializer;
+ this.committerHandler = committerHandler;
+ this.outputSerializer = null;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ committerHandler.initializeState(context);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ committerHandler.snapshotState(context);
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ emitCommittables(committerHandler.endOfInput());
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ super.notifyCheckpointComplete(checkpointId);
+ emitCommittables(committerHandler.notifyCheckpointCompleted(checkpointId));
+ }
+
+ private void emitCommittables(Collection<OutputT> committables) throws IOException {
+ if (outputSerializer != null) {
+ for (OutputT committable : committables) {
+ output.collect(
+ new StreamRecord<>(
+ SimpleVersionedSerialization.writeVersionAndSerialize(
+ outputSerializer, committable)));
+ }
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<byte[]> element) throws Exception {
+ committerHandler.processCommittables(
+ () ->
+ Collections.singletonList(
+ SimpleVersionedSerialization.readVersionAndDeSerialize(
+ inputSerializer, element.getValue())));
+ }
+
+ @Override
+ public void close() throws Exception {
+ committerHandler.close();
+ super.close();
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/ForwardCommittingHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/ForwardCommittingHandler.java
new file mode 100644
index 0000000..d8a3ea2
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/ForwardCommittingHandler.java
@@ -0,0 +1,35 @@
+/*
+ * 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.util.function.SupplierWithException;
+
+import java.util.List;
+
+/**
+ * This committer handler simply forwards all committables downstream. It's used in {@link
+ * SinkOperator} without committers but with downstream operators (in streaming, only global
+ * committer on sink; in batch, committer or global committer present).
+ */
+class ForwardCommittingHandler<CommT> extends AbstractCommitterHandler<CommT, CommT> {
+ @Override
+ public List<CommT> processCommittables(
+ SupplierWithException<List<CommT>, Exception> committableSupplier) throws Exception {
+ return committableSupplier.get();
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/NoopCommitterHandler.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/NoopCommitterHandler.java
new file mode 100644
index 0000000..fc48cbf
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/NoopCommitterHandler.java
@@ -0,0 +1,42 @@
+/*
+ * 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.util.function.SupplierWithException;
+
+import java.util.Collections;
+import java.util.List;
+
+/** Swallows all committables and emits nothing. */
+enum NoopCommitterHandler implements CommitterHandler<Object, Object> {
+ INSTANCE;
+
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> getInstance() {
+ return (CommitterHandler<InputT, OutputT>) NoopCommitterHandler.INSTANCE;
+ }
+
+ @Override
+ public List<Object> processCommittables(
+ SupplierWithException<List<Object>, Exception> committableSupplier) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void close() throws Exception {}
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
new file mode 100644
index 0000000..ec08dc8
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
@@ -0,0 +1,301 @@
+/*
+ * 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.operators.MailboxExecutor;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerialization;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.state.StateInitializationContext;
+import org.apache.flink.runtime.state.StateSnapshotContext;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.BoundedOneInput;
+import org.apache.flink.streaming.api.operators.InternalTimerService;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
+import org.apache.flink.streaming.api.operators.StreamingRuntimeContext;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.util.UserCodeClassLoader;
+import org.apache.flink.util.function.BiFunctionWithException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * An operator that processes records to be written into a {@link
+ * org.apache.flink.api.connector.sink.Sink}. It also has a way to process committables with the
+ * same parallelism or send them downstream to a {@link CommitterOperator} with a different
+ * parallelism.
+ *
+ * <p>The operator may be part of a sink pipeline and is the first operator. There are currently two
+ * ways this operator is used:
+ *
+ * <ul>
+ * <li>In streaming mode, there is this operator with parallelism p containing {@link
+ * org.apache.flink.api.connector.sink.SinkWriter} and {@link
+ * org.apache.flink.api.connector.sink.Committer} and a {@link CommitterOperator} containing
+ * the {@link org.apache.flink.api.connector.sink.GlobalCommitter} with parallelism 1.
+ * <li>In batch mode, there is this operator with parallelism p containing {@link
+ * org.apache.flink.api.connector.sink.SinkWriter} and a {@link CommitterOperator} containing
+ * the {@link org.apache.flink.api.connector.sink.Committer} and {@link
+ * org.apache.flink.api.connector.sink.GlobalCommitter} with parallelism 1.
+ * </ul>
+ *
+ * @param <InputT> the type of the committable
+ * @param <CommT> the type of the committable (to send to downstream operators)
+ * @param <WriterStateT> the type of the writer state for stateful sinks
+ */
+class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<byte[]>
+ implements OneInputStreamOperator<InputT, byte[]>, BoundedOneInput {
+
+ /** The runtime information of the input element. */
+ private final Context<InputT> context;
+
+ // ------------------------------- runtime fields ---------------------------------------
+
+ /** We listen to this ourselves because we don't have an {@link InternalTimerService}. */
+ private Long currentWatermark;
+
+ private SinkWriter<InputT, CommT, WriterStateT> sinkWriter;
+
+ private final SinkWriterStateHandler<WriterStateT> sinkWriterStateHandler;
+
+ private final CommitterHandler<CommT, CommT> committerHandler;
+ @Nullable private final SimpleVersionedSerializer<CommT> committableSerializer;
+
+ private final BiFunctionWithException<
+ Sink.InitContext,
+ List<WriterStateT>,
+ SinkWriter<InputT, CommT, WriterStateT>,
+ IOException>
+ writerFactory;
+
+ private final MailboxExecutor mailboxExecutor;
+
+ SinkOperator(
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor,
+ BiFunctionWithException<
+ Sink.InitContext,
+ List<WriterStateT>,
+ SinkWriter<InputT, CommT, WriterStateT>,
+ IOException>
+ writerFactory,
+ SinkWriterStateHandler<WriterStateT> sinkWriterStateHandler,
+ CommitterHandler<CommT, CommT> committerHandler,
+ @Nullable SimpleVersionedSerializer<CommT> committableSerializer) {
+ this.processingTimeService = checkNotNull(processingTimeService);
+ this.mailboxExecutor = checkNotNull(mailboxExecutor);
+ this.writerFactory = checkNotNull(writerFactory);
+ this.sinkWriterStateHandler = checkNotNull(sinkWriterStateHandler);
+ this.committerHandler = checkNotNull(committerHandler);
+ this.committableSerializer = committableSerializer;
+ this.context = new Context<>();
+ }
+
+ @Override
+ public void open() throws Exception {
+ super.open();
+
+ this.currentWatermark = Long.MIN_VALUE;
+ }
+
+ @Override
+ public void initializeState(StateInitializationContext context) throws Exception {
+ super.initializeState(context);
+ sinkWriter =
+ writerFactory.apply(
+ createInitContext(), sinkWriterStateHandler.initializeState(context));
+ committerHandler.initializeState(context);
+ }
+
+ @Override
+ public void snapshotState(StateSnapshotContext context) throws Exception {
+ super.snapshotState(context);
+ sinkWriterStateHandler.snapshotState(sinkWriter::snapshotState);
+ committerHandler.snapshotState(context);
+ }
+
+ @Override
+ public void processElement(StreamRecord<InputT> element) throws Exception {
+ context.element = element;
+ sinkWriter.write(element.getValue(), context);
+ }
+
+ @Override
+ public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
+ super.prepareSnapshotPreBarrier(checkpointId);
+ emitCommittables(
+ committerHandler.processCommittables(() -> sinkWriter.prepareCommit(false)));
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ super.notifyCheckpointComplete(checkpointId);
+ emitCommittables(committerHandler.notifyCheckpointCompleted(checkpointId));
+ }
+
+ @Override
+ public void processWatermark(Watermark mark) throws Exception {
+ super.processWatermark(mark);
+ this.currentWatermark = mark.getTimestamp();
+ sinkWriter.writeWatermark(
+ new org.apache.flink.api.common.eventtime.Watermark(mark.getTimestamp()));
+ }
+
+ @Override
+ public void endInput() throws Exception {
+ emitCommittables(
+ committerHandler.processCommittables(() -> sinkWriter.prepareCommit(true)));
+ emitCommittables(committerHandler.endOfInput());
+ }
+
+ private void emitCommittables(Collection<CommT> committables) throws IOException {
+ if (committableSerializer != null) {
+ for (CommT committable : committables) {
+ output.collect(
+ new StreamRecord<>(
+ SimpleVersionedSerialization.writeVersionAndSerialize(
+ committableSerializer, committable)));
+ }
+ }
+ }
+
+ @Override
+ public void close() throws Exception {
+ committerHandler.close();
+ sinkWriter.close();
+ super.close();
+ }
+
+ private Sink.InitContext createInitContext() {
+ return new InitContextImpl(
+ getRuntimeContext(), processingTimeService, mailboxExecutor, getMetricGroup());
+ }
+
+ private class Context<IN> implements SinkWriter.Context {
+
+ private StreamRecord<IN> element;
+
+ @Override
+ public long currentWatermark() {
+ return currentWatermark;
+ }
+
+ @Override
+ public Long timestamp() {
+ if (element.hasTimestamp()) {
+ return element.getTimestamp();
+ }
+ return null;
+ }
+ }
+
+ private static class InitContextImpl implements Sink.InitContext {
+
+ private final ProcessingTimeService processingTimeService;
+
+ private final MailboxExecutor mailboxExecutor;
+
+ private final MetricGroup metricGroup;
+
+ private final StreamingRuntimeContext runtimeContext;
+
+ public InitContextImpl(
+ StreamingRuntimeContext runtimeContext,
+ ProcessingTimeService processingTimeService,
+ MailboxExecutor mailboxExecutor,
+ MetricGroup metricGroup) {
+ this.runtimeContext = checkNotNull(runtimeContext);
+ this.mailboxExecutor = checkNotNull(mailboxExecutor);
+ this.processingTimeService = checkNotNull(processingTimeService);
+ this.metricGroup = checkNotNull(metricGroup);
+ }
+
+ @Override
+ public UserCodeClassLoader getUserCodeClassLoader() {
+ return new UserCodeClassLoader() {
+ @Override
+ public ClassLoader asClassLoader() {
+ return runtimeContext.getUserCodeClassLoader();
+ }
+
+ @Override
+ public void registerReleaseHookIfAbsent(
+ String releaseHookName, Runnable releaseHook) {
+ runtimeContext.registerUserCodeClassLoaderReleaseHookIfAbsent(
+ releaseHookName, releaseHook);
+ }
+ };
+ }
+
+ @Override
+ public int getNumberOfParallelSubtasks() {
+ return runtimeContext.getNumberOfParallelSubtasks();
+ }
+
+ @Override
+ public MailboxExecutor getMailboxExecutor() {
+ return mailboxExecutor;
+ }
+
+ @Override
+ public Sink.ProcessingTimeService getProcessingTimeService() {
+ return new ProcessingTimerServiceImpl(processingTimeService);
+ }
+
+ @Override
+ public int getSubtaskId() {
+ return runtimeContext.getIndexOfThisSubtask();
+ }
+
+ @Override
+ public MetricGroup metricGroup() {
+ return metricGroup;
+ }
+ }
+
+ private static class ProcessingTimerServiceImpl implements Sink.ProcessingTimeService {
+ private final ProcessingTimeService processingTimeService;
+
+ public ProcessingTimerServiceImpl(ProcessingTimeService processingTimeService) {
+ this.processingTimeService = checkNotNull(processingTimeService);
+ }
+
+ @Override
+ public long getCurrentProcessingTime() {
+ return processingTimeService.getCurrentProcessingTime();
+ }
+
+ @Override
+ public void registerProcessingTimer(
+ long time,
+ Sink.ProcessingTimeService.ProcessingTimeCallback processingTimerCallback) {
+ checkNotNull(processingTimerCallback);
+ processingTimeService.registerTimer(time, processingTimerCallback::onProcessingTime);
+ }
+ }
+}
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
new file mode 100644
index 0000000..a76285e
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterStateHandler.java
@@ -0,0 +1,43 @@
+/*
+ * 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.runtime.state.StateInitializationContext;
+import org.apache.flink.util.function.SupplierWithException;
+
+import java.io.Serializable;
+import java.util.List;
+
+/**
+ * Manages the state of a {@link org.apache.flink.api.connector.sink.SinkWriter}. There are only two
+ * flavors: stateless handled by {@link StatelessSinkWriterStateHandler} and stateful handled with
+ * {@link StatefulSinkWriterStateHandler}.
+ *
+ * @param <WriterStateT>
+ */
+interface SinkWriterStateHandler<WriterStateT> extends Serializable {
+ /**
+ * Extracts the writer state from the {@link StateInitializationContext}. The state will be used
+ * to create the writer.
+ */
+ List<WriterStateT> initializeState(StateInitializationContext context) throws Exception;
+
+ /** Stores the state of the supplier. The supplier should only be queried once. */
+ void snapshotState(SupplierWithException<List<WriterStateT>, Exception> stateSupplier)
+ throws Exception;
+}
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
new file mode 100644
index 0000000..e5925e5
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StatelessSinkWriterStateHandler.java
@@ -0,0 +1,43 @@
+/*
+ * 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.runtime.state.StateInitializationContext;
+import org.apache.flink.util.function.SupplierWithException;
+
+import java.util.Collections;
+import java.util.List;
+
+/** {@link SinkWriterStateHandler} for stateless sinks. */
+enum StatelessSinkWriterStateHandler implements SinkWriterStateHandler<Object> {
+ INSTANCE;
+
+ @SuppressWarnings("unchecked")
+ static <WriterStateT> SinkWriterStateHandler<WriterStateT> getInstance() {
+ return (SinkWriterStateHandler<WriterStateT>) StatelessSinkWriterStateHandler.INSTANCE;
+ }
+
+ @Override
+ public List<Object> initializeState(StateInitializationContext context) throws Exception {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public void snapshotState(SupplierWithException<List<Object>, Exception> stateSupplier)
+ throws Exception {}
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterOperator.java
index 8ab1dc1..d9c1d14 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingCommitterOperator.java
@@ -71,5 +71,6 @@ final class StreamingCommitterOperator<CommT>
public void close() throws Exception {
super.close();
committer.close();
+ super.close();
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingGlobalCommitterOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingGlobalCommitterOperator.java
index 3cf5193..1d1f2447 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingGlobalCommitterOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/StreamingGlobalCommitterOperator.java
@@ -97,6 +97,7 @@ public final class StreamingGlobalCommitterOperator<CommT, GlobalCommT>
public void close() throws Exception {
super.close();
globalCommitter.close();
+ super.close();
}
@Override