You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/08/31 17:28:43 UTC
[25/27] flink git commit: [hotfix] Improve Logging in
CheckpointCoordinator, StreamTask, State Code
[hotfix] Improve Logging in CheckpointCoordinator, StreamTask, State Code
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bdf9f86c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bdf9f86c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bdf9f86c
Branch: refs/heads/master
Commit: bdf9f86c51aa2122441adea6944ce46f393d729f
Parents: f7ef82b
Author: Stefan Richter <s....@data-artisans.com>
Authored: Wed Aug 31 11:19:13 2016 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Wed Aug 31 19:10:02 2016 +0200
----------------------------------------------------------------------
.../flink/runtime/checkpoint/CheckpointCoordinator.java | 4 ++--
.../runtime/state/filesystem/FsCheckpointStreamFactory.java | 4 +++-
.../org/apache/flink/streaming/runtime/tasks/StreamTask.java | 7 +++++--
3 files changed, 10 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bdf9f86c/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
index 52f6d9a..3586d98 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.checkpoint;
import akka.dispatch.Futures;
-
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.checkpoint.savepoint.SavepointStore;
import org.apache.flink.runtime.checkpoint.stats.CheckpointStatsTracker;
@@ -42,7 +41,6 @@ import org.apache.flink.runtime.state.StreamStateHandle;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
import scala.concurrent.Future;
import java.util.ArrayDeque;
@@ -778,6 +776,8 @@ public class CheckpointCoordinator {
}
}
+ LOG.info("Restoring from latest valid checkpoint: {}.", latest);
+
for (Map.Entry<JobVertexID, TaskState> taskGroupStateEntry: latest.getTaskStates().entrySet()) {
TaskState taskState = taskGroupStateEntry.getValue();
ExecutionJobVertex executionJobVertex = tasks.get(taskGroupStateEntry.getKey());
http://git-wip-us.apache.org/repos/asf/flink/blob/bdf9f86c/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
index cc13a72..c027558 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/filesystem/FsCheckpointStreamFactory.java
@@ -97,7 +97,9 @@ public class FsCheckpointStreamFactory implements CheckpointStreamFactory {
Path dir = new Path(basePath, jobId.toString());
- LOG.info("Initializing file stream factory to URI {}.", dir);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Initializing file stream factory to URI {}.", dir);
+ }
filesystem = basePath.getFileSystem();
filesystem.mkdirs(dir);
http://git-wip-us.apache.org/repos/asf/flink/blob/bdf9f86c/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index bedc8fa..9c26509 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -970,10 +970,13 @@ public abstract class StreamTask<OUT, Operator extends StreamOperator<OUT>>
if (chainedStateHandles.isEmpty() && keyedStates.isEmpty()) {
owner.getEnvironment().acknowledgeCheckpoint(checkpointId);
} else {
- owner. getEnvironment().acknowledgeCheckpoint(checkpointId, chainedStateHandles, keyedStates);
+ owner.getEnvironment().acknowledgeCheckpoint(checkpointId, chainedStateHandles, keyedStates);
}
- LOG.debug("Finished asynchronous checkpoints for checkpoint {} on task {}", checkpointId, name);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Finished asynchronous checkpoints for checkpoint {} on task {}. Returning handles on " +
+ "keyed states {}.", checkpointId, name, keyedStates);
+ }
}
catch (Exception e) {
if (owner.isRunning()) {