flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject git commit: [FLINK-1054] Fix job history view in web frontend
Date Fri, 15 Aug 2014 11:53:04 GMT
Repository: incubator-flink
Updated Branches:
  refs/heads/release-0.6 87dd7c7fb -> a05c56748


[FLINK-1054] Fix job history view in web frontend


Project: http://git-wip-us.apache.org/repos/asf/incubator-flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-flink/commit/a05c5674
Tree: http://git-wip-us.apache.org/repos/asf/incubator-flink/tree/a05c5674
Diff: http://git-wip-us.apache.org/repos/asf/incubator-flink/diff/a05c5674

Branch: refs/heads/release-0.6
Commit: a05c567481b51210319bc950f27808cf921c4ee8
Parents: 87dd7c7
Author: Robert Metzger <rmetzger@apache.org>
Authored: Fri Aug 15 13:51:06 2014 +0200
Committer: Robert Metzger <rmetzger@apache.org>
Committed: Fri Aug 15 13:51:06 2014 +0200

----------------------------------------------------------------------
 .../java/org/apache/flink/runtime/jobmanager/JobManager.java    | 5 +++++
 .../flink/runtime/jobmanager/web/JobmanagerInfoServlet.java     | 5 +----
 2 files changed, 6 insertions(+), 4 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a05c5674/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java
index be0bb48..8ef6b58 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/JobManager.java
@@ -44,6 +44,7 @@ import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.GlobalConfiguration;
@@ -1196,4 +1197,8 @@ public class JobManager implements DeploymentManager, ExtendedManagementProtocol
 	public AccumulatorEvent getAccumulatorResults(JobID jobID) throws IOException {
 		return new AccumulatorEvent(jobID, this.accumulatorManager.getJobAccumulators(jobID));
 	}
+	
+	public Map<String, Accumulator<?, ?>> getAccumulators(JobID jobID) {
+		return this.accumulatorManager.getJobAccumulators(jobID);
+	}
 }

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/a05c5674/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobmanagerInfoServlet.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobmanagerInfoServlet.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobmanagerInfoServlet.java
index 3aa17bf..386315f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobmanagerInfoServlet.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/web/JobmanagerInfoServlet.java
@@ -36,13 +36,11 @@ import javax.servlet.http.HttpServletResponse;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.flink.api.common.accumulators.AccumulatorHelper;
-import org.apache.flink.runtime.accumulators.AccumulatorEvent;
 import org.apache.flink.runtime.event.job.AbstractEvent;
 import org.apache.flink.runtime.event.job.ExecutionStateChangeEvent;
 import org.apache.flink.runtime.event.job.JobEvent;
 import org.apache.flink.runtime.event.job.RecentJobEvent;
 import org.apache.flink.runtime.execution.ExecutionState;
-import org.apache.flink.runtime.execution.librarycache.LibraryCacheManager;
 import org.apache.flink.runtime.jobgraph.JobID;
 import org.apache.flink.runtime.jobgraph.JobStatus;
 import org.apache.flink.runtime.jobmanager.JobManager;
@@ -291,8 +289,7 @@ public class JobmanagerInfoServlet extends HttpServlet {
 			wrt.write("],");
 			
 			// write accumulators
-			AccumulatorEvent accumulators = jobmanager.getAccumulatorResults(jobEvent.getJobID());
-			Map<String, Object> accMap = AccumulatorHelper.toResultMap(accumulators.getAccumulators(LibraryCacheManager.getClassLoader(jobEvent.getJobID())));
+			Map<String, Object> accMap = AccumulatorHelper.toResultMap(jobmanager.getAccumulators(jobEvent.getJobID()));
 			
 			wrt.write("\n\"accumulators\": [");
 			int i = 0;


Mime
View raw message