Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/flink/pull/2174#discussion_r69102339
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/ContinuousFileReaderOperator.java
---
@@ -334,9 +342,11 @@ public void run() {
this.pendingSplits.remove();
}
- if (this.format instanceof CheckpointableInputFormat &&
this.isSplitOpen) {
- S formatState = (S)
((CheckpointableInputFormat) format).getCurrentState();
- return new Tuple3<>(snapshot, currentSplit,
currentSplit == null ? null : formatState);
+ if (this.format instanceof CheckpointableInputFormat &&
this.currentSplit != null) {
+ S formatState = this.isSplitOpen ?
+ (S) ((CheckpointableInputFormat)
format).getCurrentState() :
+ restoredFormatState;
+ return new Tuple3<>(snapshot, currentSplit,
formatState);
} else {
LOG.info("The format used is not
checkpointable. The current input split will be restarted upon recovery.");
--- End diff --
Monitoring an empty directory looks like this:
```
11:36:42,704 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
checkpoint 1 @ 1467279402703
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,709 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,710 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:42,823 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
checkpoint 1 (in 120 ms)
11:36:44,703 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
checkpoint 2 @ 1467279404703
11:36:44,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,706 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,706 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,706 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,706 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,706 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,707 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,710 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:44,745 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
checkpoint 2 (in 18 ms)
11:36:46,703 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
checkpoint 3 @ 1467279406703
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:46,723 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
checkpoint 3 (in 8 ms)
11:36:48,703 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Triggering
checkpoint 4 @ 1467279408703
11:36:48,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,704 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,705 INFO
org.apache.flink.streaming.api.functions.source.ContinuousFileReaderOperator -
The format used is not checkpointable. The current input split will be
restarted upon recovery.
11:36:48,729 INFO
org.apache.flink.runtime.checkpoint.CheckpointCoordinator - Completed
checkpoint 4 (in 10 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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
