You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2021/08/04 08:46:29 UTC
[GitHub] [flink] AHeise opened a new pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
AHeise opened a new pull request #16701:
URL: https://github.com/apache/flink/pull/16701
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
FLINK-19531 introduced the implementation of the Sink interface. It strictly cut the different parts of the sink pipeline into 3 operators:
writer -> committer -> global committer
In streaming mode with a parallelism p, the pipeline is executed as follows
writer(parallelism=p) -> committer(parallelism=p) -> global committer(parallelism=1).
Here we could bundle writer+committer into one operator.
In batch mode with a parallelism p, the pipeline is executed as follows
writer(parallelism=p) -> committer(parallelism=1) -> global committer(parallelism=1).
Here we could bundle committer+global committer into one operator. (Committer needs to run with parallelism=1 to create a pipeline region and reduce the risk of dataloss during commit; we can hopefully fix it after FLIP-147)
Having fewer operators will decrease the need to copy the committables in the operator chain (where we currently mostly use Kryo). Thus, we can implement connection/transaction pooling in streaming, where committables are reused after successful commit.
The proposal of this ticket is to extract the functionality of the 7 different operator implementations with their factories into reusable building blocks and use them in 2 operators (writer+committer and committer+global committer).
## Brief change log
- Added 2 interfaces for managing writer state and handling committables
- Implemented building blocks with these two interfaces for streaming/batch (global) committers
- Added 2 operators that use the building blocks to provide writer+committer and committer+global committer
## Verifying this change
Adjusted the existing unit tests. FileSink ITCases also cover this refactoring.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (**yes** / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (**yes** / no / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 6ba61de2679a13b311663e3d5476a4a60d81f793 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683172137
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -48,41 +54,70 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
Review comment:
Is this correct? AFAIK the return of `Committer#commit()` is only used to retry these committables later. Unfortunately, this is not implemented yet, so it will throw once the list is not empty.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
We can also move these to methods to `AbstractStreamingCommitterHandler` because AFAIS they are not necessary for the batch committers.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.io.IOException;
+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>, IOException> committableSupplier)
+ throws IOException {
+ this.committables.addAll(committableSupplier.get());
+ return Collections.emptyList();
+ }
+
+ public List<InputT> pollCommittables() {
Review comment:
```suggestion
protected List<InputT> pollCommittables() {
```
?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
Review comment:
Nit: If the default implementation is only for `NoopCommitterHandler` we can think of just implementing it there.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Out of curiosity why does the `StreamingCommitterHandler` not implement `endOfInput`?
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
##########
@@ -139,6 +139,7 @@ public static void assertNoLateRecords(Iterable<Object> elements) {
testHarness.processElements(
input.stream().map(StreamRecord::new).collect(Collectors.toList()));
+ testHarness.prepareSnapshotPreBarrier(1);
Review comment:
Nit: Why was this changed needed?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@link org.apache.flink.api.connector.sink.Committer} or {@link
Review comment:
Nit: I find the word `multiple` a bit confusing here because afait it always exactly wraps one instance of a committer.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -46,44 +54,81 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
+ * committables are passed to this {@link GlobalCommitter} of which only a single instance
+ * exists. If a global committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()} and {@link #getGlobalCommittableSerializer()}.
*
- * @return A global committer.
- * @throws IOException if fail to create a global committer.
+ * @return A global committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter() throws IOException;
- /** Returns the serializer of the committable type. */
+ /**
+ * Returns the serializer of the committable type. The serializer is required iff the sink has a
+ * {@link Committer} or {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();
- /** Returns the serializer of the aggregated committable type. */
+ /**
+ * Returns the serializer of the aggregated committable type. The serializer is required iff the
+ * sink has a {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();
- /** Return the serializer of the writer's state type. */
- Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
Review comment:
Nit: Maybe use a docstring to reference the example file sinks. The terms old and new might be forgotten in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
* 8872518be16de80be32bd7880caf93d95b5e127e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501)
* 516ef37858f0773d94b01f1e407ae951149e3fbf UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683643569
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
##########
@@ -49,10 +49,10 @@
* @param <InputT> The input type of the {@link Committer}.
* @param <CommT> The committable type of the {@link Committer}.
*/
-abstract class AbstractStreamingCommitterOperator<InputT, CommT>
- extends AbstractStreamOperator<CommT> implements OneInputStreamOperator<InputT, CommT> {
-
- private static final long serialVersionUID = 1L;
Review comment:
Is it safe to remove this line?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -32,25 +34,42 @@
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;
/**
- * Abstract base class for operators that work with a {@link SinkWriter}.
+ * 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:
*
- * <p>Sub-classes are responsible for creating the specific {@link SinkWriter} by implementing
- * {@link #createWriter()}.
+ * <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 input type of the {@link SinkWriter}.
- * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @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
*/
-@Internal
-abstract class AbstractSinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<InputT, CommT>, BoundedOneInput {
-
- private static final long serialVersionUID = 1L;
+public class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<byte[]>
Review comment:
`@Internal`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+import org.apache.flink.streaming.api.operators.YieldingOperatorFactory;
+
+import java.util.Optional;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * SinkOperator}.
+ *
+ * @param <InputT> The input type of the {@link SinkWriter}.
+ * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @param <WriterStateT> The type of the {@link SinkWriter Writer's} state.
+ */
Review comment:
`@Internal`
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
Review comment:
Add `@Internal` annotation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892483638
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit cf728118c15e3588b9d457aebec20af161859a58 (Sat Aug 28 13:10:35 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682588404
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
I'll remove it. Originally, `CommitterHandler` was public and the composition happened in the SinkTranslater.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
}, {
"hash" : "cf728118c15e3588b9d457aebec20af161859a58",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21641",
"triggerID" : "cf728118c15e3588b9d457aebec20af161859a58",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* cf728118c15e3588b9d457aebec20af161859a58 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21641)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683223090
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
We can also move these two methods to `AbstractStreamingCommitterHandler` because AFAIS they are not necessary for the batch committers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683292583
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -46,44 +54,81 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
+ * committables are passed to this {@link GlobalCommitter} of which only a single instance
+ * exists. If a global committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()} and {@link #getGlobalCommittableSerializer()}.
*
- * @return A global committer.
- * @throws IOException if fail to create a global committer.
+ * @return A global committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter() throws IOException;
- /** Returns the serializer of the committable type. */
+ /**
+ * Returns the serializer of the committable type. The serializer is required iff the sink has a
+ * {@link Committer} or {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();
- /** Returns the serializer of the aggregated committable type. */
+ /**
+ * Returns the serializer of the aggregated committable type. The serializer is required iff the
+ * sink has a {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();
- /** Return the serializer of the writer's state type. */
- Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
Review comment:
I can't exactly use a JavaDoc link because of circular dependency issues. But I have used the proper class names in the code tag to make it more obvious.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
* 8872518be16de80be32bd7880caf93d95b5e127e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501)
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892880932
> I think we should also adapt the docstrings in the `Sink.java` to reflect the new invariants when user needs to provide Committable serializer
I have improved the doc and would merge this new commit with the first commit later. Could you please check & improve?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555)
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682601615
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
I do not necessarily see the relation to OOP here ;) but isn't it also some kind of optimization to not instantiate this Noop object?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682587311
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -81,9 +83,17 @@
/** Return the serializer of the writer's state type. */
Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
+ * when resuming from a checkpoint/savepoint.
+ */
+ default Collection<String> getCompatibleStateNames() {
Review comment:
I consider it more of a bugfix ;)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682594076
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
You are supposed to close resources always in the opposite order of creation. That's also how try-with-resource and Guava's Closer work. The main idea is that your later resources may depend on the prior resources to finish properly.
Here, it doesn't matter, but I'd still like to have it consistently correct in Flink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682543690
##########
File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
##########
@@ -161,6 +163,12 @@ private FileSink(BucketsBuilder<IN, ? extends BucketsBuilder<IN, ?>> bucketsBuil
return Optional.empty();
}
+ @Override
+ public Collection<String> getCompatibleStateNames() {
Review comment:
Personally, I am not a big fan of the `Collection` abstraction. Is this commonly favored in Flink over exposing some more information and using a `List`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
Wouldn't is be easier to just not instantiate a `GlobalBatchCommitter` if not necessary and not introduce the `NoopCommitterHandler`.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.util.Optional;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * SinkOperator}.
+ *
+ * @param <InputT> The input type of the {@link SinkWriter}.
+ * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @param <WriterStateT> The type of the {@link SinkWriter Writer's} state.
+ */
+public final class SinkOperatorFactory<InputT, CommT, WriterStateT>
Review comment:
I wonder if this class should be part of the previous commit IMO it belongs to the newly introduced operator model.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be either directly transforming them or by adding them to the
Review comment:
`by either directly ...`
FYI I seems Github suggestions are broken ...
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
+ !(committerHandler instanceof NoopCommitterHandler),
+ "committer operator without commmitter");
+ final CommitterOperator<CommT, GlobalCommT> committerOperator =
+ new CommitterOperator<>(committableSerializer, committerHandler, null);
Review comment:
I prefer having a separate constructor when fields are nullable.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -81,9 +83,17 @@
/** Return the serializer of the writer's state type. */
Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
+ * when resuming from a checkpoint/savepoint.
+ */
+ default Collection<String> getCompatibleStateNames() {
Review comment:
Nit: Do we need a FLIP for this change?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -0,0 +1,265 @@
+/*
+ * 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.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.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+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 a this operator with parallelism p containing {@link
Review comment:
Remove `a`
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
Why is this method needed?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
Review comment:
Isn't this always failing for batch mode because the batch mode set the committerHandler to `BatchCommitterHandler`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
Why do we need this change? I am not fully familiar with the exact semantics but first closing the object itself and next the `globalCommitter` looks safer.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
I do not necessarily see the relation to OOP here ;) but isn't it also some kind of optimization to not instantiate this Noop object?
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -48,41 +54,70 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
Review comment:
Is this correct? AFAIK the return of `Committer#commit()` is only used to retry these committables later. Unfortunately, this is not implemented yet, so it will throw once the list is not empty.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
We can also move these to methods to `AbstractStreamingCommitterHandler` because AFAIS they are not necessary for the batch committers.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractCommitterHandler.java
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.io.IOException;
+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>, IOException> committableSupplier)
+ throws IOException {
+ this.committables.addAll(committableSupplier.get());
+ return Collections.emptyList();
+ }
+
+ public List<InputT> pollCommittables() {
Review comment:
```suggestion
protected List<InputT> pollCommittables() {
```
?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
Review comment:
Nit: If the default implementation is only for `NoopCommitterHandler` we can think of just implementing it there.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Out of curiosity why does the `StreamingCommitterHandler` not implement `endOfInput`?
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
##########
@@ -139,6 +139,7 @@ public static void assertNoLateRecords(Iterable<Object> elements) {
testHarness.processElements(
input.stream().map(StreamRecord::new).collect(Collectors.toList()));
+ testHarness.prepareSnapshotPreBarrier(1);
Review comment:
Nit: Why was this changed needed?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@link org.apache.flink.api.connector.sink.Committer} or {@link
Review comment:
Nit: I find the word `multiple` a bit confusing here because afait it always exactly wraps one instance of a committer.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -46,44 +54,81 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
+ * committables are passed to this {@link GlobalCommitter} of which only a single instance
+ * exists. If a global committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()} and {@link #getGlobalCommittableSerializer()}.
*
- * @return A global committer.
- * @throws IOException if fail to create a global committer.
+ * @return A global committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter() throws IOException;
- /** Returns the serializer of the committable type. */
+ /**
+ * Returns the serializer of the committable type. The serializer is required iff the sink has a
+ * {@link Committer} or {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();
- /** Returns the serializer of the aggregated committable type. */
+ /**
+ * Returns the serializer of the aggregated committable type. The serializer is required iff the
+ * sink has a {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();
- /** Return the serializer of the writer's state type. */
- Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
Review comment:
Nit: Maybe use a docstring to reference the example file sinks. The terms old and new might be forgotten in the future.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
We can also move these two methods to `AbstractStreamingCommitterHandler` because AFAIS they are not necessary for the batch committers.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
I am just a bit unhappy with the overall structure of having this kind of deep inheritance but I guess refactoring this is probably out of scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682584648
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.util.Optional;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * SinkOperator}.
+ *
+ * @param <InputT> The input type of the {@link SinkWriter}.
+ * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @param <WriterStateT> The type of the {@link SinkWriter Writer's} state.
+ */
+public final class SinkOperatorFactory<InputT, CommT, WriterStateT>
Review comment:
I agree but this class depends on the specific `CommitterHandler`s which I wanted to convert in this commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683680355
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
##########
@@ -49,10 +49,10 @@
* @param <InputT> The input type of the {@link Committer}.
* @param <CommT> The committable type of the {@link Committer}.
*/
-abstract class AbstractStreamingCommitterOperator<InputT, CommT>
- extends AbstractStreamOperator<CommT> implements OneInputStreamOperator<InputT, CommT> {
-
- private static final long serialVersionUID = 1L;
Review comment:
You mean `serialVersionUID`? With `StreamOperatorFactory` (which are serializable) there is actually no need for operators to be serializable anymore and it will be removed in the long run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
* 8872518be16de80be32bd7880caf93d95b5e127e UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
}, {
"hash" : "cf728118c15e3588b9d457aebec20af161859a58",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "cf728118c15e3588b9d457aebec20af161859a58",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 6ba61de2679a13b311663e3d5476a4a60d81f793 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614)
* cf728118c15e3588b9d457aebec20af161859a58 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683683906
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/AbstractStreamingCommitterHandler.java
##########
@@ -49,10 +49,10 @@
* @param <InputT> The input type of the {@link Committer}.
* @param <CommT> The committable type of the {@link Committer}.
*/
-abstract class AbstractStreamingCommitterOperator<InputT, CommT>
- extends AbstractStreamOperator<CommT> implements OneInputStreamOperator<InputT, CommT> {
-
- private static final long serialVersionUID = 1L;
Review comment:
Note that factories should have `serialVersionUID` if we are very strict. However, actually in non-persistent serialization none of it actually matters. `serialVersionUID` is only important if you need a way to read old serialized data into changed serializable classes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683320251
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
I am just a bit unhappy with the overall structure of having this kind of deep inheritance but I guess refactoring this is probably out of scope.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555)
* 615e76371b3d251b16d530dc8b31b5c27d143122 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682584648
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.util.Optional;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * SinkOperator}.
+ *
+ * @param <InputT> The input type of the {@link SinkWriter}.
+ * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @param <WriterStateT> The type of the {@link SinkWriter Writer's} state.
+ */
+public final class SinkOperatorFactory<InputT, CommT, WriterStateT>
Review comment:
I agree but this class depends on the specific `CommitterHandler`s which I wanted to convert in this commit.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
The `NoopCommitterHandler` avoids null checks inside this class. It's a matter of flavor but in general a `NoopCommitterHandler` is the proper OOP approach (also dubbed case-less programming because of VMT).
##########
File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
##########
@@ -161,6 +163,12 @@ private FileSink(BucketsBuilder<IN, ? extends BucketsBuilder<IN, ?>> bucketsBuil
return Optional.empty();
}
+ @Override
+ public Collection<String> getCompatibleStateNames() {
Review comment:
I don't understand your comment. Why would you favor a `List` here? The only difference between a collection and a list is that the latter is implicitly ordered and indexed. None of that matters here.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -81,9 +83,17 @@
/** Return the serializer of the writer's state type. */
Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
+ * when resuming from a checkpoint/savepoint.
+ */
+ default Collection<String> getCompatibleStateNames() {
Review comment:
I consider it more of a bugfix ;)
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
I'll remove it. Originally, `CommitterHandler` was public and the composition happened in the SinkTranslater.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
Review comment:
It's always succeeding (`Precondition#checkState` fails on `false`) in your case. However, it could fail if there is no global committer in streaming and it fails if there is neither a committer nor a global committer in batch. But this is a programmatic error in `SinkTranslater`; hence the `checkState`.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
You are supposed to close resources always in the opposite order of creation. That's also how try-with-resource and Guava's Closer work. The main idea is that your later resources may depend on the prior resources to finish properly.
Here, it doesn't matter, but I'd still like to have it consistently correct in Flink.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
You are supposed to close resources always in the opposite order of creation. That's also how try-with-resource and Guava's Closer work. The main idea is that your later resources may depend on the prior resources to finish properly. (Think of a transaction that flushes, so you'd close the transaction before the client).
Here, it doesn't matter, but I'd still like to have it consistently correct in Flink.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
Here is some reference https://en.wikipedia.org/wiki/Null_object_pattern TL;DR `null` is an external construct in OOP.
Also don't worry about performance here: First, we have a singleton pattern as an enum that's very efficient. Second, we are not on the hot path.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
I can't fully remove it without introducing some ugliness: This method is for casting the singleton into the expected type. The `NoopCommitterHandler.INSTANCE` could actually be covariant to any type but you can't express that in Java.
I have moved it to `NoopCommitterHandler#getInstance()`. PTAL if it's less confusing. I'm also promoting `NoopCommitterHandler` to top-level.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Good questions. I double-checked and EOI on streaming is really just meant for flushing data. A final checkpoint happens afterwards (especially after FLIP-147).
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Good questions. I double-checked and EOI on streaming is really just meant for flushing data. A final checkpoint happens afterwards (especially after FLIP-147). Most likely we don't need that for Batch after FLIP-147 but that's a future task.
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
##########
@@ -139,6 +139,7 @@ public static void assertNoLateRecords(Iterable<Object> elements) {
testHarness.processElements(
input.stream().map(StreamRecord::new).collect(Collectors.toList()));
+ testHarness.prepareSnapshotPreBarrier(1);
Review comment:
Since writer+committer are merged, we now need to simulate more of the actual calls. The writer creates the committables on `prepareSnapshotPreBarrier` and pass it to the committer which then does it's actual work on `snapshotState`.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
They are called for all `CommitterHandler` though so we need to leave them here. Or we have to do instanceof in the operators.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -46,44 +54,81 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
+ * committables are passed to this {@link GlobalCommitter} of which only a single instance
+ * exists. If a global committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()} and {@link #getGlobalCommittableSerializer()}.
*
- * @return A global committer.
- * @throws IOException if fail to create a global committer.
+ * @return A global committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<GlobalCommitter<CommT, GlobalCommT>> createGlobalCommitter() throws IOException;
- /** Returns the serializer of the committable type. */
+ /**
+ * Returns the serializer of the committable type. The serializer is required iff the sink has a
+ * {@link Committer} or {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<CommT>> getCommittableSerializer();
- /** Returns the serializer of the aggregated committable type. */
+ /**
+ * Returns the serializer of the aggregated committable type. The serializer is required iff the
+ * sink has a {@link GlobalCommitter}.
+ */
Optional<SimpleVersionedSerializer<GlobalCommT>> getGlobalCommittableSerializer();
- /** Return the serializer of the writer's state type. */
- Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
Review comment:
I can't exactly use a JavaDoc link because of circular dependency issues. But I have used the proper class names in the code tag to make it more obvious.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -48,41 +54,70 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
Review comment:
You are right. Thanks for double-checking!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682658272
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
Here is some reference https://en.wikipedia.org/wiki/Null_object_pattern TL;DR `null` is an external construct in OOP.
Also don't worry about performance here: First, we have a singleton pattern as an enum that's very efficient. Second, we are not on the hot path.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
* 8872518be16de80be32bd7880caf93d95b5e127e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555)
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682594076
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
You are supposed to close resources always in the opposite order of creation. That's also how try-with-resource and Guava's Closer work. The main idea is that your later resources may depend on the prior resources to finish properly. (Think of a transaction that flushes, so you'd close the transaction before the client).
Here, it doesn't matter, but I'd still like to have it consistently correct in Flink.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
* 9b9975e4fbcf62c2f72df21dfdbe32695cec43db Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602)
* 6ba61de2679a13b311663e3d5476a4a60d81f793 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 9b9975e4fbcf62c2f72df21dfdbe32695cec43db Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602)
* 6ba61de2679a13b311663e3d5476a4a60d81f793 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] fapaul commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
fapaul commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682543690
##########
File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
##########
@@ -161,6 +163,12 @@ private FileSink(BucketsBuilder<IN, ? extends BucketsBuilder<IN, ?>> bucketsBuil
return Optional.empty();
}
+ @Override
+ public Collection<String> getCompatibleStateNames() {
Review comment:
Personally, I am not a big fan of the `Collection` abstraction. Is this commonly favored in Flink over exposing some more information and using a `List`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
Wouldn't is be easier to just not instantiate a `GlobalBatchCommitter` if not necessary and not introduce the `NoopCommitterHandler`.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperatorFactory.java
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.api.connector.sink.SinkWriter;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.util.Optional;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * SinkOperator}.
+ *
+ * @param <InputT> The input type of the {@link SinkWriter}.
+ * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @param <WriterStateT> The type of the {@link SinkWriter Writer's} state.
+ */
+public final class SinkOperatorFactory<InputT, CommT, WriterStateT>
Review comment:
I wonder if this class should be part of the previous commit IMO it belongs to the newly introduced operator model.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be either directly transforming them or by adding them to the
Review comment:
`by either directly ...`
FYI I seems Github suggestions are broken ...
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
+ !(committerHandler instanceof NoopCommitterHandler),
+ "committer operator without commmitter");
+ final CommitterOperator<CommT, GlobalCommT> committerOperator =
+ new CommitterOperator<>(committableSerializer, committerHandler, null);
Review comment:
I prefer having a separate constructor when fields are nullable.
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -81,9 +83,17 @@
/** Return the serializer of the writer's state type. */
Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+ /**
+ * A list of state names of sinks from which the state can be restored. For example, the new
+ * file sink can resume from the state of an old streaming file sink as a drop-in replacement
+ * when resuming from a checkpoint/savepoint.
+ */
+ default Collection<String> getCompatibleStateNames() {
Review comment:
Nit: Do we need a FLIP for this change?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -0,0 +1,265 @@
+/*
+ * 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.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.watermark.Watermark;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+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 a this operator with parallelism p containing {@link
Review comment:
Remove `a`
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
Why is this method needed?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
Review comment:
Isn't this always failing for batch mode because the batch mode set the committerHandler to `BatchCommitterHandler`?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/GlobalBatchCommitterHandler.java
##########
@@ -72,11 +58,12 @@ public void endInput() throws Exception {
}
}
globalCommitter.endOfInput();
+ return Collections.emptyList();
}
@Override
public void close() throws Exception {
- super.close();
globalCommitter.close();
+ super.close();
Review comment:
Why do we need this change? I am not fully familiar with the exact semantics but first closing the object itself and next the `globalCommitter` looks safer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682586865
##########
File path: flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/FileSink.java
##########
@@ -161,6 +163,12 @@ private FileSink(BucketsBuilder<IN, ? extends BucketsBuilder<IN, ?>> bucketsBuil
return Optional.empty();
}
+ @Override
+ public Collection<String> getCompatibleStateNames() {
Review comment:
I don't understand your comment. Why would you favor a `List` here? The only difference between a collection and a list is that the latter is implicitly ordered and indexed. None of that matters here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683282258
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Good questions. I double-checked and EOI on streaming is really just meant for flushing data. A final checkpoint happens afterwards (especially after FLIP-147). Most likely we don't need that for Batch after FLIP-147 but that's a future task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
* 8872518be16de80be32bd7880caf93d95b5e127e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501)
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise merged pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise merged pull request #16701:
URL: https://github.com/apache/flink/pull/16701
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683282258
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
Review comment:
Good questions. I double-checked and EOI on streaming is really just meant for flushing data. A final checkpoint happens afterwards (especially after FLIP-147).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683684529
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkOperator.java
##########
@@ -32,25 +34,42 @@
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;
/**
- * Abstract base class for operators that work with a {@link SinkWriter}.
+ * 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:
*
- * <p>Sub-classes are responsible for creating the specific {@link SinkWriter} by implementing
- * {@link #createWriter()}.
+ * <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 input type of the {@link SinkWriter}.
- * @param <CommT> The committable type of the {@link SinkWriter}.
+ * @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
*/
-@Internal
-abstract class AbstractSinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<InputT, CommT>, BoundedOneInput {
-
- private static final long serialVersionUID = 1L;
+public class SinkOperator<InputT, CommT, WriterStateT> extends AbstractStreamOperator<byte[]>
Review comment:
I made it non-public instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892483638
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 1c542d18e2d669baa8d4822ed291a96450aff330 (Wed Aug 04 08:50:39 UTC 2021)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683288949
##########
File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/util/TestHarnessUtil.java
##########
@@ -139,6 +139,7 @@ public static void assertNoLateRecords(Iterable<Object> elements) {
testHarness.processElements(
input.stream().map(StreamRecord::new).collect(Collectors.toList()));
+ testHarness.prepareSnapshotPreBarrier(1);
Review comment:
Since writer+committer are merged, we now need to simulate more of the actual calls. The writer creates the committables on `prepareSnapshotPreBarrier` and pass it to the committer which then does it's actual work on `snapshotState`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682836051
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.java
##########
@@ -0,0 +1,88 @@
+/*
+ * 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 multiple {@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 be 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>, IOException> committableSupplier)
+ throws IOException;
+
+ /**
+ * 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 {
+ return Collections.emptyList();
+ }
+
+ default void close() throws Exception {}
+
+ /** Called when a checkpoint is completed and returns a list of output to be sent downstream. */
+ default Collection<OutputT> notifyCheckpointCompleted(long checkpointId) throws IOException {
+ return Collections.emptyList();
+ }
+
+ /** Swallows all committables and emits nothing. */
+ @SuppressWarnings("unchecked")
+ static <InputT, OutputT> CommitterHandler<InputT, OutputT> noop() {
Review comment:
I can't fully remove it without introducing some ugliness: This method is for casting the singleton into the expected type. The `NoopCommitterHandler.INSTANCE` could actually be covariant to any type but you can't express that in Java.
I have moved it to `NoopCommitterHandler#getInstance()`. PTAL if it's less confusing. I'm also promoting `NoopCommitterHandler` to top-level.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8872518be16de80be32bd7880caf93d95b5e127e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501)
* 516ef37858f0773d94b01f1e407ae951149e3fbf Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892880932
> I think we should also adapt the docstrings in the `Sink.java` to reflect the new invariants when user needs to provide Committable serializer
I have improved the doc and would merge this new commit with the first commit later. Could you please check & improve?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
* 9b9975e4fbcf62c2f72df21dfdbe32695cec43db UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
}, {
"hash" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614",
"triggerID" : "6ba61de2679a13b311663e3d5476a4a60d81f793",
"triggerType" : "PUSH"
}, {
"hash" : "cf728118c15e3588b9d457aebec20af161859a58",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21641",
"triggerID" : "cf728118c15e3588b9d457aebec20af161859a58",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 6ba61de2679a13b311663e3d5476a4a60d81f793 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21614)
* cf728118c15e3588b9d457aebec20af161859a58 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21641)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682585866
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/BatchCommitterHandler.java
##########
@@ -33,44 +30,52 @@
* Runtime {@link org.apache.flink.streaming.api.operators.StreamOperator} for executing {@link
* Committer} in the batch execution mode.
*
- * @param <CommT> The committable type of the {@link Committer}.
+ * @param <InputT> The committable type of the {@link Committer}.
*/
-final class BatchCommitterOperator<CommT> extends AbstractStreamOperator<CommT>
- implements OneInputStreamOperator<CommT, CommT>, BoundedOneInput {
+final class BatchCommitterHandler<InputT, OutputT>
+ extends AbstractCommitterHandler<InputT, OutputT> {
/** Responsible for committing the committable to the external system. */
- private final Committer<CommT> committer;
+ private final Committer<InputT> committer;
- /** Record all the committables until the end of the input. */
- private final List<CommT> allCommittables;
+ /**
+ * The committer that is chained to this committer. It's either {@link
+ * GlobalBatchCommitterHandler} or {@link NoopCommitterHandler}.
Review comment:
The `NoopCommitterHandler` avoids null checks inside this class. It's a matter of flavor but in general a `NoopCommitterHandler` is the proper OOP approach (also dubbed case-less programming because of VMT).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r682590569
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperatorFactory.java
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.sink.Committer;
+import org.apache.flink.api.connector.sink.GlobalCommitter;
+import org.apache.flink.api.connector.sink.Sink;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A {@link org.apache.flink.streaming.api.operators.StreamOperatorFactory} for {@link
+ * CommitterOperator}.
+ *
+ * @param <CommT> the type of the committable
+ * @param <GlobalCommT> the type of the global committable
+ */
+public final class CommitterOperatorFactory<CommT, GlobalCommT>
+ extends AbstractStreamOperatorFactory<byte[]>
+ implements OneInputStreamOperatorFactory<byte[], byte[]> {
+
+ private final Sink<?, CommT, ?, GlobalCommT> sink;
+ private final boolean batch;
+
+ public CommitterOperatorFactory(Sink<?, CommT, ?, GlobalCommT> sink, boolean batch) {
+ this.sink = sink;
+ this.batch = batch;
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends StreamOperator<byte[]>> T createStreamOperator(
+ StreamOperatorParameters<byte[]> parameters) {
+
+ SimpleVersionedSerializer<CommT> committableSerializer =
+ sink.getCommittableSerializer().orElseThrow(this::noSerializerFound);
+ try {
+ CommitterHandler<CommT, GlobalCommT> committerHandler = getGlobalCommitterHandler();
+ if (batch) {
+ Optional<Committer<CommT>> committer = sink.createCommitter();
+ if (committer.isPresent()) {
+ committerHandler =
+ new BatchCommitterHandler<>(committer.get(), committerHandler);
+ }
+ }
+
+ checkState(
Review comment:
It's always succeeding (`Precondition#checkState` fails on `false`) in your case. However, it could fail if there is no global committer in streaming and it fails if there is neither a committer nor a global committer in batch. But this is a programmatic error in `SinkTranslater`; hence the `checkState`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1c542d18e2d669baa8d4822ed291a96450aff330 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683290799
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterHandler.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.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 multiple {@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 {}
Review comment:
They are called for all `CommitterHandler` though so we need to leave them here. Or we have to do instanceof in the operators.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 1f5428b5a9d8a0656cb85e1d9ab090af13aa2533 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552)
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* a8ac5bff7d0e18110d4bc7821af3667734cf90b7 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #16701:
URL: https://github.com/apache/flink/pull/16701#issuecomment-892491133
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21487",
"triggerID" : "1c542d18e2d669baa8d4822ed291a96450aff330",
"triggerType" : "PUSH"
}, {
"hash" : "8872518be16de80be32bd7880caf93d95b5e127e",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21501",
"triggerID" : "8872518be16de80be32bd7880caf93d95b5e127e",
"triggerType" : "PUSH"
}, {
"hash" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21513",
"triggerID" : "516ef37858f0773d94b01f1e407ae951149e3fbf",
"triggerType" : "PUSH"
}, {
"hash" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21552",
"triggerID" : "1f5428b5a9d8a0656cb85e1d9ab090af13aa2533",
"triggerType" : "PUSH"
}, {
"hash" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "34884c0567ded09b7413aaeffd8cc4d5819c3bd3",
"triggerType" : "PUSH"
}, {
"hash" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21555",
"triggerID" : "a8ac5bff7d0e18110d4bc7821af3667734cf90b7",
"triggerType" : "PUSH"
}, {
"hash" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559",
"triggerID" : "615e76371b3d251b16d530dc8b31b5c27d143122",
"triggerType" : "PUSH"
}, {
"hash" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602",
"triggerID" : "9b9975e4fbcf62c2f72df21dfdbe32695cec43db",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 34884c0567ded09b7413aaeffd8cc4d5819c3bd3 UNKNOWN
* 615e76371b3d251b16d530dc8b31b5c27d143122 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21559)
* 9b9975e4fbcf62c2f72df21dfdbe32695cec43db Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=21602)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] AHeise commented on a change in pull request #16701: [FLINK-23617][datastream] Co-locate sink operators with same parallelism
Posted by GitBox <gi...@apache.org>.
AHeise commented on a change in pull request #16701:
URL: https://github.com/apache/flink/pull/16701#discussion_r683293393
##########
File path: flink-core/src/main/java/org/apache/flink/api/connector/sink/Sink.java
##########
@@ -48,41 +54,70 @@
public interface Sink<InputT, CommT, WriterStateT, GlobalCommT> extends Serializable {
/**
- * Create a {@link SinkWriter}.
+ * Create a {@link SinkWriter}. If the application is resumed from a checkpoint or savepoint and
+ * the sink is stateful, it will receive the corresponding state obtained with {@link
+ * SinkWriter#snapshotState()} and serialized with {@link #getWriterStateSerializer()}. If no
+ * state exists, the first existing, compatible state specified in {@link
+ * #getCompatibleStateNames()} will be loaded and passed.
*
* @param context the runtime context.
- * @param states the writer's state.
+ * @param states the writer's previous state.
* @return A sink writer.
- * @throws IOException if fail to create a writer.
+ * @throws IOException for any failure during creation.
+ * @see SinkWriter#snapshotState()
+ * @see #getWriterStateSerializer()
+ * @see #getCompatibleStateNames()
*/
SinkWriter<InputT, CommT, WriterStateT> createWriter(
InitContext context, List<WriterStateT> states) throws IOException;
/**
- * Creates a {@link Committer}.
+ * Any stateful sink needs to provide this state serializer and implement {@link
+ * SinkWriter#snapshotState()} properly. The respective state is used in {@link
+ * #createWriter(InitContext, List)} on recovery.
+ *
+ * @return the serializer of the writer's state type.
+ */
+ Optional<SimpleVersionedSerializer<WriterStateT>> getWriterStateSerializer();
+
+ /**
+ * Creates a {@link Committer} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to this committer and persisted with {@link
+ * Committer#commit(List)}. If a committer is returned, the sink must also return a {@link
+ * #getCommittableSerializer()}.
*
- * @return A committer.
- * @throws IOException if fail to create a committer.
+ * @return A committer for the 2-phase-commit protocol.
+ * @throws IOException for any failure during creation.
*/
Optional<Committer<CommT>> createCommitter() throws IOException;
/**
- * Creates a {@link GlobalCommitter}.
+ * Creates a {@link GlobalCommitter} which is part of a 2-phase-commit protocol. The {@link
+ * SinkWriter} creates committables through {@link SinkWriter#prepareCommit(boolean)} in the
+ * first phase. The committables are then passed to the Committer and persisted with {@link
+ * Committer#commit(List)} which also can return an aggregated committable. This aggregated
Review comment:
You are right. Thanks for double-checking!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org