You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fa...@apache.org on 2023/09/09 13:52:32 UTC
[flink] 02/02: [FLINK-33042][flamegraph] Allow trigger flamegraph when task is initializing
This is an automated email from the ASF dual-hosted git repository.
fanrui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 53702736382b80fa905494e69654061a84741e84
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Thu Sep 7 23:46:32 2023 +0800
[FLINK-33042][flamegraph] Allow trigger flamegraph when task is initializing
---
...job-overview-drawer-backpressure.component.html | 9 +-
.../job-overview-drawer-flamegraph.component.html | 11 +-
.../job-overview-drawer-flamegraph.component.ts | 12 +-
.../threadinfo/VertexThreadInfoTracker.java | 5 +-
.../executiongraph/ExecutionGraphTestUtils.java | 10 ++
.../threadinfo/VertexThreadInfoTrackerTest.java | 156 ++++++++++++---------
6 files changed, 125 insertions(+), 78 deletions(-)
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
index 5628679476b..a37ceaf9f9f 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/backpressure/job-overview-drawer-backpressure.component.html
@@ -50,8 +50,13 @@
></flink-dynamic-host>
</span>
</span>
- <span *ngIf="selectedVertex?.detail?.status !== 'RUNNING'">
- Operator is not running. Cannot sample back pressure.
+ <span
+ *ngIf="
+ selectedVertex?.detail?.status !== 'RUNNING' &&
+ selectedVertex?.detail?.status !== 'INITIALIZING'
+ "
+ >
+ Operator is not running or initializing. Cannot sample back pressure.
</span>
</th>
</tr>
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
index 3c34fbc40ba..3939ce93aad 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
@@ -16,8 +16,13 @@
~ limitations under the License.
-->
-<span *ngIf="selectedVertex?.detail?.status !== 'RUNNING'">
- Operator is not running. Cannot sample back pressure.
+<span
+ *ngIf="
+ selectedVertex?.detail?.status !== 'RUNNING' &&
+ selectedVertex?.detail?.status !== 'INITIALIZING'
+ "
+>
+ Operator is not running or initializing. Cannot sample flame graph.
</span>
Type:
@@ -56,7 +61,7 @@ Subtask:
(ngModelChange)="selectSubtask($event)"
>
<nz-option
- *ngFor="let name of listOfRunningSubtasks"
+ *ngFor="let name of listOfSampleableSubtasks"
[nzLabel]="name"
[nzValue]="name"
nzCustomContent
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
index b4a7e92b46a..3eccf1df3e1 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
@@ -60,7 +60,7 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
public selectedVertex: NodesItemCorrect | null;
public flameGraph = {} as JobFlameGraph;
public allSubtasks = 'all';
- public listOfRunningSubtasks: string[] = [this.allSubtasks];
+ public listOfSampleableSubtasks: string[] = [this.allSubtasks];
public graphType = FlameGraphType.ON_CPU;
public subtaskIndex = this.allSubtasks;
@@ -131,17 +131,17 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
)
.subscribe(
data => {
- const runningSubtasks = data?.subtasks
- .filter(subtaskInfo => subtaskInfo.status === 'RUNNING')
+ const sampleableSubtasks = data?.subtasks
+ .filter(subtaskInfo => subtaskInfo.status === 'RUNNING' || subtaskInfo.status === 'INITIALIZING')
.map(subtaskInfo => subtaskInfo.subtask.toString());
- if (isNil(runningSubtasks)) {
+ if (isNil(sampleableSubtasks)) {
return;
}
- this.listOfRunningSubtasks = [this.allSubtasks, ...runningSubtasks];
+ this.listOfSampleableSubtasks = [this.allSubtasks, ...sampleableSubtasks];
this.cdr.markForCheck();
},
() => {
- this.listOfRunningSubtasks = [this.allSubtasks];
+ this.listOfSampleableSubtasks = [this.allSubtasks];
this.cdr.markForCheck();
}
);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTracker.java
index 9e71f256f1a..a0ac5d4f423 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTracker.java
@@ -317,9 +317,10 @@ public class VertexThreadInfoTracker implements VertexStatsTracker<VertexThreadI
new HashMap<>();
for (AccessExecutionVertex executionVertex : executionVertices) {
- if (executionVertex.getExecutionState() != ExecutionState.RUNNING) {
+ if (executionVertex.getExecutionState() != ExecutionState.RUNNING
+ && executionVertex.getExecutionState() != ExecutionState.INITIALIZING) {
LOG.trace(
- "{} not running, but {}; not sampling",
+ "{} not running or initializing, but {}; not sampling",
executionVertex.getTaskNameWithSubtaskIndex(),
executionVertex.getExecutionState());
continue;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
index 8ccb5183dfe..b7b8529a77c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
@@ -229,6 +229,16 @@ public class ExecutionGraphTestUtils {
return (AccessExecution execution) -> execution.getState() == executionState;
}
+ /**
+ * Takes all vertices in the given ExecutionGraph and switches their current execution to
+ * INITIALIZING.
+ */
+ public static void switchAllVerticesToInitializing(ExecutionGraph eg) {
+ for (ExecutionVertex vertex : eg.getAllExecutionVertices()) {
+ vertex.getCurrentExecutionAttempt().switchToInitializing();
+ }
+ }
+
/**
* Takes all vertices in the given ExecutionGraph and switches their current execution to
* RUNNING.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTrackerTest.java
index 86462ad5b5e..d39ecf265f1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoTrackerTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.webmonitor.threadinfo;
import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
+import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils;
@@ -37,6 +38,9 @@ import org.apache.flink.runtime.taskexecutor.TaskExecutorThreadInfoGateway;
import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
import org.apache.flink.runtime.util.JvmUtils;
import org.apache.flink.testutils.TestingUtils;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
import org.apache.flink.shaded.guava31.com.google.common.cache.Cache;
import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder;
@@ -44,9 +48,10 @@ import org.apache.flink.shaded.guava31.com.google.common.cache.RemovalListener;
import org.apache.flink.shaded.guava31.com.google.common.cache.RemovalNotification;
import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableSet;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -78,22 +83,13 @@ import static org.apache.flink.util.Preconditions.checkState;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
/** Tests for the {@link VertexThreadInfoTracker}. */
+@ExtendWith(ParameterizedTestExtension.class)
class VertexThreadInfoTrackerTest {
private static final int REQUEST_ID = 0;
private static final int PARALLELISM = 10;
- private static final ExecutionJobVertex EXECUTION_JOB_VERTEX = createExecutionJobVertex();
- private static final ExecutionAttemptID[] ATTEMPT_IDS =
- Arrays.stream(EXECUTION_JOB_VERTEX.getTaskVertices())
- .map(
- executionVertex ->
- executionVertex.getCurrentExecutionAttempt().getAttemptId())
- .sorted(Comparator.comparingInt(ExecutionAttemptID::getSubtaskIndex))
- .toArray(ExecutionAttemptID[]::new);
private static final JobID JOB_ID = new JobID();
- private static VertexThreadInfoStats threadInfoStatsDefaultSample;
-
private static final Duration CLEAN_UP_INTERVAL = Duration.ofSeconds(60);
private static final Duration STATS_REFRESH_INTERVAL = Duration.ofSeconds(60);
private static final Duration TIME_GAP = Duration.ofSeconds(60);
@@ -106,8 +102,29 @@ class VertexThreadInfoTrackerTest {
private static ScheduledExecutorService executor;
- @BeforeAll
- static void setUp() {
+ private ExecutionJobVertex executionJobVertex;
+
+ private ExecutionAttemptID[] attemptIDS;
+
+ private VertexThreadInfoStats threadInfoStatsDefaultSample;
+
+ @Parameter public ExecutionState executionState;
+
+ @Parameters(name = "executionState={0}")
+ private static Collection<ExecutionState> parameters() {
+ return Arrays.asList(ExecutionState.RUNNING, ExecutionState.INITIALIZING);
+ }
+
+ @BeforeEach
+ void setUp() {
+ executionJobVertex = createExecutionJobVertex();
+ attemptIDS =
+ Arrays.stream(executionJobVertex.getTaskVertices())
+ .map(
+ executionVertex ->
+ executionVertex.getCurrentExecutionAttempt().getAttemptId())
+ .sorted(Comparator.comparingInt(ExecutionAttemptID::getSubtaskIndex))
+ .toArray(ExecutionAttemptID[]::new);
// Time gap determines endTime of stats, which controls if the "refresh" is triggered:
// now >= stats.getEndTime() + statsRefreshInterval
// Using a small gap to be able to test cache updates without much delay.
@@ -115,15 +132,15 @@ class VertexThreadInfoTrackerTest {
executor = Executors.newScheduledThreadPool(1);
}
- @AfterAll
- static void tearDown() {
+ @AfterEach
+ void tearDown() {
if (executor != null) {
executor.shutdownNow();
}
}
/** Tests successful thread info stats request. */
- @Test
+ @TestTemplate
void testGetThreadInfoStats() throws Exception {
doInitialJobVertexRequestAndVerifyResult(createThreadInfoTracker());
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
@@ -132,7 +149,7 @@ class VertexThreadInfoTrackerTest {
}
/** Tests that cached result is reused within refresh interval. */
- @Test
+ @TestTemplate
void testCachedStatsNotUpdatedWithinRefreshInterval() throws Exception {
final VertexThreadInfoStats unusedThreadInfoStats = createThreadInfoStats(1, TIME_GAP);
@@ -145,14 +162,14 @@ class VertexThreadInfoTrackerTest {
// stores threadInfoStatsDefaultSample in job vertex cache and execution vertex cache
doInitialJobVertexRequestAndVerifyResult(tracker);
Optional<VertexThreadInfoStats> result =
- tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
+ tracker.getJobVertexStats(JOB_ID, executionJobVertex);
// cached result is returned instead of unusedThreadInfoStats
assertExpectedEqualsReceived(threadInfoStatsDefaultSample, result);
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsDefaultSample, ATTEMPT_IDS[subtaskIndex]),
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ threadInfoStatsDefaultSample, attemptIDS[subtaskIndex]),
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
}
// Test for trigger request at execution vertex level
@@ -167,13 +184,13 @@ class VertexThreadInfoTrackerTest {
// cached result is returned instead of unusedThreadInfoStats
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsDefaultSample, ATTEMPT_IDS[0]),
- tracker1.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ threadInfoStatsDefaultSample, attemptIDS[0]),
+ tracker1.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
}
}
/** Tests that cached job vertex result is NOT reused after refresh interval. */
- @Test
+ @TestTemplate
void testJobVertexCachedStatsUpdatedAfterRefreshInterval() throws Exception {
final Duration shortRefreshInterval = Duration.ofMillis(1000);
@@ -200,25 +217,25 @@ class VertexThreadInfoTrackerTest {
threadInfoStatsAfterRefresh);
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
// block until the async call completes and the first result is available
tracker.getResultAvailableFuture().get();
// retrieve the entry, triggering the refresh as side effect
assertExpectedEqualsReceived(
- initialThreadInfoStats, tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX));
+ initialThreadInfoStats, tracker.getJobVertexStats(JOB_ID, executionJobVertex));
// wait until the entry is refreshed
cacheRefreshed.await();
// verify that we get the second result on the next request
Optional<VertexThreadInfoStats> result =
- tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
+ tracker.getJobVertexStats(JOB_ID, executionJobVertex);
assertExpectedEqualsReceived(threadInfoStatsAfterRefresh, result);
}
/** Tests that cached execution vertex result is NOT reused after refresh interval. */
- @Test
+ @TestTemplate
void testExecutionVertexCachedStatsUpdatedAfterRefreshInterval() throws Exception {
final Duration shortRefreshInterval = Duration.ofMillis(1000);
@@ -248,7 +265,7 @@ class VertexThreadInfoTrackerTest {
threadInfoStatsAfterRefresh);
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex))
.isNotPresent();
// block until the async call completes and the first result is available
tracker.getResultAvailableFuture().get();
@@ -256,18 +273,18 @@ class VertexThreadInfoTrackerTest {
// retrieve the entry, triggering the refresh as side effect
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- initialThreadInfoStats, ATTEMPT_IDS[subtaskIndex]),
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ initialThreadInfoStats, attemptIDS[subtaskIndex]),
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
// wait until the entry is refreshed
cacheRefreshed.await();
// verify that we get the second result on the next request
Optional<VertexThreadInfoStats> result =
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex);
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex);
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsAfterRefresh, ATTEMPT_IDS[subtaskIndex]),
+ threadInfoStatsAfterRefresh, attemptIDS[subtaskIndex]),
result);
}
}
@@ -276,7 +293,7 @@ class VertexThreadInfoTrackerTest {
* Tests that the execution vertex request should be ignored when the corresponding job vertex
* request is pending.
*/
- @Test
+ @TestTemplate
void testExecutionVertexShouldBeIgnoredWhenJobVertexIsPending() throws Exception {
CompletableFuture<VertexThreadInfoStats> statsFuture = new CompletableFuture<>();
TestingBlockingAndCountableCoordinator coordinator =
@@ -286,13 +303,13 @@ class VertexThreadInfoTrackerTest {
CLEAN_UP_INTERVAL, STATS_REFRESH_INTERVAL, null, null, coordinator);
// Request the job vertex stats and keep pending
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
assertThat(coordinator.getTriggerCounter()).isOne();
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
// These execution vertex requests shouldn't return any result, and should be ignored
// directly due to the corresponding job vertex is pending.
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex))
.isNotPresent();
assertThat(tracker.getResultAvailableFuture()).isNotCompleted();
assertThat(coordinator.getTriggerCounter()).isOne();
@@ -303,7 +320,7 @@ class VertexThreadInfoTrackerTest {
tracker.getResultAvailableFuture().get();
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex))
.isPresent();
// These execution vertex requests still should be ignored due to cached result is
// reused within refresh interval.
@@ -312,7 +329,7 @@ class VertexThreadInfoTrackerTest {
}
/** Tests that cached results are removed within the cleanup interval. */
- @Test
+ @TestTemplate
void testCachedStatsCleanedAfterCleanupInterval() throws Exception {
final Duration shortCleanUpInterval = Duration.ofMillis(1);
@@ -330,11 +347,11 @@ class VertexThreadInfoTrackerTest {
threadInfoStatsDefaultSample);
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
// wait until one eviction was registered
cacheExpired.await();
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
// Test for trigger request at execution vertex level
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
@@ -356,20 +373,20 @@ class VertexThreadInfoTrackerTest {
// no stats yet, but the request triggers async collection of stats
assertThat(
executionTracker.getExecutionVertexStats(
- JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ JOB_ID, executionJobVertex, subtaskIndex))
.isNotPresent();
// wait until one eviction was registered
executionCacheExpired.await();
assertThat(
executionTracker.getExecutionVertexStats(
- JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ JOB_ID, executionJobVertex, subtaskIndex))
.isNotPresent();
}
}
/** Tests that cached results are NOT removed within the cleanup interval. */
- @Test
+ @TestTemplate
void testCachedStatsNotCleanedWithinCleanupInterval() throws Exception {
final VertexThreadInfoTracker tracker = createThreadInfoTracker();
@@ -379,17 +396,17 @@ class VertexThreadInfoTrackerTest {
// the thread info stats with the same requestId should still be there
assertExpectedEqualsReceived(
threadInfoStatsDefaultSample,
- tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX));
+ tracker.getJobVertexStats(JOB_ID, executionJobVertex));
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsDefaultSample, ATTEMPT_IDS[subtaskIndex]),
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ threadInfoStatsDefaultSample, attemptIDS[subtaskIndex]),
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
}
}
/** Tests that cached results are not served after the shutdown. */
- @Test
+ @TestTemplate
void testShutDown() throws Exception {
final VertexThreadInfoTracker tracker = createThreadInfoTracker();
doInitialJobVertexRequestAndVerifyResult(tracker);
@@ -398,11 +415,11 @@ class VertexThreadInfoTrackerTest {
tracker.shutDown();
// verify that the previous cached result is invalid and trigger another request
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, 0)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, 0)).isNotPresent();
// verify no response after shutdown
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, 0)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, 0)).isNotPresent();
}
private <K> Cache<K, VertexThreadInfoStats> createCache(
@@ -417,17 +434,17 @@ class VertexThreadInfoTrackerTest {
private void doInitialJobVertexRequestAndVerifyResult(VertexThreadInfoTracker tracker)
throws InterruptedException, ExecutionException {
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
+ assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
// block until the async call completes and the first result is available
tracker.getResultAvailableFuture().get();
assertExpectedEqualsReceived(
threadInfoStatsDefaultSample,
- tracker.getJobVertexStats(JOB_ID, EXECUTION_JOB_VERTEX));
+ tracker.getJobVertexStats(JOB_ID, executionJobVertex));
for (int subtaskIndex = 0; subtaskIndex < PARALLELISM; subtaskIndex++) {
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsDefaultSample, ATTEMPT_IDS[subtaskIndex]),
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ threadInfoStatsDefaultSample, attemptIDS[subtaskIndex]),
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
}
}
@@ -435,14 +452,14 @@ class VertexThreadInfoTrackerTest {
VertexThreadInfoTracker tracker, int subtaskIndex)
throws InterruptedException, ExecutionException {
// no stats yet, but the request triggers async collection of stats
- assertThat(tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex))
+ assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex))
.isNotPresent();
// block until the async call completes and the first result is available
tracker.getResultAvailableFuture().get();
assertExpectedEqualsReceived(
generateThreadInfoStatsForExecutionVertex(
- threadInfoStatsDefaultSample, ATTEMPT_IDS[subtaskIndex]),
- tracker.getExecutionVertexStats(JOB_ID, EXECUTION_JOB_VERTEX, subtaskIndex));
+ threadInfoStatsDefaultSample, attemptIDS[subtaskIndex]),
+ tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, subtaskIndex));
}
private static VertexThreadInfoStats generateThreadInfoStatsForExecutionVertex(
@@ -535,17 +552,17 @@ class VertexThreadInfoTrackerTest {
.build();
}
- private static VertexThreadInfoStats createThreadInfoStats(int requestId, Duration timeGap) {
+ private VertexThreadInfoStats createThreadInfoStats(int requestId, Duration timeGap) {
return createThreadInfoStats(Instant.now(), requestId, timeGap);
}
- private static VertexThreadInfoStats createThreadInfoStats(
+ private VertexThreadInfoStats createThreadInfoStats(
Instant startTime, int requestId, Duration timeGap) {
Instant endTime = startTime.plus(timeGap);
final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samples = new HashMap<>();
- for (ExecutionVertex vertex : EXECUTION_JOB_VERTEX.getTaskVertices()) {
+ for (ExecutionVertex vertex : executionJobVertex.getTaskVertices()) {
Optional<ThreadInfoSample> threadInfoSample =
JvmUtils.createThreadInfoSample(
Thread.currentThread().getId(), MAX_STACK_TRACE_DEPTH);
@@ -559,7 +576,7 @@ class VertexThreadInfoTrackerTest {
requestId, startTime.toEpochMilli(), endTime.toEpochMilli(), samples);
}
- private static ExecutionJobVertex createExecutionJobVertex() {
+ private ExecutionJobVertex createExecutionJobVertex() {
try {
JobVertex jobVertex = new JobVertex("testVertex");
jobVertex.setParallelism(PARALLELISM);
@@ -572,10 +589,19 @@ class VertexThreadInfoTrackerTest {
new DirectScheduledExecutorService());
final ExecutionGraph eg = scheduler.getExecutionGraph();
scheduler.startScheduling();
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
+ switch (executionState) {
+ case RUNNING:
+ ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
+ break;
+ case INITIALIZING:
+ ExecutionGraphTestUtils.switchAllVerticesToInitializing(eg);
+ break;
+ default:
+ throw new IllegalArgumentException("Just support RUNNING and INITIALIZING.");
+ }
return scheduler.getExecutionJobVertex(jobVertex.getID());
} catch (Exception e) {
- throw new RuntimeException("Failed to create ExecutionJobVertex.");
+ throw new RuntimeException("Failed to create ExecutionJobVertex.", e);
}
}