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/04/09 09:48:09 UTC
[GitHub] [flink] zhijiangW opened a new pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
zhijiangW opened a new pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687
## What is the purpose of the change
During recovery process for unaligned checkpoint, the remote input channel state should also be recovered besides with existing operator states.
The remote channel would request buffer and then interact with ChannelStateReader to fill in the state data. The filled buffer would be inserted into respective data queue for processing in normal way.
It should guarantee that the new data from upstream partition should not overtake the input state data to avoid mis-order issue.
## Brief change log
- *Delay request partition in `RemoteInputChannel` after finishing reading channel state.*
- *Add related interfaces method from input channel to handler stacks to notify the read of channel state*
- *Implement the main process to read channel state via interacting with `ChannelStateReader`*
## Verifying this change
Add some new unit tests to verify the changes
## 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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-612792024
Considering another possible option to implement the input recovery, I thought through a bit for the proposal of introducing a parallel `StreamTaskSpilledInput` /`SpilledInputGate`/`SpilledInputChannel `. If we want to further support the new unaligned checkpoint progress during recovery, that means two separate components should run in parallel future. Then it seems harder to control and bring more component changes if we work on more high-level components.
Regarding the lowest-level component `SpilledInputChannel`, I also think it should be more complex if running in parallel with respective `RemoteInputChannel` future for below concerns.
- `SingleInputGate` would maintain two suits of input channel collections. The normal remote/local channel collection can only receive events like `CheckpointBarrier`, `EndOfPartition` during recovery, and the other `SpilledInputChannel` collection can read state. We might need to adjust the queue `inputChannelsWithData` in gate to control the priority of data from different channel collection.
- How to handle the exclusive/floating buffer distribution inside `SpilledInputChannel`? There might be two options. If we want to recovery only one channel state at a time by one thread, then we can give all the available buffers(regardless of exclusive/floating) to this channel. And after finishing this channel, we can move forward to the next channel with all available buffers. If we want o recovery multiple channels in parallel with different threads(as this PR did now), we might probably back to the existing mechanisms as in `RemoteInputChannel` , then we need to consider the code duplication issue.
- We still need to adjust the internal logic inside `RemoteInputChannel` to do so. E.g. when it receives the barrier from network, it should not request floating buffers based on backlog, along with the related logics with exclusive buffer and floating buffers, until all the `SpilledInputChannel` finish.
- When all the `SpilledInputChannel` finish, it should notify the `RemoteInputChannel` restore the normal exclusive/floating buffers logic. The interaction is needed among different channels via `SingleInputGate`.
In conclusion, the essential difference with current PR implementation is whether to extract the logic `readChannelState` out of `RemoteInputChannel` to relief its load. In contrast, this logic is introduced into new `SpilledInputChannel` which would bring extra complexity in `SingleInputGate` component and the trouble for reusing the existing buffer distribution/request logic.
So I a bit prefer the current way in PR at-least for MVP now until we fully consider well for everything future.
But we can make some partial changes to control the channel state process on `SingleInputGate` component. E.g. introduce a separate thread pool to execute the channel read instead of reusing netty threads as above discussed, and control the channel read one by one instead of concurrently if have the requirement.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-612652864
Thanks for the review and suggestions @pnowojski !
Regarding the issue of which thread handling read state, I also fully considered your concerns of netty thread before. Let me further elaborate a bit:
> Netty pool size is configured for handling network traffic, not for reading data from the spilled files. They > should be configured separately, as 32 threads for netty is not that uncommon, while that many threads > for File IO doesn't make much sense.
The default number of netty thread is bounded to slot amount if not config explicitly. I also mentioned about the consideration of how many threads are proper for serving state read in previous comment https://github.com/apache/flink/pull/11687#issuecomment-611465439. Firstly I think one thread should probably be not enough is practice, the reason explained here https://github.com/apache/flink/pull/11687#discussion_r407222896. IMO 10 threads might be an ideally amount in SATA disks, and for SDD we can have more threads concurrent reading in general. Actually the proper size is also related to the current IO pressure, so it is hard to determine beforehand, then I am willing to leave the choice to exiting netty threads to avoid bringing additional thread cost.
It might seem a bit tricky to make netty thread not only work for network traffic by design. But it already exists the same scenarios now that netty thread would read spilled file for blocking partition in batch jobs, even the case is almost the same with reading channel state to some extent. From another aspect, the netty threads on consumer side take the role of feeding data for task processing either from network or from state files. If we do not use netty threads, they are almost idle during state recovery, except receiving new checkpoint barrier occasionally for future supporting.
> Reading File IO is blocking operation. Blocking Netty threads can cause problems.
In general I am always against blocking operation in netty thread. But for this case the netty threads should be almost idle during recovery as I mentioned above. And I also considered to support receiving new checkpoint barrier during recovery future, which makes netty thread also work for network read/write ops by internal (e.g. after reading every state buffer). If it takes several minutes to read one buffer in extremely environment to delay new checkpoint receive, I think it should be a minor issue. Because in such bad IO situation, the total recovery might take hours to finish, then supporting minutes level progress for new unaligned checkpoint in this period should be enough. In other words, it seems meaningless if we still support second level progress for new checkpoint during recovery, because the state consumption probably has not been moved forward yet.
In general I like your ideas of the possibility to implement the input recovery from another layer, which I have not thought of before. I agree that it might be better to not touch the existing `RemoteInputChannel` if possible because it is already involved in many things. I would further think though how many efforts we need to pay if make it work on separate `StreamTaskSpilledInput `/`SpilledInputGate`/`SpilledInputChannel` level or whether it is feasible to work in cleaner way, and feedback to you later.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408569272
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -243,7 +249,23 @@ void requestPartitions() throws IOException, InterruptedException {
}
for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition();
+ executor.submit(() -> {
+ try {
+ inputChannel.initializeState(reader);
+ } catch (Throwable t) {
+ inputChannel.setError(t);
+ }
+ });
+ }
+
+ for (InputChannel inputChannel : inputChannels.values()) {
+ executor.submit(() -> {
+ try {
+ inputChannel.requestSubpartition();
Review comment:
Yes, it has another issue here. In order to compatible with previous process, we can divide the previous `requestSubpartition` into two steps. The first step is to create client which can be done during `InputGate#setup` as before. The second step is to actual request partition which is triggered after reading channel state as done in this PR change.
To do so, this check is still valid, but I guess it is still breaking the previous assumption which actually wants to guarantee the partition is really requested before calling `getNextBuffer`. But now the existing client does not mean the partition was requested already. So another option might remove this `checkState`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611438822
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 c8c0da9e2badf68c11e8466943c7a8e14156c858 (Thu Apr 09 09:50:27 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r406231112
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- protected void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ protected void requestSubpartition() throws IOException, InterruptedException {
Review comment:
Are those changes about dropping `int subpartitionIndex` (significant part of this commit) related to the main purpose of the PR? Shouldn't they be a separate commit?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-616153893
@flinkbot run travis
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408342152
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
checkState(this.bufferPool == null, "Bug in input gate setup logic: Already registered buffer pool.");
// assign exclusive buffers to input channels directly and use the rest for floating buffers
assignExclusiveSegments();
BufferPool bufferPool = bufferPoolFactory.get();
setBufferPool(bufferPool);
+ }
+
- requestPartitions();
Review comment:
who is now calling this method in batch tasks?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-612652864
Thanks for the review and suggestions @pnowojski !
Regarding the issue of which thread handling read state, I also fully considered your concerns of netty thread before. Let me further elaborate a bit:
> Netty pool size is configured for handling network traffic, not for reading data from the spilled files. They > should be configured separately, as 32 threads for netty is not that uncommon, while that many threads > for File IO doesn't make much sense.
The default number of netty thread is bounded to slot amount if not config explicitly. I also mentioned about the consideration of how many threads are proper for serving state read in previous comment https://github.com/apache/flink/pull/11687#issuecomment-611465439. Firstly I think one thread should probably be not enough is practice, the reason explained here https://github.com/apache/flink/pull/11687#discussion_r407222896. IMO 10 threads might be an ideally amount in SATA disks, and for SDD we can have more threads concurrent reading in general. Actually the proper size is also related to the current IO pressure, so it is hard to determine beforehand, then I am willing to leave the choice to exiting netty threads to avoid bringing additional thread cost.
It might seem a bit tricky to make netty thread not only work for network traffic by design. But it already exists the same scenarios now that netty thread would read spilled file for blocking partition in batch jobs, even the case is almost the same with reading channel state to some extent. From another aspect, the netty threads on consumer side take the role of feeding data for task processing either from network or from state files. If we do not use netty threads, they are almost idle during state recovery, except receiving new checkpoint barrier occasionally for future supporting.
> Reading File IO is blocking operation. Blocking Netty threads can cause problems.
In general I am always against blocking operation in netty thread. But for this case the netty threads should be almost idle during recovery as I mentioned above. And I also considered to support receiving new checkpoint barrier during recovery future, which makes netty thread also work for network read/write ops by internal (e.g. after reading every state buffer). If it takes several minutes to read one buffer in extremely environment to delay new checkpoint receive, I think it should be a minor issue. Because in such bad IO situation, the total recovery might take hours to finish, then supporting minutes level progress for new unaligned checkpoint in this period should be enough. In other words, it seems meaningless if we still support second level progress for new checkpoint during recovery, because the state consumption probably has not been moved forward yet.
In general I like your ideas of the possibility to implement the input recovery from another layer, which I have not thought of before. I agree that it might be better to not touch the existing `RemoteInputChannel` if possible, because it was already involved in many things. I would further think though how many efforts we need to pay if make it work on separate `StreamTaskSpilledInput `/`SpilledInputGate`/`SpilledInputChannel` level or whether it is feasible to work in cleaner way, and feedback to you later.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c8c0da9e2badf68c11e8466943c7a8e14156c858 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/159567897) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160726649) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661)
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
}, {
"hash" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160918825",
"triggerID" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"triggerType" : "PUSH"
}, {
"hash" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7737",
"triggerID" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* d51ce7f47381d99b843278cd701dcff223761a0b UNKNOWN
* 2d03b238ddb9bb7d1b32261d6d61be96983b8fb4 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160918825) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7737)
<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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r407224535
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- protected void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ protected void requestSubpartition() throws IOException, InterruptedException {
Review comment:
I actually considered whether to extract it as a separate commit or not when implementation. I think both ways can be interpreted from different motivations.
- If making it as a separate hotfix or refactor commit, we only save one argument from this interface method and fetch it from `SingleInputGate` instead. Then the benefit seems not very obvious if without the following input recovery change.
- If making it along with the input recovery changes, the motivation seems a must-work to do. Since we migrate the actual usage of this argument outside of this method, then this argument should also be cleaned up as a result.
Therefore I preferred a bit not to make it a separate commit before. It is fine for my either ways actually. If you are in favor of separate commit, i am willing to change 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-616153853
@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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408333775
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- protected void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ protected void requestSubpartition() throws IOException, InterruptedException {
Review comment:
It made the final commit a bit smaller, so I think it was worth it (to speed up reviewing). As hotfix on it's own it's also a nice simplification I think.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b32c0f208a47925d94e290a91132ec15590236e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160240460) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475)
<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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW removed a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW removed a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-615378176
@flinkbot run travis
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408574292
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
TBH the current wait way is also not in my favor as I mentioned as unsatisfied points, because how long time to wait might be an issue.
I also considered other ways, but the key problem is that we can not exit this runnable to let the thread execute other runables in advance, so it seems somehow as blocking way here. In contrast, the previous floating request from netty thread was non-blocking way, so we do not need the mechanism of `notifyAll` inside `RemoteInputChannel#recycle` and `RemoteInputChannel#notifyBufferAvailable`.
Maybe it is possible to add `notifyAll` in above two methods for compatible with the new process. WDYT?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
}, {
"hash" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "2d03b238ddb9bb7d1b32261d6d61be96983b8fb4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b UNKNOWN
* 2d03b238ddb9bb7d1b32261d6d61be96983b8fb4 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b32c0f208a47925d94e290a91132ec15590236e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160240460) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475)
* c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160726649) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408338769
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -243,7 +249,23 @@ void requestPartitions() throws IOException, InterruptedException {
}
for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition();
+ executor.submit(() -> {
+ try {
+ inputChannel.initializeState(reader);
+ } catch (Throwable t) {
+ inputChannel.setError(t);
+ }
+ });
Review comment:
I would move the for loop inside the executor, to avoid race conditions and not depend on `executor` being single threaded.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408569446
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -243,7 +249,23 @@ void requestPartitions() throws IOException, InterruptedException {
}
for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition();
+ executor.submit(() -> {
+ try {
+ inputChannel.initializeState(reader);
+ } catch (Throwable t) {
+ inputChannel.setError(t);
+ }
+ });
Review comment:
good point
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408728931
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
Ok I see. We can not block the shared executor for unspecified long time. This method would have to be rewritten in a such way, that once buffers are not available, it exits, and re-enqueuing action is done once more buffers are ready.
If we spawned our custom thread for unspilling, we could block it however we want, but in that case we would also need a mechanism to wake it up once more buffers are ready, instead of sleeping for fixed amount of time.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r410380876
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -243,7 +249,23 @@ void requestPartitions() throws IOException, InterruptedException {
}
for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition();
+ executor.submit(() -> {
+ try {
+ inputChannel.initializeState(reader);
+ } catch (Throwable t) {
+ inputChannel.setError(t);
+ }
+ });
+ }
+
+ for (InputChannel inputChannel : inputChannels.values()) {
+ executor.submit(() -> {
+ try {
+ inputChannel.requestSubpartition();
Review comment:
I currently remove this `checkState` in PR for simple. If you have other thoughts, then we can further sync 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r407224535
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- protected void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ protected void requestSubpartition() throws IOException, InterruptedException {
Review comment:
I actually considered whether to extract it as a separate commit or not when implementation. I think both ways can be interpreted from different motivations.
If making it as a separate hotfix or refactor commit, we only save one argument from this interface method and fetch it from `SingleInputGate` instead. Then the benefit seems not very obvious if without the following input recovery change.
If making it along with the input recovery changes, the motivation seems a must-work to do. Since we migrate the actual usage of this argument outside of this method, then this argument should also be cleaned up as a result.
Therefore I preferred a bit not to make it a separate commit before. It is fine for my either ways actually. If you are in favor of separate commit, i am willing to change 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r406825039
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,16 +154,58 @@ void assignExclusiveSegments() throws IOException {
}
}
+ /**
+ * Reads the channel state data executed by netty thread, so it can make use of almost all the
+ * existing processes to avoid bringing additional race conditions with task thread. Also it can
+ * avoid introducing another thread pool to handle this work to make things more complex.
+ */
+ private void readInputChannelState() throws IOException {
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer != null) {
+ bufferQueue.addFloatingBuffer(buffer);
+ continue;
+ } else {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForStateBuffers = true;
+ return;
+ }
+ }
+ }
+
+ ChannelStateReader.ReadResult result = inputGate.stateReader.readInputData(channelInfo, buffer);
Review comment:
`stateReader.readInputData` is not thread safe as it is currently and this `readInputChannelState` method can be executed by netty threads concurrently for different channels, right?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r407222896
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,16 +154,58 @@ void assignExclusiveSegments() throws IOException {
}
}
+ /**
+ * Reads the channel state data executed by netty thread, so it can make use of almost all the
+ * existing processes to avoid bringing additional race conditions with task thread. Also it can
+ * avoid introducing another thread pool to handle this work to make things more complex.
+ */
+ private void readInputChannelState() throws IOException {
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer != null) {
+ bufferQueue.addFloatingBuffer(buffer);
+ continue;
+ } else {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForStateBuffers = true;
+ return;
+ }
+ }
+ }
+
+ ChannelStateReader.ReadResult result = inputGate.stateReader.readInputData(channelInfo, buffer);
Review comment:
Yes, `readInputChannelState` can be executed concurrently by multiple netty threads for different channels by design. I think in general task processing should be more faster than reading states, so one thread might not be enough for filling buffer to feed task thread well. And every channel actually has exclusive buffers which can be used in parallel to speed up recovery process.
I overlooked the `NotThreadSafe` annotation in `ChanelStateReaderImpl`. Since every input channel handle will actually generate a separate `ChannelStateStreamReader` and respective stream, I was supposed one input channel state should not be read by multiple threads, but different channel states can be read by different threads concurrent. I would further confirm with Roman whether there are other limitations.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-615378176
@flinkbot run travis
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409402195
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -289,6 +292,8 @@ protected StreamTask(
} else {
this.timerService = timerService;
}
+
+ this.ioExecutor = Executors.newSingleThreadExecutor(new ExecutorThreadFactory("state-io"));
Review comment:
rename `channelStateUnspillingExecutor`?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409283520
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
Yes, we have the same issue in another option of `SpilledInputChannel`/`SpilledInputGate`.
In conclusion, there may be several solutions:
- `wait()` if unavailable buffers: block the unspilling thread always and it fits our current requirements. We should not exit current channel to switch another channel temporarily, which might bring random IO. And it requires the wakeup mechanism when buffer available again.
- `wait(timeout)` if unavailable buffers: more or less the same with above `wait()`, but wakeup mechanism is not a mandatory, can be regarded as somehow improvement to wakeup eerily.
- unblocking way: terminate the current channel and unspill another channel with available buffers. It would bring random IO as mentioned above and not the current suggestion.
Based on the current situation with custom thread for unspilling, and allow only one thread to unspill channel one by one to avoid random IO, so I choose the unblocking option 1 or 2.
Regarding the option 2, wakeup is not necessary, then we do not need to touch the previous processes `RemoteInputChannel#recycle` and `RemoteInputChannel#notifyBufferAvailable`. If we want to add the wakeup mechanism, option 1 also makes sense.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408340272
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
Is someone notifing this thread to wake up? Can not we wait on the buffer in another way?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r406292661
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -311,10 +367,16 @@ public void recycle(MemorySegment segment) {
ExceptionUtils.rethrow(t);
}
}
+
+ shouldNotifyChannelState = isWaitingForStateBuffers;
+ isWaitingForStateBuffers = false;
+
numAddedBuffers = bufferQueue.addExclusiveBuffer(new NetworkBuffer(segment, this), numRequiredBuffers);
}
- if (numAddedBuffers > 0 && unannouncedCredit.getAndAdd(numAddedBuffers) == 0) {
+ if (shouldNotifyChannelState) {
+ notifyReadChannelState();
Review comment:
`notifyStateBuffersAvailable()`? `notifyBuffersForReadChannelStateAvailable`?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408728931
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
Ok I see. We can not block the shared executor for unspecified long time. This method would have to be rewritten in a such way, that once buffers are not available, it exits, and re-enqueuing action is done once more buffers are ready.
If we spawned our custom thread for unspilling, we could block it however we want, but in that case we would also need a mechanism to wake it up once more buffers are ready, instead of sleeping for fixed amount of time.
edit: I guess we would have exactly the same issues with a separate unspilling component like `SpilledInputChannel` or `SpilledInputGate`?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b32c0f208a47925d94e290a91132ec15590236e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160240460) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475)
* c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e 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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r407222979
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -311,10 +367,16 @@ public void recycle(MemorySegment segment) {
ExceptionUtils.rethrow(t);
}
}
+
+ shouldNotifyChannelState = isWaitingForStateBuffers;
+ isWaitingForStateBuffers = false;
+
numAddedBuffers = bufferQueue.addExclusiveBuffer(new NetworkBuffer(segment, this), numRequiredBuffers);
}
- if (numAddedBuffers > 0 && unannouncedCredit.getAndAdd(numAddedBuffers) == 0) {
+ if (shouldNotifyChannelState) {
+ notifyReadChannelState();
Review comment:
I like `notifyStateBuffersAvailable`
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c8c0da9e2badf68c11e8466943c7a8e14156c858 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/159567897) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278)
<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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409409374
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
Review comment:
could we pass the state recovery parameters here? And have a single step `setup` method instead of introducing 2nd initialisation step (`setup` and `initializeState`).
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r410226251
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
Review comment:
> The previous assumption was that the output recovery should execute earlier than the input recovery in order to occupy more floating buffers from global, to get the benefit to speedup recovery process. Now the output recovery is executed by task thread during StreamTask#beforeInvoke. If we integrate the input recovery within setup process, then it would perform before output recovery to occupy more floating buffers in limited buffers environment.
Ok, I haven't considered that. I guess it's best to keep it as it is?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r410380283
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
Review comment:
Yes, I guess so.
I found another benefit to execute the input state recovery inside `StreamTask` instead of `Task` stack(setup integration). The new dedicated executor for unspilling channel state is only meaningful for the unaligned checkpoint mode. So I tried not to create this executor for normal checkpoints, also for batch jobs to reduce total threads in TaskManager. This improvement can only be done inside `StreamTask` because it can judge the checkpoint mode via `StreamConfig`. And task class can not get this option because of `runtime` module can not depend on `streaming` module.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c8c0da9e2badf68c11e8466943c7a8e14156c858 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/159567897) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278)
* 7b32c0f208a47925d94e290a91132ec15590236e Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160240460) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475)
<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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408336918
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -243,7 +249,23 @@ void requestPartitions() throws IOException, InterruptedException {
}
for (InputChannel inputChannel : inputChannels.values()) {
- inputChannel.requestSubpartition();
+ executor.submit(() -> {
+ try {
+ inputChannel.initializeState(reader);
+ } catch (Throwable t) {
+ inputChannel.setError(t);
+ }
+ });
+ }
+
+ for (InputChannel inputChannel : inputChannels.values()) {
+ executor.submit(() -> {
+ try {
+ inputChannel.requestSubpartition();
Review comment:
Could it be that we start reading from a channel before we actually request subpartition? For example:
```
Optional<BufferAndAvailability> RemoteInputChannel#getNextBuffer() throws IOException {
(...)
checkState(partitionRequestClient != null, "Queried for a buffer before requesting a queue."); // ??
```
or what if `partitionRequestClient` is already not null, but
```
partitionRequestClient.requestSubpartition(partitionId, subpartitionIndex, this, 0);
```
hasn't completed?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611465439
Some thoughts to share during implementation.
1. Which kind of thread should read the channel state?
We can not use the task thread, because it will process and consume the filled buffers to recycle them to load more channel states. In other words, the task thread and state loading thread should work in parallel.
If we introduce another thread pool for reading state, it has more troubles. First we need to maintain the lifecycle of it. Second it will bring more threads overhead in a TaskExecutor container. Third it is hard to determine how many threads are suitable based on demands. One thread might not be enough for serving many input channels at the same time. Fourth it bring the thread model more complex besides with both task and netty threads.
Therefore, the existing netty threads are the most ideally candidate to solve all the above concerns. We do not need to worry about the netty thread is involving in IO operations, because it already exists the similar case on upstream side to read blocking partitions by netty thread. And we can also make use of the default load balance among input channels with netty threads.
2. The current implementation will delay partition request until finishing channel state recovery. It is mostly easy to avoid data order issues. But in future if we want to perform new checkpoints during recovery, the partition request should also be done meanwhile, and it would bring some difficulties.
One option is we give 0 credits during partition request to avoid receiving any data during recovery, then we need to adjust the logic when to announce the real credit again after recovery. But another trouble is that we can also receive `EndOfPartitionEvent` besides `CheckpointBarrier` without credits. So how to store this event in buffer queue to avoid mis-order issue.
Another potential issue is how to switch between reading channel state and receiving network data in netty thread if we want to support checkpoint during recovery. So I take the simple way in MVP now, to further think through how to support checkpoint during recovery future.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r408575571
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
checkState(this.bufferPool == null, "Bug in input gate setup logic: Already registered buffer pool.");
// assign exclusive buffers to input channels directly and use the rest for floating buffers
assignExclusiveSegments();
BufferPool bufferPool = bufferPoolFactory.get();
setBufferPool(bufferPool);
+ }
+
- requestPartitions();
Review comment:
No matter streaming or batch, this method is always called by task thread when the `SingleInputGate` is created to setup. The only difference between streaming and batch is for RPC call `updateInputChannel` which is only valid in batch. But for batch we do not support unaligned cp, so the new transformed remote channel from unknown can request partition directly.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] zhijiangW commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
zhijiangW commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409995125
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/SingleInputGate.java
##########
@@ -213,19 +215,23 @@ public SingleInputGate(
}
@Override
- public void setup() throws IOException, InterruptedException {
+ public void setup() throws IOException {
Review comment:
I agree that it has benefits to reduce steps for managing the lifecycle of input gate. But I thought of another potential concern if integrating it within `setup`.
The previous assumption was that the output recovery should execute earlier than the input recovery in order to occupy more floating buffers from global, to get the benefit to speedup recovery process. Now the output recovery is executed by task thread during `StreamTask#beforeInvoke`. If we integrate the input recovery within `setup` process, then it would perform before output recovery to occupy more floating buffers in limited buffers environment.
I also considered integrating the output recovery within `ResultPartition#setup`, but the output recovery is executed by task thread then it would block the following operations, so I wonder it might also bring potential risks to do so.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160726649) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661)
* 5bad018f00f85a7359345187a12d7938aa510d25 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409414113
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
Review comment:
Does it make sense to rename `initializeState` to `readRecoveredState`? or `processRecoveredState`?
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot commented on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot commented on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c8c0da9e2badf68c11e8466943c7a8e14156c858 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "CANCELED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [CANCELED](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r409415021
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/RemoteInputChannel.java
##########
@@ -149,6 +149,47 @@ void assignExclusiveSegments() throws IOException {
}
}
+ @Override
+ public void initializeState(ChannelStateReader reader) throws IOException, InterruptedException {
+ numRequiredBuffers = initialCredit + inputGate.getBufferPool().getMaxNumberOfMemorySegments();
+ unannouncedCredit.set(initialCredit);
+
+ while (true) {
+ Buffer buffer;
+ synchronized (bufferQueue) {
+ buffer = bufferQueue.takeBuffer();
+ if (buffer == null) {
+ if (isReleased()) {
+ return;
+ }
+
+ if (!isWaitingForFloatingBuffers) {
+ buffer = inputGate.getBufferPool().requestBuffer();
+ if (buffer == null) {
+ inputGate.getBufferProvider().addBufferListener(this);
+ isWaitingForFloatingBuffers = true;
+ }
+ }
+ }
+ }
+
+ if (buffer == null) {
+ wait(10);
Review comment:
Oh, I missed that we are already using a separate thread pool. In that case we don't need the not blocking option now.
It would be still good to get rid of timed wait in favour of some notifications though.
----------------------------------------------------------------
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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
}, {
"hash" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "d51ce7f47381d99b843278cd701dcff223761a0b",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "615378176",
"triggerType" : "MANUAL"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160745576",
"triggerID" : "616153893",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160745576) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7679)
* d51ce7f47381d99b843278cd701dcff223761a0b 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "PENDING",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 7b32c0f208a47925d94e290a91132ec15590236e Travis: [PENDING](https://travis-ci.com/github/flink-ci/flink/builds/160240460) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475)
<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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "FAILURE",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c8c0da9e2badf68c11e8466943c7a8e14156c858 Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/159567897) Azure: [FAILURE](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278)
* 7b32c0f208a47925d94e290a91132ec15590236e 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
With regards,
Apache Git Services
[GitHub] [flink] flinkbot edited a comment on issue #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on issue #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#issuecomment-611445542
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/159567897",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7278",
"triggerID" : "c8c0da9e2badf68c11e8466943c7a8e14156c858",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7475",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "7b32c0f208a47925d94e290a91132ec15590236e",
"status" : "DELETED",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160240460",
"triggerID" : "7b32c0f208a47925d94e290a91132ec15590236e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "FAILURE",
"url" : "https://travis-ci.com/github/flink-ci/flink/builds/160726649",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"status" : "PENDING",
"url" : "https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661",
"triggerID" : "c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e",
"triggerType" : "PUSH"
}, {
"hash" : "5bad018f00f85a7359345187a12d7938aa510d25",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "5bad018f00f85a7359345187a12d7938aa510d25",
"triggerType" : "PUSH"
}, {
"hash" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "371f4a669c47fb1ad3d7f7c643edd71c54405195",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* c15a33465aff4a5bcbb0b32a75393c19ef3c4e1e Travis: [FAILURE](https://travis-ci.com/github/flink-ci/flink/builds/160726649) Azure: [PENDING](https://dev.azure.com/rmetzger/5bd3ef0a-4359-41af-abca-811b04098d2e/_build/results?buildId=7661)
* 5bad018f00f85a7359345187a12d7938aa510d25 UNKNOWN
* 371f4a669c47fb1ad3d7f7c643edd71c54405195 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
With regards,
Apache Git Services
[GitHub] [flink] pnowojski commented on a change in pull request #11687:
[FLINK-16536][network][checkpointing] Implement InputChannel state recovery
for unaligned checkpoint
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #11687: [FLINK-16536][network][checkpointing] Implement InputChannel state recovery for unaligned checkpoint
URL: https://github.com/apache/flink/pull/11687#discussion_r406231112
##########
File path: flink-runtime/src/main/java/org/apache/flink/runtime/io/network/partition/consumer/LocalInputChannel.java
##########
@@ -95,7 +95,7 @@ public LocalInputChannel(
// ------------------------------------------------------------------------
@Override
- protected void requestSubpartition(int subpartitionIndex) throws IOException, InterruptedException {
+ protected void requestSubpartition() throws IOException, InterruptedException {
Review comment:
Could those changes about dropping `int subpartitionIndex` be a separate commit?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services