flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From se...@apache.org
Subject [4/7] flink git commit: [hotfix] [dist. coordination] Move metrics out of the Execution Graph
Date Wed, 29 Mar 2017 22:51:11 GMT
[hotfix] [dist. coordination] Move metrics out of the Execution Graph

ExecutionGraph-based metrics should be in their own package 'org.apache.flink.runtime.executiongraph.metrics'.
They are instantiated by whoever builds the execution graph, but not by the execution graph
itself.
This separates concerns more elegantly.


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

Branch: refs/heads/master
Commit: c277ee17388c85c40d9c3956fe9ac524c3157130
Parents: 85f75a5
Author: Stephan Ewen <sewen@apache.org>
Authored: Wed Mar 29 16:54:17 2017 +0200
Committer: Stephan Ewen <sewen@apache.org>
Committed: Wed Mar 29 17:11:49 2017 +0200

----------------------------------------------------------------------
 .../runtime/executiongraph/ExecutionGraph.java  |  45 +----
 .../executiongraph/ExecutionGraphBuilder.java   |   8 +-
 .../metrics/RestartTimeGauge.java               |  80 +++++++++
 .../executiongraph/metrics/package-info.java    |  23 +++
 ...ExecutionGraphCheckpointCoordinatorTest.java |   4 +-
 .../ExecutionGraphMetricsTest.java              | 173 ++++++-------------
 .../partitioner/RescalePartitionerTest.java     |   5 +-
 7 files changed, 169 insertions(+), 169 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
index 0564fd0..06b2f9a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -29,9 +29,6 @@ import org.apache.flink.api.common.accumulators.Accumulator;
 import org.apache.flink.api.common.accumulators.AccumulatorHelper;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.JobException;
 import org.apache.flink.runtime.StoppingException;
 import org.apache.flink.runtime.accumulators.AccumulatorSnapshot;
@@ -126,8 +123,6 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive
 	/** The log object used for debugging. */
 	static final Logger LOG = LoggerFactory.getLogger(ExecutionGraph.class);
 
-	static final String RESTARTING_TIME_METRIC_NAME = "restartingTime";
-
 	// --------------------------------------------------------------------------------------------
 
 	/** The lock used to secure all access to mutable fields, especially the tracking of progress
@@ -258,9 +253,7 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive
 			Collections.<BlobKey>emptyList(),
 			Collections.<URL>emptyList(),
 			slotProvider,
-			ExecutionGraph.class.getClassLoader(),
-			new UnregisteredMetricsGroup()
-		);
+			ExecutionGraph.class.getClassLoader());
 	}
 
 	public ExecutionGraph(
@@ -275,8 +268,7 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive
 			List<BlobKey> requiredJarFiles,
 			List<URL> requiredClasspaths,
 			SlotProvider slotProvider,
-			ClassLoader userClassLoader,
-			MetricGroup metricGroup) throws IOException {
+			ClassLoader userClassLoader) throws IOException {
 
 		checkNotNull(futureExecutor);
 		checkNotNull(jobId);
@@ -315,9 +307,6 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive
 		this.scheduleAllocationTimeout = checkNotNull(timeout);
 
 		this.restartStrategy = restartStrategy;
-
-		metricGroup.gauge(RESTARTING_TIME_METRIC_NAME, new RestartTimeGauge());
-
 		this.kvStateLocationRegistry = new KvStateLocationRegistry(jobId, getAllVertices());
 	}
 
@@ -1449,36 +1438,6 @@ public class ExecutionGraph implements AccessExecutionGraph, Archiveable<Archive
 		}
 	}
 
-	/**
-	 * Gauge which returns the last restarting time. Restarting time is the time between
-	 * JobStatus.RESTARTING and JobStatus.RUNNING or a terminal state if JobStatus.RUNNING was
not
-	 * reached. If the job has not yet reached either of these states, then the time is measured
-	 * since reaching JobStatus.RESTARTING. If it is still the initial job execution, then the
-	 * gauge will return 0.
-	 */
-	private class RestartTimeGauge implements Gauge<Long> {
-
-		@Override
-		public Long getValue() {
-			long restartingTimestamp = stateTimestamps[JobStatus.RESTARTING.ordinal()];
-
-			if (restartingTimestamp <= 0) {
-				// we haven't yet restarted our job
-				return 0L;
-			} else if (stateTimestamps[JobStatus.RUNNING.ordinal()] >= restartingTimestamp) {
-				// we have transitioned to RUNNING since the last restart
-				return stateTimestamps[JobStatus.RUNNING.ordinal()] - restartingTimestamp;
-			} else if (state.isTerminalState()) {
-				// since the last restart we've switched to a terminal state without touching
-				// the RUNNING state (e.g. failing from RESTARTING)
-				return stateTimestamps[state.ordinal()] - restartingTimestamp;
-			} else {
-				// we're still somwhere between RESTARTING and RUNNING
-				return System.currentTimeMillis() - restartingTimestamp;
-			}
-		}
-	}
-
 	@Override
 	public ArchivedExecutionGraph archive() {
 		Map<JobVertexID, ArchivedExecutionJobVertex> archivedTasks = new HashMap<>();

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
index 8471178..494b7a2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraphBuilder.java
@@ -33,6 +33,7 @@ import org.apache.flink.runtime.checkpoint.CompletedCheckpointStore;
 import org.apache.flink.runtime.checkpoint.CheckpointStatsTracker;
 import org.apache.flink.runtime.client.JobExecutionException;
 import org.apache.flink.runtime.client.JobSubmissionException;
+import org.apache.flink.runtime.executiongraph.metrics.RestartTimeGauge;
 import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
 import org.apache.flink.runtime.instance.SlotProvider;
 import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -102,8 +103,7 @@ public class ExecutionGraphBuilder {
 						jobGraph.getUserJarBlobKeys(),
 						jobGraph.getClasspaths(),
 						slotProvider,
-						classLoader,
-						metrics);
+						classLoader);
 		} catch (IOException e) {
 			throw new JobException("Could not create the execution graph.", e);
 		}
