This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.16
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.16 by this push:
new a9592c075d3 [FLINK-22243] Reactive Mode parallelism changes are not
shown in the job graph visualization in the UI
a9592c075d3 is described below
commit a9592c075d3130e1d6af3748f201de3e973c3d0e
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 | 57 ++++++++++++++++++++++
4 files changed, 123 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 9f3c9ba321a..54a65ccddcd 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;
@@ -285,11 +288,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 5939dbd9b94..d8dd6a86bab 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;
@@ -35,6 +36,9 @@ import
org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
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.util.jackson.JacksonMapperFactory;
+
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Rule;
import org.junit.Test;
@@ -42,6 +46,8 @@ import org.junit.rules.TemporaryFolder;
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;
@@ -49,6 +55,8 @@ public class ReactiveModeITCase extends TestLogger {
private static final Configuration configuration =
getReactiveModeConfiguration();
+ private static final ObjectMapper OBJECT_MAPPER =
JacksonMapperFactory.createObjectMapper();
+
@Rule public TemporaryFolder tempFolder = new TemporaryFolder();
@Rule
@@ -110,6 +118,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
