You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by zentol <gi...@git.apache.org> on 2016/03/15 13:14:44 UTC
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
GitHub user zentol opened a pull request:
https://github.com/apache/flink/pull/1794
[FLINK-3492] Configurable interval between Checkpoints
This PR allows the user to set a minimum interval between checkpoints. This interval is used if a checkpoint is triggered, while another one is still in progress. At that point, the next checkpoint trigger is scheduled for timeOfLastCompletedCheckpoint + minInterval.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/zentol/flink 3492_checkpoint_pause
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/1794.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1794
----
commit 541456da4c3cd5745a39da92b1457a90f163ca17
Author: zentol <ch...@apache.org>
Date: 2016-03-15T11:13:23Z
[FLINK-3492] Configurable interval between Checkpoints
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56155005
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---
@@ -277,6 +277,29 @@ public StreamExecutionEnvironment enableCheckpointing(long interval) {
/**
* Enables checkpointing for the streaming job. The distributed state of the streaming
* dataflow will be periodically snapshotted. In case of a failure, the streaming
+ * dataflow will be restarted from the latest completed checkpoint. This method selects
+ * {@link CheckpointingMode#EXACTLY_ONCE} guarantees.
+ *
+ * <p>The job draws checkpoints periodically, in the given interval. The state will be
+ * stored in the configured state backend.</p>
+ *
+ * <p>NOTE: Checkpointing iterative streaming dataflows in not properly supported at
+ * the moment. For that reason, iterative jobs will not be started if used
+ * with enabled checkpointing. To override this mechanism, use the
+ * {@link #enableCheckpointing(long, CheckpointingMode, boolean)} method.</p>
+ *
+ * @param interval Time interval between state checkpoints in milliseconds.
+ * @param minInterval Time interval between the last finished checkpoint and the next one in mills.
+ */
+ public StreamExecutionEnvironment enableCheckpointing(long interval, long minInterval) {
--- End diff --
`minInterval` might be confusing for users because it actually is the pause between successive checkpoints.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-196800536
Good work @zentol :-) I had some inline comments. It could be good to add a test case for the min pause interval.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r59208369
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java ---
@@ -1066,26 +1069,23 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
coord.startCheckpointScheduler();
- //trigger first checkpoint
- Thread.sleep(100);
-
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //new checkpoint has been triggered
- Thread.sleep(300);
- assertEquals(2, numCalls.get());
+ for (int x=0; x<20; x++) {
+ Thread.sleep(100);
+ if (numCalls.get() > 0) {
--- End diff --
@uce done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-205306447
How about adjusting the test such that it works the following way:
- it polls the number of checkpoint calls (from the atomic integer), say every 100 ms
- it starts measuring time from when the count goes to `1`
- it validates that `(numCalls - 2) * pause` is always smaller than the elapsed time since starting the time measurement
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56154469
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -921,11 +921,20 @@ public ActorGateway createActivatorDeactivator(ActorSystem actorSystem, UUID lea
// ------------------------------------------------------------------------
private class ScheduledTrigger extends TimerTask {
+ private final long processingInterval;
+ private long lastCheckpoint = 0;
+
+ public ScheduledTrigger(long guaranteedInterval) {
+ this.processingInterval = guaranteedInterval;
+ }
@Override
public void run() {
try {
- triggerCheckpoint(System.currentTimeMillis());
+ long time = lastCheckpoint;
--- End diff --
where do you update `lastCheckpoint`? If I'm not mistaken, then it should always stay `0`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-208438856
merged in 2e63d1afb2358760109ba0c90011ef565eaae0ff
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol closed the pull request at:
https://github.com/apache/flink/pull/1794
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197340895
you are correct @uce , looking back that's what the ScheduledTrigger was supposed to do; which i reverted a bit too quickly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r59193042
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java ---
@@ -1066,26 +1069,23 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
coord.startCheckpointScheduler();
- //trigger first checkpoint
- Thread.sleep(100);
-
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //new checkpoint has been triggered
- Thread.sleep(300);
- assertEquals(2, numCalls.get());
+ for (int x=0; x<20; x++) {
+ Thread.sleep(100);
+ if (numCalls.get() > 0) {
--- End diff --
Since you are waiting on this condition, I would add a check after the loop with a good error message if <= 0 in case that the call did not happen in 20 * 100 ms.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-206262455
Very nice!
+1 to merge this
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-196801937
The changes from the PR should also be reflected in the documentation and there are no tests for it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56154789
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -921,11 +921,20 @@ public ActorGateway createActivatorDeactivator(ActorSystem actorSystem, UUID lea
// ------------------------------------------------------------------------
private class ScheduledTrigger extends TimerTask {
+ private final long processingInterval;
+ private long lastCheckpoint = 0;
+
+ public ScheduledTrigger(long guaranteedInterval) {
+ this.processingInterval = guaranteedInterval;
--- End diff --
Why do we need the guaranteedInterval within the `ScheduledTrigger`? Can we be sure that the `time - lastCheckpoint` is always greater than `processingInterval` if the task is executed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197338128
Is it in line with the documentation that this only applies if there are queued requests, i.e. is it OK that a checkpoint might be triggered right after a checkpoint has completed? For example a checkpoint needs X ms to complete and the interval is X+1 ms. There will be no queued requests and the min pause will be ignored, right?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56155661
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---
@@ -277,6 +277,29 @@ public StreamExecutionEnvironment enableCheckpointing(long interval) {
/**
* Enables checkpointing for the streaming job. The distributed state of the streaming
* dataflow will be periodically snapshotted. In case of a failure, the streaming
+ * dataflow will be restarted from the latest completed checkpoint. This method selects
--- End diff --
I think its sufficient to have the configuration parameter in the `CheckpointConfig`, without any convenience methods in the `StreamExecutionEnvironment`.
The `StreamExecutionEnvironment` is getting hard to maintain with those changes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-205887495
reworked test to be more reliable.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197462397
I had a feeling that this solution was too simple to be correct, and in fact it is the case.
on the bright side, i have a pretty good idea on how to solve this properly now, stay tuned.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56155936
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java ---
@@ -921,11 +921,20 @@ public ActorGateway createActivatorDeactivator(ActorSystem actorSystem, UUID lea
// ------------------------------------------------------------------------
private class ScheduledTrigger extends TimerTask {
+ private final long processingInterval;
+ private long lastCheckpoint = 0;
+
+ public ScheduledTrigger(long guaranteedInterval) {
+ this.processingInterval = guaranteedInterval;
--- End diff --
hmm..yeah we don't need it. remnants of the initial idea.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197832914
Third time's the charm:
* the time of the last triggered checkpoint is stored
* when a checkpoint is triggered we check whether the interval has passed, if so we reschedule the next checkpoint with a delay of minInterval
* test added
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r59193164
--- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinatorTest.java ---
@@ -1066,26 +1069,23 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
coord.startCheckpointScheduler();
- //trigger first checkpoint
- Thread.sleep(100);
-
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //no new checkpoint has been triggered
- Thread.sleep(100);
- assertEquals(1, numCalls.get());
-
- //new checkpoint has been triggered
- Thread.sleep(300);
- assertEquals(2, numCalls.get());
+ for (int x=0; x<20; x++) {
+ Thread.sleep(100);
+ if (numCalls.get() > 0) {
--- End diff --
And somehow you removed the comments. I would like to add them back (at the end of the test as well) to make it easier to understand what's happening. ;)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-208381444
Thanks, Chesnay! Can you merge this to `master`? :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by StephanEwen <gi...@git.apache.org>.
Github user StephanEwen commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-205303755
The code looks good, but the tests are bound to be unstable. Making the assumption that sleeps actually sleep as long as they should is often violated (seems especially on Travis).
I think we need to find a better way to test this...
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197341137
i mean, the ScheduledTrigger was supposed to make sure that a min Interval is kept between the last completed Checkpoint and the next one to trigger.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/1794#discussion_r56330207
--- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java ---
@@ -277,6 +277,29 @@ public StreamExecutionEnvironment enableCheckpointing(long interval) {
/**
* Enables checkpointing for the streaming job. The distributed state of the streaming
* dataflow will be periodically snapshotted. In case of a failure, the streaming
+ * dataflow will be restarted from the latest completed checkpoint. This method selects
--- End diff --
I agree. Let's remove this variant.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request: [FLINK-3492] Configurable interval between Che...
Posted by zentol <gi...@git.apache.org>.
Github user zentol commented on the pull request:
https://github.com/apache/flink/pull/1794#issuecomment-197385184
I've revisited this PR:
* ScheduledTrigger remains unchanged
* StreamEecutionEnvironment remains unchanged
* if the minIntervalBetweenCheckpoints is > 0 a different schedule method is used
* scheduleAtFixedRate() is used when minInterval == 0. This is the current behaviour, meaning checkpoints can build up if they aren't processed fast enough
* schedule() is used when minInterval > 0. New stuff right here. It schedules the next checkpoint relative to the completion of the last. So we just pop the interval as the period in there and are done.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---