You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "AHeise (via GitHub)" <gi...@apache.org> on 2024/09/05 14:18:31 UTC
[PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
AHeise opened a new pull request, #25292:
URL: https://github.com/apache/flink/pull/25292
<!--
*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-25920]
In case of a failure after final checkpoint, EOI is called twice. SinkWriter should ignore the second call to avoid emitting duplicate committables. This commit uses a union state to remember that EOI happened and suppress additional handling.
## Brief change log
* Improve sink test assertions
* Straighten EOI handling in CommittableCollector
* Ignore duplicate EOI in SinkWriter <- main fix
* Improve logging in committable handling of the sink
## Verifying this change
Please make sure both new and modified tests in this PR follow [the conventions for tests defined in our code quality guide](https://flink.apache.org/how-to-contribute/code-style-and-quality-common/#7-testing).
- Added integration tests that covers the fix
## 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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756485743
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +216,18 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ // endOfInputState is union state, so it's enough if one task adds something to it
Review Comment:
You are correct that this was not correct, especially also on partially completed tasks.
I changed it to regular state and let the committer handle duplicates now. The main idea is to avoid extra transactions as much as possible in the writer and if it's not possible, let the committer clean them up anyhow.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756475601
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -156,23 +156,16 @@ public void endInput() throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- if (endInput) {
- // This is the final checkpoint, all committables should be committed
- lastCompletedCheckpointId = Long.MAX_VALUE;
- } else {
- lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
- }
+ lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
}
private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
+ long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
do {
for (CheckpointCommittableManager<CommT> manager :
- committableCollector.getCheckpointCommittablesUpTo(lastCompletedCheckpointId)) {
- // wait for all committables of the current manager before submission
- boolean fullyReceived =
- !endInput && manager.getCheckpointId() == lastCompletedCheckpointId;
- commitAndEmit(manager, fullyReceived);
+ committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
Review Comment:
@fapaul please double-check why we did it in this complicated manner originally.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #25292:
URL: https://github.com/apache/flink/pull/25292#issuecomment-2356464478
@flinkbot run azure
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1763233550
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -149,15 +190,20 @@ public void processElement(StreamRecord<IN> element) throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- commitUpToCheckpoint(checkpointId);
+ if (!this.endOfInput) {
+ commitUpToCheckpoint(checkpointId);
+ }
}
@Override
public void endInput() throws Exception {
- buckets.onProcessingTime(Long.MAX_VALUE);
- helper.snapshotState(Long.MAX_VALUE);
- output.emitWatermark(new Watermark(Long.MAX_VALUE));
- commitUpToCheckpoint(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ buckets.onProcessingTime(Long.MAX_VALUE);
+ helper.snapshotState(Long.MAX_VALUE);
+ output.emitWatermark(new Watermark(Long.MAX_VALUE));
+ commitUpToCheckpoint(Long.MAX_VALUE);
Review Comment:
I thought about that but decided against it. This is not related to CommittableSummary directly, so it would feel weird to import it just for the EOI. And someone wise said that this class is deprecated anyways, so I didn't want to change too much.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758790201
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +231,15 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!endOfInput) {
+ endOfInput = true;
Review Comment:
Would it make sense to disallow `SinkWriterOperator` to `processElement` anymore after this?
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758857878
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -147,15 +146,16 @@ Collection<CommittableWithLineage<CommT>> drainFinished() {
}
CheckpointCommittableManagerImpl<CommT> merge(CheckpointCommittableManagerImpl<CommT> other) {
- checkArgument(Objects.equals(other.checkpointId, checkpointId));
+ checkArgument(other.checkpointId == checkpointId);
Review Comment:
Good catch. I'll try to move it.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756479346
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -72,18 +73,24 @@ Collection<SubtaskCommittableManager<CommT>> getSubtaskCommittableManagers() {
return subtasksCommittableManagers.values();
}
- void upsertSummary(CommittableSummary<CommT> summary) {
+ void addSummary(CommittableSummary<CommT> summary) {
Review Comment:
Changed the name as upsert imples that we always merge but we really just want to do this for EOI now.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758839320
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +231,15 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!endOfInput) {
+ endOfInput = true;
Review Comment:
Interesting idea but wouldn't we validate the framework then?
Or would you also add the assertion after recovery? That's kinda done with the `ClosedSinkWriter`.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1759078561
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -156,23 +156,16 @@ public void endInput() throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- if (endInput) {
- // This is the final checkpoint, all committables should be committed
- lastCompletedCheckpointId = Long.MAX_VALUE;
- } else {
- lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
- }
+ lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
}
private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
+ long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
do {
for (CheckpointCommittableManager<CommT> manager :
- committableCollector.getCheckpointCommittablesUpTo(lastCompletedCheckpointId)) {
- // wait for all committables of the current manager before submission
- boolean fullyReceived =
- !endInput && manager.getCheckpointId() == lastCompletedCheckpointId;
- commitAndEmit(manager, fullyReceived);
+ committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
Review Comment:
As discussed offline, fullyReceived should always be true because we always want to have complete batches. For earlier checkpoints, partially committed batches are still considered fullyReceived as long as all committables arrived at some point (`fullyReceived === (#pending + #completed + #failed == #expected)`)
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756482710
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,7 +169,19 @@ public void initializeState(StateInitializationContext context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ // use union state to ensure that rescaling works correctly
+ this.endOfInputState =
+ context.getOperatorStateStore().getUnionListState(END_OF_INPUT_STATE_DESC);
Review Comment:
We can and need to. Think about the committables inside the CommitterOperator. We absolutely need to track them. EOI just means `processElement` isn't called again and you can't call collector afterwards.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1763062367
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -123,6 +142,27 @@ public void bucketInactive(Bucket<IN, String> bucket) {
bucketCheckInterval);
currentWatermark = Long.MIN_VALUE;
+
+ // Figure out if we have seen end of input before and if we should anything downstream. We
+ // have the following
+ // cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink that didn't handle EOI
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we regularly handle EOI, potentially resulting in unnecessary .
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI.
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true,
+ // we recover from a final checkpoint (for this subtask) and can ignore another EOI
+ // else we have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are
+ // true and do the same as above. As soon as one entry is false, we regularly start
+ // the writer and potentially emit duplicate summaries if we indeed recovered from a
+ // final checkpoint.
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ List<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
Review Comment:
Nit: Can we use `List.of` here and avoid using the shaded guava dependency?
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -149,15 +190,20 @@ public void processElement(StreamRecord<IN> element) throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- commitUpToCheckpoint(checkpointId);
+ if (!this.endOfInput) {
+ commitUpToCheckpoint(checkpointId);
+ }
}
@Override
public void endInput() throws Exception {
- buckets.onProcessingTime(Long.MAX_VALUE);
- helper.snapshotState(Long.MAX_VALUE);
- output.emitWatermark(new Watermark(Long.MAX_VALUE));
- commitUpToCheckpoint(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ buckets.onProcessingTime(Long.MAX_VALUE);
+ helper.snapshotState(Long.MAX_VALUE);
+ output.emitWatermark(new Watermark(Long.MAX_VALUE));
+ commitUpToCheckpoint(Long.MAX_VALUE);
Review Comment:
Can we use the `EOI` variable instead of `Long.MAX_VALUE`.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise merged PR #25292:
URL: https://github.com/apache/flink/pull/25292
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1748074820
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,7 +169,19 @@ public void initializeState(StateInitializationContext context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ // use union state to ensure that rescaling works correctly
+ this.endOfInputState =
+ context.getOperatorStateStore().getUnionListState(END_OF_INPUT_STATE_DESC);
Review Comment:
I need help understanding using the state.
I would expect that after receiving EOI we can also not persist anything anymore.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -249,13 +276,18 @@ private void emit(
committables.size(),
0)));
for (CommT committable : committables) {
- output.collect(
+ emit(
new StreamRecord<>(
new CommittableWithLineage<>(
committable, checkpointId, indexOfThisSubtask)));
}
}
+ private void emit(StreamRecord<CommittableMessage<CommT>> message) {
+ output.collect(message);
+ LOG.debug("Sending message to committer: {}", message);
Review Comment:
Nit: Either move the statement before sending or please do a slight rename.
```suggestion
LOG.debug("Sent message to committer: {}", message);
```
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758859753
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -178,7 +214,7 @@ public void processElement(StreamRecord<InputT> element) throws Exception {
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
super.prepareSnapshotPreBarrier(checkpointId);
- if (!endOfInput) {
+ if (!this.endOfInput) {
Review Comment:
Yes, I'll remove.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758856122
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java:
##########
@@ -91,13 +90,13 @@ public CommittableMessage<CommT> deserialize(int version, byte[] serialized)
return new CommittableWithLineage<>(
SimpleVersionedSerialization.readVersionAndDeSerialize(
committableSerializer, in),
- readCheckpointId(in),
+ in.readLong(),
Review Comment:
Thanks for challenging that. It's one of the parts where most uncertainty still resides. But let's look again on what is happening and has happened:
* On write: we replaced null with EOI, now we get EOI and write it; so it should be the same byte sequence.
* On read: we have always used `readLong` which is not null-aware. We have replaced EOI with null.
So for serialization nothing has changed. However, around serialization we replace null with EOI in all instances of the Message.
That should be safe to change without migration, no?
For compatibility, I left getCheckpointId the same, so it will return an OptionalLong.empty() on EOI where it previously returned it on null. I have not found any usage of the method outside of Flink anyhow.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758844552
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java:
##########
@@ -138,15 +136,15 @@ public void processElement(StreamRecord<CompactorRequest> element) throws Except
@Override
public void endInput() throws Exception {
// add collecting requests into the final snapshot
- checkpointRequests.put(Long.MAX_VALUE, collectingRequests);
+ checkpointRequests.put(CommittableMessage.EOI, collectingRequests);
collectingRequests = new ArrayList<>();
// submit all requests and wait until they are done
- submitUntil(Long.MAX_VALUE);
+ submitUntil(CommittableMessage.EOI);
assert checkpointRequests.isEmpty();
getAllTasksFuture().join();
- emitCompacted(null);
Review Comment:
Afaik yes. We pass the param directly to both `CommittableMessage`s. The serializer replaces `null` with `EOI`. So it effectively results in the same bytes.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1763232011
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -123,6 +142,27 @@ public void bucketInactive(Bucket<IN, String> bucket) {
bucketCheckInterval);
currentWatermark = Long.MIN_VALUE;
+
+ // Figure out if we have seen end of input before and if we should anything downstream. We
+ // have the following
+ // cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink that didn't handle EOI
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we regularly handle EOI, potentially resulting in unnecessary .
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI.
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true,
+ // we recover from a final checkpoint (for this subtask) and can ignore another EOI
+ // else we have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are
+ // true and do the same as above. As soon as one entry is false, we regularly start
+ // the writer and potentially emit duplicate summaries if we indeed recovered from a
+ // final checkpoint.
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ List<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
Review Comment:
Sure if you drop Java8 support first ;).
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #25292:
URL: https://github.com/apache/flink/pull/25292#issuecomment-2331889938
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "0587c702d816f061854210ba6ba3c1aacd062b77",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0587c702d816f061854210ba6ba3c1aacd062b77",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0587c702d816f061854210ba6ba3c1aacd062b77 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756452491
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -208,8 +209,8 @@ public void processElement(StreamRecord<CommittableMessage<CommT>> element) thro
// in case of unaligned checkpoint, we may receive notifyCheckpointComplete before the
// committables
- OptionalLong checkpointId = element.getValue().getCheckpointId();
- if (checkpointId.isPresent() && checkpointId.getAsLong() <= lastCompletedCheckpointId) {
+ long checkpointId = element.getValue().getCheckpointIdOrEOI();
+ if (checkpointId <= lastCompletedCheckpointId) {
Review Comment:
This is kinda of a change in the semantics on first glance. However, we should not receive any elements after EOI, so this code path is actually never triggered and now it's simpler.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758859370
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,13 +171,38 @@ public void initializeState(StateInitializationContext context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ // We can have the following cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we properly initialize the data and handle EOI, potentially resulting in
+ // duplicate summaries that the CommitterOperator needs to handle.
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true, we
+ // recover from a final checkpoint (for this subtask) and can ignore another EOI else we
+ // have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are true
+ // and do the same as above. As soon as one entry is EOI, we regularly start the writer
+ // and potentially emit duplicate summaries if we indeed recovered from a final
+ // checkpoint.
+ ArrayList<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
+ this.endOfInput = !previousState.isEmpty() && !previousState.contains(false);
+ sinkWriter =
+ this.endOfInput
+ ? new ClosedWriter<>()
Review Comment:
Fixed in a fixup commit. PTAL.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #25292:
URL: https://github.com/apache/flink/pull/25292#issuecomment-2349011518
I added the fixup commits inline. PTAL.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1759079752
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -208,8 +209,8 @@ public void processElement(StreamRecord<CommittableMessage<CommT>> element) thro
// in case of unaligned checkpoint, we may receive notifyCheckpointComplete before the
// committables
- OptionalLong checkpointId = element.getValue().getCheckpointId();
- if (checkpointId.isPresent() && checkpointId.getAsLong() <= lastCompletedCheckpointId) {
+ long checkpointId = element.getValue().getCheckpointIdOrEOI();
+ if (checkpointId <= lastCompletedCheckpointId) {
Review Comment:
Yes, we mean the same thing. For EOI committables, this check will always yield false. Even without the special case on EOI.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1749179970
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +216,18 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ // endOfInputState is union state, so it's enough if one task adds something to it
+ if (getRuntimeContext().getTaskInfo().getIndexOfThisSubtask() == 0) {
+ endOfInputState.add(true);
Review Comment:
What happens if we up-scale after this?
should we drop all other states?
Otherwise, should we check in initState() all items?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +216,18 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ // endOfInputState is union state, so it's enough if one task adds something to it
Review Comment:
What happens to the sinkWriter when two subtasks with different endOfInputState got merged on down-scaling? Shouldn't sink writer be be flushed? (I see it also uses operator list state ("writer_raw_states")) IIUC, it wouldn't
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #25292:
URL: https://github.com/apache/flink/pull/25292#issuecomment-2337429844
> Besides of that, I think `CompactorOperator.java` needs to be adjusted:
Thanks for the pointers. I'll address it later today.
> Also, could you clarify what is the exact path when a duplicate summary for the same checkpoint was emitted? IIUC, the 2nd code path is the "normal" `endInput` by `SinkWriterOperator` for then bounded input. But what is the 1st path?
Let's first recap why we need to emit in EOI at all: For streaming jobs, emitting committables on barrier is the correct behavior. For batch, emitting on EOI.
However, there is an additional case around streaming with bounded input and final checkpoints. The operator first receives an EOI and then the final barrier. After EOI, the operator is not allowed to emit another record. So just for this case, streaming also needs to emit on EOI and suppress on barrier.
Now the question is what happens on failure after final checkpoint. Logically speaking, the EOI before the checkpoint should influence the operator state in the checkpoint in such a way that after recover we are still not allowed to emit records. For sinks that means that all committables have already been transferred to the committer operator for the snapshot. For technical reasons, we still receive a second EOI after recovery. Logically, the first EOI should have sufficed. I'm assuming Flink does the second EOI mostly for channel management but maybe the proper fix is actually not calling EOI twice. @pnowojski could you PTAL?
Roman, you raise good questions concerning rescaling and more. The current implementation assumes that EOI will be received by all subtasks at the same logical time (e.g. on final checkpoint). Can we have instances where some subtasks shutdown earlier?
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756488654
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +216,18 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ // endOfInputState is union state, so it's enough if one task adds something to it
+ if (getRuntimeContext().getTaskInfo().getIndexOfThisSubtask() == 0) {
+ endOfInputState.add(true);
Review Comment:
I changed the logic entirely. We now drop and readd the state in `snapshotState`. This means if a downscaling happens and EOI_state = [true, false], on init endOfInput = false, and on next checkpoint EOI_state=[false]. I discussed the different cases in code.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on PR #25292:
URL: https://github.com/apache/flink/pull/25292#issuecomment-2345744447
I addressed your comments and took a different approach to state management (it looks similar but is conceptually rather different).
I restructured the PR and added 2 more commits to it. Unfortunately, the fixups looked then rather messy so I decided to force push everything again. So I'm sorry, but you more or less have to review again ;).
Btw I figured that the EOI changes were a breaking change of the experimental CommittableMessage and made them non-breaking instead. So the commit is essentially the same but looks a bit different.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "fapaul (via GitHub)" <gi...@apache.org>.
fapaul commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758462817
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java:
##########
@@ -138,15 +136,15 @@ public void processElement(StreamRecord<CompactorRequest> element) throws Except
@Override
public void endInput() throws Exception {
// add collecting requests into the final snapshot
- checkpointRequests.put(Long.MAX_VALUE, collectingRequests);
+ checkpointRequests.put(CommittableMessage.EOI, collectingRequests);
collectingRequests = new ArrayList<>();
// submit all requests and wait until they are done
- submitUntil(Long.MAX_VALUE);
+ submitUntil(CommittableMessage.EOI);
assert checkpointRequests.isEmpty();
getAllTasksFuture().join();
- emitCompacted(null);
Review Comment:
Is it safe to change this?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessageSerializer.java:
##########
@@ -91,13 +90,13 @@ public CommittableMessage<CommT> deserialize(int version, byte[] serialized)
return new CommittableWithLineage<>(
SimpleVersionedSerialization.readVersionAndDeSerialize(
committableSerializer, in),
- readCheckpointId(in),
+ in.readLong(),
Review Comment:
Do we need consider migration cases from SinkV1 where afaik the checkpointId is always `null`.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,13 +171,38 @@ public void initializeState(StateInitializationContext context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ // We can have the following cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we properly initialize the data and handle EOI, potentially resulting in
+ // duplicate summaries that the CommitterOperator needs to handle.
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true, we
+ // recover from a final checkpoint (for this subtask) and can ignore another EOI else we
+ // have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are true
+ // and do the same as above. As soon as one entry is EOI, we regularly start the writer
+ // and potentially emit duplicate summaries if we indeed recovered from a final
+ // checkpoint.
+ ArrayList<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
+ this.endOfInput = !previousState.isEmpty() && !previousState.contains(false);
+ sinkWriter =
+ this.endOfInput
+ ? new ClosedWriter<>()
Review Comment:
As discussed offline, this probably leaves an unclean state from the SinkWriter when it previously crashed.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -156,23 +156,16 @@ public void endInput() throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- if (endInput) {
- // This is the final checkpoint, all committables should be committed
- lastCompletedCheckpointId = Long.MAX_VALUE;
- } else {
- lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
- }
+ lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
}
private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
+ long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
do {
for (CheckpointCommittableManager<CommT> manager :
- committableCollector.getCheckpointCommittablesUpTo(lastCompletedCheckpointId)) {
- // wait for all committables of the current manager before submission
- boolean fullyReceived =
- !endInput && manager.getCheckpointId() == lastCompletedCheckpointId;
- commitAndEmit(manager, fullyReceived);
+ committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
Review Comment:
On first glance this change doesn't look correct.
By removing `fullyReceived`, can we now commit committables that are from the "current" checkpoint but on receival from a delayed notifyCheckpointComplete from a previous checkpoint.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -178,7 +214,7 @@ public void processElement(StreamRecord<InputT> element) throws Exception {
@Override
public void prepareSnapshotPreBarrier(long checkpointId) throws Exception {
super.prepareSnapshotPreBarrier(checkpointId);
- if (!endOfInput) {
+ if (!this.endOfInput) {
Review Comment:
Nit: Why `this` here we do not use inside the other methods?
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java:
##########
@@ -147,15 +146,16 @@ Collection<CommittableWithLineage<CommT>> drainFinished() {
}
CheckpointCommittableManagerImpl<CommT> merge(CheckpointCommittableManagerImpl<CommT> other) {
- checkArgument(Objects.equals(other.checkpointId, checkpointId));
+ checkArgument(other.checkpointId == checkpointId);
Review Comment:
Nit: This change should probably go into the commit, changing the type of the checkpoint fields
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,13 +171,38 @@ public void initializeState(StateInitializationContext context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ // We can have the following cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we properly initialize the data and handle EOI, potentially resulting in
+ // duplicate summaries that the CommitterOperator needs to handle.
+ // 2. state is not empty:
+ // - This implies Flink restores from a version that handles EOI
+ // - If there is one entry, no rescaling happened (for this subtask), so if it's true, we
+ // recover from a final checkpoint (for this subtask) and can ignore another EOI else we
+ // have a regular checkpoint.
+ // - If there are multiple entries, Flink downscaled, and we need to check if all are true
+ // and do the same as above. As soon as one entry is EOI, we regularly start the writer
+ // and potentially emit duplicate summaries if we indeed recovered from a final
+ // checkpoint.
Review Comment:
Very nice 👍
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -208,8 +209,8 @@ public void processElement(StreamRecord<CommittableMessage<CommT>> element) thro
// in case of unaligned checkpoint, we may receive notifyCheckpointComplete before the
// committables
- OptionalLong checkpointId = element.getValue().getCheckpointId();
- if (checkpointId.isPresent() && checkpointId.getAsLong() <= lastCompletedCheckpointId) {
+ long checkpointId = element.getValue().getCheckpointIdOrEOI();
+ if (checkpointId <= lastCompletedCheckpointId) {
Review Comment:
I do not fully understand the comment.
- Why is it a change in semantics? The condition looks the same the only case that is removed is a committable without checkpoint. Which scenario was this before the change?
- I thought that for the final checkpoint, we would first receive EOI and then do a final checkpoint. This would mean the committer receives data after EOI.
--
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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758906870
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +231,15 @@ public void processWatermark(Watermark mark) throws Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!endOfInput) {
+ endOfInput = true;
Review Comment:
I added the assertion. It can't hurt because we may have upstream operators that misbehave (custom 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
Re: [PR] [FLINK-25920] Ignore duplicate EOI in SinkWriter [flink]
Posted by "AHeise (via GitHub)" <gi...@apache.org>.
AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758998259
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java:
##########
@@ -156,23 +156,16 @@ public void endInput() throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- if (endInput) {
- // This is the final checkpoint, all committables should be committed
- lastCompletedCheckpointId = Long.MAX_VALUE;
- } else {
- lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
- }
+ lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
}
private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
+ long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
do {
for (CheckpointCommittableManager<CommT> manager :
- committableCollector.getCheckpointCommittablesUpTo(lastCompletedCheckpointId)) {
- // wait for all committables of the current manager before submission
- boolean fullyReceived =
- !endInput && manager.getCheckpointId() == lastCompletedCheckpointId;
- commitAndEmit(manager, fullyReceived);
+ committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
Review Comment:
Just in case you haven't seen it: `fullyReceived` is now implicitly always true.
--
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