This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new d6e8e4287ea [FLINK-22243] Reactive Mode parallelism changes are not
shown in the job graph visualization in the UI
d6e8e4287ea is described below
commit d6e8e4287eaaf144a3a4798b89825370170dce97
Author: Dawid Wysakowicz <[email protected]>
AuthorDate: Fri Sep 30 11:14:19 2022 +0200
[FLINK-22243] Reactive Mode parallelism changes are not shown in the job
graph visualization in the UI
Update the JsonPlan of ExecutionGraph with the correct parallelism instead
of using the original one from the JobGraph.
---
.../jobgraph/jsonplan/JsonPlanGenerator.java | 50 ++++++++++++++++---
.../scheduler/adaptive/CreatingExecutionGraph.java | 19 ++++++++
.../adaptive/StateTrackingMockExecutionGraph.java | 8 ++--
.../flink/test/scheduling/ReactiveModeITCase.java | 56 ++++++++++++++++++++++
4 files changed, 122 insertions(+), 11 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
index 63e95ea81f6..30414506435 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
@@ -18,9 +18,14 @@
package org.apache.flink.runtime.jobgraph.jsonplan;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.jobgraph.JobEdge;
import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobgraph.JobVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonFactory;
import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
@@ -28,14 +33,43 @@ import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator
import org.apache.commons.lang3.StringEscapeUtils;
import java.io.StringWriter;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
+@Internal
public class JsonPlanGenerator {
private static final String NOT_SET = "";
private static final String EMPTY = "{}";
+ private static final VertexParallelism EMPTY_VERTEX_PARALLELISM =
+ new VertexParallelism() {
+ @Override
+ public Map<JobVertexID, Integer>
getMaxParallelismForVertices() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public int getParallelism(JobVertexID jobVertexId) {
+ return -1;
+ }
+ };
public static String generatePlan(JobGraph jg) {
+ return generatePlan(
+ jg.getJobID(),
+ jg.getName(),
+ jg.getJobType(),
+ jg.getVertices(),
+ EMPTY_VERTEX_PARALLELISM);
+ }
+
+ public static String generatePlan(
+ JobID jobID,
+ String jobName,
+ JobType jobType,
+ Iterable<JobVertex> vertices,
+ VertexParallelism vertexParallelism) {
try {
final StringWriter writer = new StringWriter(1024);
@@ -44,13 +78,13 @@ public class JsonPlanGenerator {
// start of everything
gen.writeStartObject();
- gen.writeStringField("jid", jg.getJobID().toString());
- gen.writeStringField("name", jg.getName());
- gen.writeStringField("type", jg.getJobType().name());
+ gen.writeStringField("jid", jobID.toString());
+ gen.writeStringField("name", jobName);
+ gen.writeStringField("type", jobType.name());
gen.writeArrayFieldStart("nodes");
// info per vertex
- for (JobVertex vertex : jg.getVertices()) {
+ for (JobVertex vertex : vertices) {
String operator =
vertex.getOperatorName() != null ?
vertex.getOperatorName() : NOT_SET;
@@ -81,8 +115,12 @@ public class JsonPlanGenerator {
gen.writeStartObject();
// write the core properties
- gen.writeStringField("id", vertex.getID().toString());
- gen.writeNumberField("parallelism", vertex.getParallelism());
+ JobVertexID vertexID = vertex.getID();
+ int storeParallelism =
vertexParallelism.getParallelism(vertexID);
+ gen.writeStringField("id", vertexID.toString());
+ gen.writeNumberField(
+ "parallelism",
+ storeParallelism != -1 ? storeParallelism :
vertex.getParallelism());
gen.writeStringField("operator", operator);
gen.writeStringField("operator_strategy", operatorDescr);
gen.writeStringField("description", description);
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
index a4db25a552c..08ce28e3b51 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
@@ -24,6 +24,8 @@ import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.jobgraph.JobType;
+import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
import org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.GlobalFailureHandler;
@@ -41,6 +43,7 @@ import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
+import java.util.stream.StreamSupport;
/**
* State which waits for the creation of the {@link ExecutionGraph}. If the
creation fails, then the
@@ -115,6 +118,22 @@ public class CreatingExecutionGraph implements State {
operatorCoordinatorHandler.initializeOperatorCoordinators(
context.getMainThreadExecutor());
operatorCoordinatorHandler.startAllOperatorCoordinators();
+ String updatedPlan =
+ JsonPlanGenerator.generatePlan(
+ executionGraph.getJobID(),
+ executionGraph.getJobName(),
+ JobType.STREAMING, // Adaptive scheduler works
only with STREAMING
+ // jobs
+ () ->
+ StreamSupport.stream(
+ executionGraph
+
.getAllExecutionVertices()
+ .spliterator(),
+ false)
+ .map(v ->
v.getJobVertex().getJobVertex())
+ .iterator(),
+
executionGraphWithVertexParallelism.getVertexParallelism());
+ executionGraph.setJsonPlan(updatedPlan);
context.goToExecuting(
result.getExecutionGraph(),
executionGraphHandler,
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
index 222bf986a8a..0ab8623a65f 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
@@ -172,6 +172,9 @@ class StateTrackingMockExecutionGraph implements
ExecutionGraph {
return "";
}
+ @Override
+ public void setJsonPlan(String jsonPlan) {}
+
@Override
public JobID getJobID() {
return jobId;
@@ -274,11 +277,6 @@ class StateTrackingMockExecutionGraph implements
ExecutionGraph {
throw new UnsupportedOperationException();
}
- @Override
- public void setJsonPlan(String jsonPlan) {
- throw new UnsupportedOperationException();
- }
-
@Override
public Configuration getJobConfiguration() {
throw new UnsupportedOperationException();
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
index ee09c293b8a..346e3013ac9 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/ReactiveModeITCase.java
@@ -25,6 +25,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.configuration.SchedulerExecutionMode;
import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.rest.messages.job.JobDetailsInfo;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
@@ -36,11 +37,15 @@ import
org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.test.util.MiniClusterWithClientResource;
import org.apache.flink.util.TestLogger;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+
import org.junit.Rule;
import org.junit.Test;
import java.util.concurrent.ExecutionException;
+import static org.assertj.core.api.Assertions.assertThat;
+
/** Tests for Reactive Mode (FLIP-159). */
public class ReactiveModeITCase extends TestLogger {
private static final int NUMBER_SLOTS_PER_TASK_MANAGER = 2;
@@ -48,6 +53,8 @@ public class ReactiveModeITCase extends TestLogger {
private static final Configuration configuration =
getReactiveModeConfiguration();
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
@Rule
public final MiniClusterWithClientResource miniClusterResource =
new MiniClusterWithClientResource(
@@ -107,6 +114,55 @@ public class ReactiveModeITCase extends TestLogger {
NUMBER_SLOTS_PER_TASK_MANAGER * (INITIAL_NUMBER_TASK_MANAGERS
+ 1));
}
+ @Test
+ public void testJsonPlanParallelismAfterRescale() throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ final DataStream<String> input = env.addSource(new DummySource());
+ input.addSink(new DiscardingSink<>());
+
+ final JobClient jobClient = env.executeAsync();
+
+ int initialParallelism = NUMBER_SLOTS_PER_TASK_MANAGER *
INITIAL_NUMBER_TASK_MANAGERS;
+ waitUntilParallelismForVertexReached(
+ miniClusterResource.getRestClusterClient(),
+ jobClient.getJobID(),
+ initialParallelism);
+
+ ArchivedExecutionGraph archivedExecutionGraph =
+ miniClusterResource
+ .getMiniCluster()
+ .getArchivedExecutionGraph(jobClient.getJobID())
+ .get();
+
+ assertThat(
+ OBJECT_MAPPER
+ .readTree(archivedExecutionGraph.getJsonPlan())
+ .findValues("parallelism"))
+ .allMatch(n -> n.asInt() == initialParallelism);
+
+ // scale up to 2 TaskManagers:
+ miniClusterResource.getMiniCluster().startTaskManager();
+
+ int rescaledParallelism =
+ NUMBER_SLOTS_PER_TASK_MANAGER * (INITIAL_NUMBER_TASK_MANAGERS
+ 1);
+ waitUntilParallelismForVertexReached(
+ miniClusterResource.getRestClusterClient(),
+ jobClient.getJobID(),
+ rescaledParallelism);
+
+ archivedExecutionGraph =
+ miniClusterResource
+ .getMiniCluster()
+ .getArchivedExecutionGraph(jobClient.getJobID())
+ .get();
+
+ assertThat(
+ OBJECT_MAPPER
+ .readTree(archivedExecutionGraph.getJsonPlan())
+ .findValues("parallelism"))
+ .allMatch(n -> n.asInt() == rescaledParallelism);
+ }
+
@Test
public void testScaleDownOnTaskManagerLoss() throws Exception {
// test preparation: ensure we have 2 TaskManagers running
