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 2020/10/16 14:22:17 UTC
[GitHub] [flink] pnowojski opened a new pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
pnowojski opened a new pull request #13667:
URL: https://github.com/apache/flink/pull/13667
This PR deduplicates code between `CheckpointBarrierUnaligner`, `CheckpointBarrierAligner` and `AlternatingCheckpointBarrierHandler`
## 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/Mesos, 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770)
* de848b220414a40d24a3b6ec2656dcbe1ad30f3a Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-713002086
Azure is green, merging.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
}, {
"hash" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7933",
"triggerID" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de848b220414a40d24a3b6ec2656dcbe1ad30f3a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788)
* 9be8ddb9eb5c86dc7a1d46132c43c2f62007178e Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7933)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-712848488
Thanks for the review.
No I haven't but none of the changes are on any of the critical code paths (per record, or even per buffer). It only affects per checkpoint, and the difference is tiny (one additional "bimorphic" virtual call), so I wouldn't worry about performance impact.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710079579
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 cbca395729de539e4ee0d4f76056b09b09f3c935 (Fri Oct 16 14:24:51 UTC 2020)
**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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cbca395729de539e4ee0d4f76056b09b09f3c935 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767)
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cbca395729de539e4ee0d4f76056b09b09f3c935 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508482513
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
I see your point. Let's put it aside at least until we have design/implementation of new triggers.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508393114
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
- For passive timeouting aligned checkpoints I'm planning to at least return `boolean` from `barrierReceived()` or `barrierAnnouncementReceived()`
- For active timeouting there will be some even more elaborate/complicated/non trivial triggering mechanism on some timeout. I'm not entirely sure how this would be implemented (but this might not make it to 1.12, so we could put this aside for the time being).
- I wanted to have the control mechanism when to trigger checkpoint encapsulated in the `SingleCheckpointBarrierHandler` and not hard coded inside the `CheckpointBarrierHandler`. Having an enum, would move this logic actually to the `CheckpointBarrierHandler`, which would have hard coded `if-ologic` code what to do where, instead of encapsulating it inside the controller.
- I'm not sure how error prone this is. Please take a look how trivial are those classes. Controllers are trivial. Contract for triggering the checkpoints between them is also trivial - `return true`. (for active timeout this might be replaced by some other callback)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508401170
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean lastBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
I have also added a bit of javadocs to explain the contract 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
}, {
"hash" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de848b220414a40d24a3b6ec2656dcbe1ad30f3a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788)
* 9be8ddb9eb5c86dc7a1d46132c43c2f62007178e 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cbca395729de539e4ee0d4f76056b09b09f3c935 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767)
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770)
* de848b220414a40d24a3b6ec2656dcbe1ad30f3a 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] rkhachatryan commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
rkhachatryan commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508019514
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
##########
@@ -39,57 +39,83 @@
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_INPUT_END_OF_STREAM;
import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_SUBSUMED;
/**
- * {@link CheckpointBarrierUnaligner} is used for triggering checkpoint while reading the first barrier
- * and keeping track of the number of received barriers and consumed barriers.
+ * {@link SingleCheckpointBarrierHandler} is used for triggering checkpoint while reading the first barrier
+ * and keeping track of the number of received barriers and consumed barriers. It can handle/track
+ * just single checkpoint at a time. The behaviour when to actually trigger the checkpoint and
+ * what the {@link CheckpointableInput} should do is controlled by {@link CheckpointBarrierBehaviourController}.
*/
@Internal
@NotThreadSafe
-public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler {
+public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
- private static final Logger LOG = LoggerFactory.getLogger(CheckpointBarrierUnaligner.class);
+ private static final Logger LOG = LoggerFactory.getLogger(SingleCheckpointBarrierHandler.class);
- private final String taskName;
+ protected final String taskName;
Review comment:
Is `protected` visibility intentional?
I don't see any descendants and would prefer to have them `private` therefore.
(ditto: all fields)
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
Returning boolean from these two (first/last) methods seem unclear and error-prone to me.
How about
1. introducing `enum CheckpointTriggerStrategy { ON_FIRST_BARRIER, ON_LAST_BARRIER }`
2. a method `CheckpointBarrierBehaviourController.getTriggerStrategy`
3. returning void from first/last
?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean lastBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
The relation between `lastBarrierReceived` and `barrierReceived` is not very clear: whether `barrierReceived` will be called for the last barrier and in which order. The same for `firstBarrierReceived`.
How about renaming to `postProcessLastBarrier` (and `preProcessFirstBarrier`)?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/SingleCheckpointBarrierHandler.java
##########
@@ -39,57 +39,83 @@
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_INPUT_END_OF_STREAM;
import static org.apache.flink.runtime.checkpoint.CheckpointFailureReason.CHECKPOINT_DECLINED_SUBSUMED;
/**
- * {@link CheckpointBarrierUnaligner} is used for triggering checkpoint while reading the first barrier
- * and keeping track of the number of received barriers and consumed barriers.
+ * {@link SingleCheckpointBarrierHandler} is used for triggering checkpoint while reading the first barrier
+ * and keeping track of the number of received barriers and consumed barriers. It can handle/track
+ * just single checkpoint at a time. The behaviour when to actually trigger the checkpoint and
+ * what the {@link CheckpointableInput} should do is controlled by {@link CheckpointBarrierBehaviourController}.
*/
@Internal
@NotThreadSafe
-public class CheckpointBarrierUnaligner extends CheckpointBarrierHandler {
+public class SingleCheckpointBarrierHandler extends CheckpointBarrierHandler {
- private static final Logger LOG = LoggerFactory.getLogger(CheckpointBarrierUnaligner.class);
+ private static final Logger LOG = LoggerFactory.getLogger(SingleCheckpointBarrierHandler.class);
- private final String taskName;
+ protected final String taskName;
- private int numBarriersReceived;
+ protected final CheckpointBarrierBehaviourController controller;
- /** A future indicating that all barriers of the a given checkpoint have been read. */
- private CompletableFuture<Void> allBarriersReceivedFuture = FutureUtils.completedVoidFuture();
+ protected int numBarriersReceived;
/**
* The checkpoint id to guarantee that we would trigger only one checkpoint when reading the same barrier from
* different channels.
*/
- private long currentCheckpointId = -1L;
+ protected long currentCheckpointId = -1L;
- private int numOpenChannels;
+ protected int numOpenChannels;
- private final SubtaskCheckpointCoordinator checkpointCoordinator;
+ protected CompletableFuture<Void> allBarriersReceivedFuture = FutureUtils.completedVoidFuture();
- private final CheckpointableInput[] inputs;
-
- CheckpointBarrierUnaligner(
+ @VisibleForTesting
+ static SingleCheckpointBarrierHandler createUnalignedCheckpointBarrierHandler(
SubtaskCheckpointCoordinator checkpointCoordinator,
String taskName,
AbstractInvokable toNotifyOnCheckpoint,
CheckpointableInput... inputs) {
+ return new SingleCheckpointBarrierHandler(
+ taskName,
+ toNotifyOnCheckpoint,
+ (int) Arrays.stream(inputs).flatMap(gate -> gate.getChannelInfos().stream()).count(),
+ new UnalignedController(checkpointCoordinator, inputs));
+ }
+
+ SingleCheckpointBarrierHandler(
+ String taskName,
+ AbstractInvokable toNotifyOnCheckpoint,
+ int numOpenChannels,
+ CheckpointBarrierBehaviourController controller) {
super(toNotifyOnCheckpoint);
this.taskName = taskName;
- this.inputs = inputs;
- numOpenChannels = (int) Arrays.stream(inputs).flatMap(gate -> gate.getChannelInfos().stream()).count();
- this.checkpointCoordinator = checkpointCoordinator;
+ this.numOpenChannels = numOpenChannels;
+ this.controller = controller;
+ }
+
+ @Override
+ protected void abortPendingCheckpoint(
Review comment:
This method is not used anymore (now _this_ class uses `controller.abortPendingCheckpoint`)
so the method can be deleted.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508393114
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
- For passive timeouting aligned checkpoints I'm planning to at least return `boolean` from `barrierReceived()` or `barrierAnnouncementReceived()`
- For active timeouting there will be some even more elaborate/complicated/non trivial triggering mechanism on some timeout. I'm not entirely sure how this would be implemented (but this might not make it to 1.12, so we could put this aside for the time being).
- I wanted to have the control mechanism when to trigger checkpoint encapsulated in the `SingleCheckpointBarrierHandler` and not hard coded inside the `CheckpointBarrierHandler`. Having an enum, would move this logic actually to the `CheckpointBarrierHandler`, which would have hard coded `if-ologic` code what to do where, instead of encapsulating it inside the controller.
- I'm not sure how error prone this is. Please take a look how trivial are those classes. Controllers are trivial. Contract for triggering the checkpoints between them is also trivial - `return true`. (for active timeout this might be replaced by some other callback)
All in all I think `enum` would be a bad idea and also a step in a wrong direction considering the future planned changes.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski merged pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski merged pull request #13667:
URL: https://github.com/apache/flink/pull/13667
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* cbca395729de539e4ee0d4f76056b09b09f3c935 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767)
* 5ab532006ccbfb5632b4ad60053203a5e5dea3a5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
}, {
"hash" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7933",
"triggerID" : "9be8ddb9eb5c86dc7a1d46132c43c2f62007178e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9be8ddb9eb5c86dc7a1d46132c43c2f62007178e Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7933)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13667:
URL: https://github.com/apache/flink/pull/13667#issuecomment-710095324
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7767",
"triggerID" : "cbca395729de539e4ee0d4f76056b09b09f3c935",
"triggerType" : "PUSH"
}, {
"hash" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7770",
"triggerID" : "5ab532006ccbfb5632b4ad60053203a5e5dea3a5",
"triggerType" : "PUSH"
}, {
"hash" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788",
"triggerID" : "de848b220414a40d24a3b6ec2656dcbe1ad30f3a",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* de848b220414a40d24a3b6ec2656dcbe1ad30f3a Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7788)
<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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13667: [FLINK-19679] Deduplicate code between CheckpointBarrierUnaligner and CheckpointBarrierAligner
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13667:
URL: https://github.com/apache/flink/pull/13667#discussion_r508393114
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/CheckpointBarrierBehaviourController.java
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.InputChannelInfo;
+import org.apache.flink.runtime.io.network.api.CheckpointBarrier;
+
+import java.io.IOException;
+
+/**
+ * Controls when the checkpoint should be actually triggered.
+ */
+@Internal
+public interface CheckpointBarrierBehaviourController {
+
+ void barrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier);
+
+ /**
+ * @return {@code true} if checkpoint should be triggered.
+ */
+ boolean firstBarrierReceived(InputChannelInfo channelInfo, CheckpointBarrier barrier) throws IOException;
Review comment:
- For passive timeouting aligned checkpoints I'm planning to at least return `boolean` from `barrierReceived()` or `barrierAnnouncementReceived()`
- For active timeouting there will be some even more elaborate/complicated/non trivial triggering mechanism on some timeout. I'm not entirely sure how this would be implemented (but this might not make it to 1.12, so we could put this aside for the time being).
- I wanted to have the control mechanism when to trigger checkpoint encapsulated in the `SingleCheckpointBarrierHandler` and not hard coded inside the `CheckpointBarrierHandler`. Having an enum, would move this logic actually to the `CheckpointBarrierHandler`, which would have hard coded `if-ologic` code what to do where, instead of encapsulating it inside the controller.
- I'm not sure how error prone this is. Please take a look how trivial are those classes. Controllers are trivial. Contract for triggering the checkpoints between them is also trivial - `return true`. (for active timeout this might be replaced by some other callback)
All in all I think `enum` would be a bad idea even now and also a step in a wrong direction considering the future planned changes.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org