You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/10/25 12:39:57 UTC
[2/2] flink git commit: [FLINK-4218] [checkpoints] Do not fail
checkpoints when state size cannot be determined
[FLINK-4218] [checkpoints] Do not fail checkpoints when state size cannot be determined
Eventually consistent file systems (like S3) may fail on calls to determine the state size.
Since the state size is only informational (display in metrics) this should be tolerated and
not fail a checkpoint.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/198f2b07
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/198f2b07
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/198f2b07
Branch: refs/heads/release-1.1
Commit: 198f2b072761d6e6ca93ff5e93bf6391ed736260
Parents: 6624348
Author: Stephan Ewen <se...@apache.org>
Authored: Mon Oct 24 15:35:57 2016 +0200
Committer: Stephan Ewen <se...@apache.org>
Committed: Tue Oct 25 14:38:17 2016 +0200
----------------------------------------------------------------------
.../filesystem/AbstractFileStateHandle.java | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/198f2b07/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateHandle.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateHandle.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateHandle.java
index 0585062..e9d54dc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateHandle.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/AbstractFileStateHandle.java
@@ -22,7 +22,10 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.AbstractCloseableHandle;
import org.apache.flink.runtime.state.StateObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.FileNotFoundException;
import java.io.IOException;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -34,6 +37,8 @@ public abstract class AbstractFileStateHandle extends AbstractCloseableHandle im
private static final long serialVersionUID = 350284443258002355L;
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractFileStateHandle.class);
+
/** The path to the file in the filesystem, fully describing the file system */
private final Path filePath;
@@ -93,6 +98,18 @@ public abstract class AbstractFileStateHandle extends AbstractCloseableHandle im
* @throws IOException Thrown if the file system cannot be accessed.
*/
protected long getFileSize() throws IOException {
- return getFileSystem().getFileStatus(filePath).getLen();
+ try {
+ return getFileSystem().getFileStatus(filePath).getLen();
+ }
+ catch (FileNotFoundException e) {
+ LOG.info("Could not determine state size - state will appear as size '0'. " +
+ "If the underlying filesystem is eventually consistency, that is most likely the cause.");
+ }
+ catch (IOException e) {
+ // this can happen on eventually consistent file systems (like for example S3)
+ LOG.info("Could not determine state size - state will appear as size '0'.", e);
+ }
+
+ return 0;
}
}