You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by uc...@apache.org on 2016/02/20 00:16:10 UTC
flink git commit: [FLINK-3446] [runtime-web] Don't trigger back
pressure sample for archived job
Repository: flink
Updated Branches:
refs/heads/master a3b9d9717 -> 5aa7c254a
[FLINK-3446] [runtime-web] Don't trigger back pressure sample for archived job
This closes #1673.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5aa7c254
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5aa7c254
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5aa7c254
Branch: refs/heads/master
Commit: 5aa7c254a5b6f38d7a7cab0893fc87efe6da9a44
Parents: a3b9d97
Author: Ufuk Celebi <uc...@apache.org>
Authored: Thu Feb 18 22:29:33 2016 +0100
Committer: Ufuk Celebi <uc...@apache.org>
Committed: Sat Feb 20 00:15:41 2016 +0100
----------------------------------------------------------------------
.../webmonitor/BackPressureStatsTracker.java | 34 +++++++++++++-------
.../BackPressureStatsTrackerITCase.java | 11 ++++++-
.../BackPressureStatsTrackerTest.java | 7 ++--
3 files changed, 38 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa7c254/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java
index ff0573a..3eb2f58 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTracker.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Option;
+import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;
import scala.concurrent.duration.FiniteDuration;
@@ -150,26 +151,37 @@ public class BackPressureStatsTracker {
* is ignored.
*
* @param vertex Operator to get the stats for.
+ * @return Flag indicating whether a sample with triggered.
*/
@SuppressWarnings("unchecked")
- public void triggerStackTraceSample(ExecutionJobVertex vertex) {
+ public boolean triggerStackTraceSample(ExecutionJobVertex vertex) {
synchronized (lock) {
if (shutDown) {
- return;
+ return false;
}
- if (!pendingStats.contains(vertex)) {
- pendingStats.add(vertex);
+ if (!pendingStats.contains(vertex) &&
+ !vertex.getGraph().getState().isTerminalState()) {
- Future<StackTraceSample> sample = coordinator.triggerStackTraceSample(
- vertex.getTaskVertices(),
- numSamples,
- delayBetweenSamples,
- MAX_STACK_TRACE_DEPTH);
+ ExecutionContext executionContext = vertex.getGraph().getExecutionContext();
- sample.onComplete(new StackTraceSampleCompletionCallback(
- vertex), vertex.getGraph().getExecutionContext());
+ // Only trigger if still active job
+ if (executionContext != null) {
+ pendingStats.add(vertex);
+
+ Future<StackTraceSample> sample = coordinator.triggerStackTraceSample(
+ vertex.getTaskVertices(),
+ numSamples,
+ delayBetweenSamples,
+ MAX_STACK_TRACE_DEPTH);
+
+ sample.onComplete(new StackTraceSampleCompletionCallback(vertex), executionContext);
+
+ return true;
+ }
}
+
+ return false;
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa7c254/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
index 52b0794..2659185 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerITCase.java
@@ -55,6 +55,8 @@ import static org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.Ex
import static org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.RequestExecutionGraph;
import static org.apache.flink.runtime.testingUtils.TestingJobManagerMessages.WaitForAllVerticesToBeRunning;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
@@ -238,6 +240,13 @@ public class BackPressureStatsTrackerITCase extends TestLogger {
// Response to removal notification
expectMsgEquals(true);
+
+ //
+ // 3) Trigger stats for archived job
+ //
+ statsTracker.invalidateOperatorStatsCache();
+ assertFalse("Unexpected trigger", statsTracker.triggerStackTraceSample(vertex));
+
} catch (Exception e) {
e.printStackTrace();
fail(e.getMessage());
@@ -265,7 +274,7 @@ public class BackPressureStatsTrackerITCase extends TestLogger {
ExecutionJobVertex vertex) throws InterruptedException {
statsTracker.invalidateOperatorStatsCache();
- statsTracker.triggerStackTraceSample(vertex);
+ assertTrue("Failed to trigger", statsTracker.triggerStackTraceSample(vertex));
// Sleep minimum duration
Thread.sleep(20 * 10);
http://git-wip-us.apache.org/repos/asf/flink/blob/5aa7c254/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java
index b0955e1..b2b0afd 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/BackPressureStatsTrackerTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.jobgraph.JobStatus;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.junit.Test;
import scala.concurrent.ExecutionContext;
@@ -37,6 +38,7 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
@@ -62,6 +64,7 @@ public class BackPressureStatsTrackerTest {
anyInt())).thenReturn(samplePromise.future());
ExecutionGraph graph = mock(ExecutionGraph.class);
+ when(graph.getState()).thenReturn(JobStatus.RUNNING);
// Same Thread execution context
when(graph.getExecutionContext()).thenReturn(new ExecutionContext() {
@@ -102,7 +105,7 @@ public class BackPressureStatsTrackerTest {
sampleCoordinator, 9999, numSamples, delayBetweenSamples);
// Trigger
- tracker.triggerStackTraceSample(jobVertex);
+ assertTrue("Failed to trigger", tracker.triggerStackTraceSample(jobVertex));
verify(sampleCoordinator).triggerStackTraceSample(
eq(taskVertices),
@@ -111,7 +114,7 @@ public class BackPressureStatsTrackerTest {
eq(BackPressureStatsTracker.MAX_STACK_TRACE_DEPTH));
// Trigger again for pending request, should not fire
- tracker.triggerStackTraceSample(jobVertex);
+ assertFalse("Unexpected trigger", tracker.triggerStackTraceSample(jobVertex));
assertTrue(tracker.getOperatorBackPressureStats(jobVertex).isEmpty());