@@ -250,6 +250,10 @@ public class ExecutionGraphBuilder {
 					checkpointStatsTracker);
 		}
 
+		// create all the metrics for the Execution Graph
+
+		metrics.gauge(RestartTimeGauge.METRIC_NAME, new RestartTimeGauge(executionGraph));
+
 		return executionGraph;
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/RestartTimeGauge.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/RestartTimeGauge.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/RestartTimeGauge.java
new file mode 100644
index 0000000..e0a22e3
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/RestartTimeGauge.java
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph.metrics;
+
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobgraph.JobStatus;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Gauge which returns the last restarting time.
+ * 
+ * <p>Restarting time is the time between {@link JobStatus#RESTARTING} and {@link JobStatus#RUNNING},
+ * or a terminal state if {@link JobStatus#RUNNING} was not reached.
+ * 
+ * <p>If the job has not yet reached either of these states, then the time is measured
since reaching
+ * {@link JobStatus#RESTARTING}. If it is still the initial job execution, then the gauge
will return 0.
+ */
+public class RestartTimeGauge implements Gauge<Long> {
+
+	public static final String METRIC_NAME = "restartingTime";
+
+	// ------------------------------------------------------------------------
+
+	private final ExecutionGraph eg;
+
+	public RestartTimeGauge(ExecutionGraph executionGraph) {
+		this.eg = checkNotNull(executionGraph);
+	}
+
+	// ------------------------------------------------------------------------
+
+	@Override
+	public Long getValue() {
+		final JobStatus status = eg.getState();
+
+		final long restartingTimestamp = eg.getStatusTimestamp(JobStatus.RESTARTING);
+
+		final long switchToRunningTimestamp;
+		final long lastRestartTime;
+
+		if (restartingTimestamp <= 0) {
+			// we haven't yet restarted our job
+			return 0L;
+		}
+		else if ((switchToRunningTimestamp = eg.getStatusTimestamp(JobStatus.RUNNING)) >= restartingTimestamp)
{
+			// we have transitioned to RUNNING since the last restart
+			lastRestartTime = switchToRunningTimestamp - restartingTimestamp;
+		}
+		else if (status.isTerminalState()) {
+			// since the last restart we've switched to a terminal state without touching
+			// the RUNNING state (e.g. failing from RESTARTING)
+			lastRestartTime = eg.getStatusTimestamp(status) - restartingTimestamp;
+		}
+		else {
+			// we're still somewhere between RESTARTING and RUNNING
+			lastRestartTime  = System.currentTimeMillis() - restartingTimestamp;
+		}
+
+		// we guard this with 'Math.max' to avoid negative timestamps when clocks re-sync 
+		return Math.max(lastRestartTime, 0);
+	}
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/package-info.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/package-info.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/package-info.java
new file mode 100644
index 0000000..8b9d205
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/metrics/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * This package contains classes implementing various metrics for the job execution,
+ * based on the {@link org.apache.flink.runtime.executiongraph.ExecutionGraph Execution Graph}.
+ */
+package org.apache.flink.runtime.executiongraph.metrics;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
index 81162b6..0ab031e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java
@@ -22,7 +22,6 @@ import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.time.Time;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.blob.BlobKey;
 import org.apache.flink.runtime.executiongraph.ExecutionGraph;
 import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
@@ -95,8 +94,7 @@ public class ExecutionGraphCheckpointCoordinatorTest {
 			Collections.<BlobKey>emptyList(),
 			Collections.<URL>emptyList(),
 			new Scheduler(TestingUtils.defaultExecutionContext()),
-			ClassLoader.getSystemClassLoader(),
-			new UnregisteredMetricsGroup());
+			ClassLoader.getSystemClassLoader());
 
 		executionGraph.enableCheckpointing(
 				100,

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
index 5496e35..97127c7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphMetricsTest.java
@@ -20,13 +20,7 @@ package org.apache.flink.runtime.executiongraph;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.time.Time;
-import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.Gauge;
-import org.apache.flink.metrics.Metric;
-import org.apache.flink.metrics.MetricConfig;
-import org.apache.flink.metrics.MetricGroup;
-import org.apache.flink.metrics.reporter.MetricReporter;
 import org.apache.flink.runtime.JobException;
 import org.apache.flink.runtime.blob.BlobKey;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
@@ -35,13 +29,9 @@ import org.apache.flink.runtime.concurrent.impl.FlinkCompletableFuture;
 import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.execution.SuppressRestartsException;
+import org.apache.flink.runtime.executiongraph.metrics.RestartTimeGauge;
 import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
 import org.apache.flink.runtime.instance.Instance;
-import org.apache.flink.runtime.jobmanager.slots.AllocatedSlot;
-import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
-import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
-import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 import org.apache.flink.runtime.instance.SimpleSlot;
 import org.apache.flink.runtime.instance.Slot;
 import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -49,9 +39,11 @@ import org.apache.flink.runtime.jobgraph.JobStatus;
 import org.apache.flink.runtime.jobgraph.JobVertex;
 import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
 import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
-import org.apache.flink.runtime.metrics.MetricRegistry;
-import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
+import org.apache.flink.runtime.jobmanager.slots.AllocatedSlot;
+import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.messages.Acknowledge;
 import org.apache.flink.runtime.taskmanager.TaskExecutionState;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
 import org.apache.flink.runtime.testtasks.NoOpInvokable;
 import org.apache.flink.util.SerializedValue;
 import org.apache.flink.util.TestLogger;
@@ -63,17 +55,16 @@ import java.io.IOException;
 import java.net.URL;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
-
-import static org.mockito.Mockito.*;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyBoolean;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class ExecutionGraphMetricsTest extends TestLogger {
 
@@ -86,32 +77,14 @@ public class ExecutionGraphMetricsTest extends TestLogger {
 		try {
 			// setup execution graph with mocked scheduling logic
 			int parallelism = 1;
-	
+
 			JobVertex jobVertex = new JobVertex("TestVertex");
 			jobVertex.setParallelism(parallelism);
 			jobVertex.setInvokableClass(NoOpInvokable.class);
 			JobGraph jobGraph = new JobGraph("Test Job", jobVertex);
-	
-			Configuration config = new Configuration();
-			config.setString(ConfigConstants.METRICS_REPORTERS_LIST, "test");
-			config.setString(ConfigConstants.METRICS_REPORTER_PREFIX + "test." + ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX,
TestingReporter.class.getName());
-	
+
 			Configuration jobConfig = new Configuration();
-	
 			Time timeout = Time.seconds(10L);
-	
-			MetricRegistry metricRegistry = new MetricRegistry(MetricRegistryConfiguration.fromConfiguration(config));
-	
-			assertTrue(metricRegistry.getReporters().size() == 1);
-	
-			MetricReporter reporter = metricRegistry.getReporters().get(0);
-	
-			assertTrue(reporter instanceof TestingReporter);
-	
-			TestingReporter testingReporter = (TestingReporter) reporter;
-	
-			MetricGroup metricGroup = new JobManagerMetricGroup(metricRegistry, "localhost");
-	
 			Scheduler scheduler = mock(Scheduler.class);
 
 			ResourceID taskManagerId = ResourceID.generate();
@@ -163,163 +136,127 @@ public class ExecutionGraphMetricsTest extends TestLogger {
 				Collections.<BlobKey>emptyList(),
 				Collections.<URL>emptyList(),
 				scheduler,
-				getClass().getClassLoader(),
-				metricGroup);
-	
-			// get restarting time metric
-			Metric metric = testingReporter.getMetric(ExecutionGraph.RESTARTING_TIME_METRIC_NAME);
-	
-			assertNotNull(metric);
-			assertTrue(metric instanceof Gauge);
-	
-			@SuppressWarnings("unchecked")
-			Gauge<Long> restartingTime = (Gauge<Long>) metric;
-	
+				getClass().getClassLoader());
+
+			RestartTimeGauge restartingTime = new RestartTimeGauge(executionGraph);
+
 			// check that the restarting time is 0 since it's the initial start
-			assertTrue(0L == restartingTime.getValue());
-	
+			assertEquals(0L, restartingTime.getValue().longValue());
+
 			executionGraph.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
-	
+
 			// start execution
-		executionGraph.scheduleForExecution();
+			executionGraph.scheduleForExecution();
+			assertEquals(0L, restartingTime.getValue().longValue());
 
-			assertTrue(0L == restartingTime.getValue());
-	
 			List<ExecutionAttemptID> executionIDs = new ArrayList<>();
-	
+
 			for (ExecutionVertex executionVertex: executionGraph.getAllExecutionVertices()) {
 				executionIDs.add(executionVertex.getCurrentExecutionAttempt().getAttemptId());
 			}
-	
+
 			// tell execution graph that the tasks are in state running --> job status switches
to state running
 			for (ExecutionAttemptID executionID : executionIDs) {
 				executionGraph.updateState(new TaskExecutionState(jobGraph.getJobID(), executionID, ExecutionState.RUNNING));
 			}
-	
+
 			assertEquals(JobStatus.RUNNING, executionGraph.getState());
-	
-			assertTrue(0L == restartingTime.getValue());
-	
+			assertEquals(0L, restartingTime.getValue().longValue());
+
 			// fail the job so that it goes into state restarting
 			for (ExecutionAttemptID executionID : executionIDs) {
 				executionGraph.updateState(new TaskExecutionState(jobGraph.getJobID(), executionID, ExecutionState.FAILED,
new Exception()));
 			}
-	
+
 			assertEquals(JobStatus.RESTARTING, executionGraph.getState());
-	
+
 			long firstRestartingTimestamp = executionGraph.getStatusTimestamp(JobStatus.RESTARTING);
-	
+
 			// wait some time so that the restarting time gauge shows a value different from 0
 			Thread.sleep(50);
-	
+
 			long previousRestartingTime = restartingTime.getValue();
-	
+
 			// check that the restarting time is monotonically increasing
 			for (int i = 0; i < 10; i++) {
 				long currentRestartingTime = restartingTime.getValue();
-	
+
 				assertTrue(currentRestartingTime >= previousRestartingTime);
 				previousRestartingTime = currentRestartingTime;
 			}
-	
+
 			// check that we have measured some restarting time
 			assertTrue(previousRestartingTime > 0);
-	
+
 			// restart job
 			testingRestartStrategy.restartExecutionGraph();
-	
+
 			executionIDs.clear();
-	
+
 			for (ExecutionVertex executionVertex: executionGraph.getAllExecutionVertices()) {
 				executionIDs.add(executionVertex.getCurrentExecutionAttempt().getAttemptId());
 			}
-	
+
 			for (ExecutionAttemptID executionID : executionIDs) {
 				executionGraph.updateState(new TaskExecutionState(jobGraph.getJobID(), executionID, ExecutionState.RUNNING));
 			}
-	
+
 			assertEquals(JobStatus.RUNNING, executionGraph.getState());
-	
+
 			assertTrue(firstRestartingTimestamp != 0);
-	
+
 			previousRestartingTime = restartingTime.getValue();
-	
+
 			// check that the restarting time does not increase after we've reached the running state
 			for (int i = 0; i < 10; i++) {
 				long currentRestartingTime = restartingTime.getValue();
-	
+
 				assertTrue(currentRestartingTime == previousRestartingTime);
 				previousRestartingTime = currentRestartingTime;
 			}
-	
+
 			// fail job again
 			for (ExecutionAttemptID executionID : executionIDs) {
 				executionGraph.updateState(new TaskExecutionState(jobGraph.getJobID(), executionID, ExecutionState.FAILED,
new Exception()));
 			}
-	
+
 			assertEquals(JobStatus.RESTARTING, executionGraph.getState());
-	
+
 			long secondRestartingTimestamp = executionGraph.getStatusTimestamp(JobStatus.RESTARTING);
-	
+
 			assertTrue(firstRestartingTimestamp != secondRestartingTimestamp);
-	
+
 			Thread.sleep(50);
-	
+
 			previousRestartingTime = restartingTime.getValue();
-	
+
 			// check that the restarting time is increasing again
 			for (int i = 0; i < 10; i++) {
 				long currentRestartingTime = restartingTime.getValue();
-	
+
 				assertTrue(currentRestartingTime >= previousRestartingTime);
 				previousRestartingTime = currentRestartingTime;
 			}
-	
+
 			assertTrue(previousRestartingTime > 0);
-	
+
 			// now lets fail the job while it is in restarting and see whether the restarting time
then stops to increase
 			// for this to work, we have to use a SuppressRestartException
 			executionGraph.fail(new SuppressRestartsException(new Exception()));
-	
+
 			assertEquals(JobStatus.FAILED, executionGraph.getState());
-	
+
 			previousRestartingTime = restartingTime.getValue();
-	
+
 			for (int i = 0; i < 10; i++) {
 				long currentRestartingTime = restartingTime.getValue();
-	
+
 				assertTrue(currentRestartingTime == previousRestartingTime);
 				previousRestartingTime = currentRestartingTime;
 			}
 		} finally {
 			executor.shutdownNow();
 		}
-
-	}
-
-	public static class TestingReporter implements MetricReporter {
-
-		private final Map<String, Metric> metrics = new HashMap<>();
-
-		@Override
-		public void open(MetricConfig config) {}
-
-		@Override
-		public void close() {}
-
-		@Override
-		public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
-			metrics.put(metricName, metric);
-		}
-
-		@Override
-		public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group)
{
-			metrics.remove(metricName);
-		}
-
-		Metric getMetric(String metricName) {
-			return metrics.get(metricName);
-		}
 	}
 
 	static class TestingRestartStrategy implements RestartStrategy {

http://git-wip-us.apache.org/repos/asf/flink/blob/c277ee17/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
index 43fe169..d72c37b 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.java.tuple.Tuple;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.configuration.Configuration;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.JobException;
 import org.apache.flink.runtime.akka.AkkaUtils;
 import org.apache.flink.runtime.blob.BlobKey;
@@ -44,6 +43,7 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
 import org.apache.flink.util.Collector;
 import org.apache.flink.util.SerializedValue;
 import org.apache.flink.util.TestLogger;
+
 import org.junit.Before;
 import org.junit.Test;
 
@@ -150,8 +150,7 @@ public class RescalePartitionerTest extends TestLogger {
 			new ArrayList<BlobKey>(),
 			new ArrayList<URL>(),
 			new Scheduler(TestingUtils.defaultExecutionContext()),
-			ExecutionGraph.class.getClassLoader(),
-			new UnregisteredMetricsGroup());
+			ExecutionGraph.class.getClassLoader());
 		try {
 			eg.attachJobGraph(jobVertices);
 		}


Mime
View raw message