You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2025/04/09 23:37:45 UTC
(flink) branch master updated: [FLINK-37511] Use Jackson serialization in JobPlanInfo.Plan
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new fc51d4e09b4 [FLINK-37511] Use Jackson serialization in JobPlanInfo.Plan
fc51d4e09b4 is described below
commit fc51d4e09b45064e180d48fb3cfcc68865d950d0
Author: Efrat Levitan <ef...@gmail.com>
AuthorDate: Thu Mar 13 08:10:31 2025 +0200
[FLINK-37511] Use Jackson serialization in JobPlanInfo.Plan
fixes
generate-rest-snapshot
generate-rest-snapshot
generate-rest-snapshot
---
.../shortcodes/generated/rest_v1_dispatcher.html | 203 ++++++++++++-
docs/static/generated/rest_v1_dispatcher.yml | 55 +++-
.../client/program/rest/RestClusterClientTest.java | 2 +-
.../webmonitor/handlers/JarSubmissionITCase.java | 2 +-
.../src/test/resources/rest_api_v1.snapshot | 270 ++++++++++++++++-
.../executiongraph/AccessExecutionGraph.java | 7 +-
.../executiongraph/ArchivedExecutionGraph.java | 17 +-
.../executiongraph/DefaultExecutionGraph.java | 11 +-
.../DefaultExecutionGraphBuilder.java | 7 +-
.../runtime/executiongraph/ExecutionGraph.java | 3 +-
.../jobgraph/jsonplan/JsonPlanGenerator.java | 101 +++----
.../rest/handler/job/JobDetailsHandler.java | 2 +-
.../runtime/rest/handler/job/JobPlanHandler.java | 2 +-
.../flink/runtime/rest/messages/JobPlanInfo.java | 336 ++++++++++++++++++++-
.../runtime/rest/messages/job/JobDetailsInfo.java | 14 +-
.../scheduler/adaptive/CreatingExecutionGraph.java | 5 +-
.../adaptivebatch/AdaptiveBatchScheduler.java | 9 +-
.../ExecutionGraphInfoStoreTestUtils.java | 3 +-
.../executiongraph/ArchivedExecutionGraphTest.java | 2 +-
.../jobgraph/jsonplan/JsonGeneratorTest.java | 31 +-
.../utils/ArchivedExecutionGraphBuilder.java | 17 +-
.../rest/messages/job/JobDetailsInfoTest.java | 4 +-
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 8 +-
.../adaptive/StateTrackingMockExecutionGraph.java | 8 +-
.../flink/test/scheduling/ReactiveModeITCase.java | 17 +-
25 files changed, 954 insertions(+), 182 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
index 41b3c4b502c..a0788dd04cc 100644
--- a/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
+++ b/docs/layouts/shortcodes/generated/rest_v1_dispatcher.html
@@ -509,7 +509,72 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"properties" : {
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "description" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "caching" : {
+ "type" : "string"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ }
+ }
+ }
+ },
+ "type" : {
+ "type" : "string"
+ }
+ }
}
}
}</code></pre>
@@ -1126,7 +1191,72 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "description" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "caching" : {
+ "type" : "string"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ }
+ }
+ }
+ },
+ "type" : {
+ "type" : "string"
+ }
+ }
},
"start-time" : {
"type" : "integer"
@@ -1143,7 +1273,7 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
},
"stream-graph" : {
"type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
},
"timestamps" : {
"type" : "object",
@@ -2879,7 +3009,72 @@ Using 'curl' you can upload a jar via 'curl -X POST -H "Expect:" -F "jarfile=@pa
"properties" : {
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "description" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "caching" : {
+ "type" : "string"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "id" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ }
+ }
+ }
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ }
+ }
+ }
+ },
+ "type" : {
+ "type" : "string"
+ }
+ }
}
}
}</code></pre>
diff --git a/docs/static/generated/rest_v1_dispatcher.yml b/docs/static/generated/rest_v1_dispatcher.yml
index 03bc5b539a9..181fe786ab0 100644
--- a/docs/static/generated/rest_v1_dispatcher.yml
+++ b/docs/static/generated/rest_v1_dispatcher.yml
@@ -6,7 +6,7 @@ info:
license:
name: Apache 2.0
url: https://www.apache.org/licenses/LICENSE-2.0.html
- version: v1/2.0-SNAPSHOT
+ version: v1/2.1-SNAPSHOT
paths:
/cluster:
delete:
@@ -2149,6 +2149,22 @@ components:
enum:
- IN_PROGRESS
- COMPLETED
+ Input:
+ type: object
+ properties:
+ caching:
+ type: string
+ exchange:
+ type: string
+ id:
+ type: string
+ local_strategy:
+ type: string
+ num:
+ type: integer
+ format: int64
+ ship_strategy:
+ type: string
IntermediateDataSetID:
pattern: "[0-9a-f]{32}"
type: string
@@ -2337,7 +2353,7 @@ components:
type: integer
format: int32
plan:
- $ref: '#/components/schemas/RawJson'
+ $ref: '#/components/schemas/Plan'
start-time:
type: integer
format: int64
@@ -2436,7 +2452,7 @@ components:
type: object
properties:
plan:
- $ref: '#/components/schemas/RawJson'
+ $ref: '#/components/schemas/Plan'
JobResourceRequirementsBody:
type: object
additionalProperties:
@@ -2663,6 +2679,39 @@ components:
type: object
allOf:
- $ref: '#/components/schemas/SubtaskCheckpointStatistics'
+ Plan:
+ type: object
+ properties:
+ jid:
+ type: string
+ name:
+ type: string
+ nodes:
+ type: array
+ items:
+ $ref: '#/components/schemas/PlanNode'
+ type:
+ type: string
+ PlanNode:
+ type: object
+ properties:
+ description:
+ type: string
+ id:
+ type: string
+ inputs:
+ type: array
+ items:
+ $ref: '#/components/schemas/Input'
+ operator:
+ type: string
+ operator_strategy:
+ type: string
+ optimizer_properties:
+ type: string
+ parallelism:
+ type: integer
+ format: int64
ProcessingMode:
type: string
enum:
diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
index 19e5b2684c8..27e1835b263 100644
--- a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
+++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientTest.java
@@ -1277,7 +1277,7 @@ class RestClusterClientTest {
Collections.singletonMap(JobStatus.RUNNING, 1L),
jobVertexDetailsInfos,
Collections.singletonMap(ExecutionState.RUNNING, 1),
- new JobPlanInfo.RawJson("{\"id\":\"1234\"}"),
+ new JobPlanInfo.Plan("1243", "", "", new ArrayList<>()),
new JobPlanInfo.RawJson("{\"id\":\"1234\"}"),
0);
final TestJobDetailsInfoHandler jobDetailsInfoHandler =
diff --git a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarSubmissionITCase.java b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarSubmissionITCase.java
index af9fc56606b..1961de6acf0 100644
--- a/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarSubmissionITCase.java
+++ b/flink-runtime-web/src/test/java/org/apache/flink/runtime/webmonitor/handlers/JarSubmissionITCase.java
@@ -94,7 +94,7 @@ class JarSubmissionITCase {
final JobPlanInfo planResponse = showPlan(planHandler, storedJarName, restfulGateway);
// we're only interested in the core functionality so checking for a small detail is
// sufficient
- assertThat(planResponse.getJsonPlan()).contains("\"name\":\"Flink Streaming Job\"");
+ assertThat(planResponse.getPlan().getName()).isEqualTo("Flink Streaming Job");
runJar(runHandler, storedJarName, restfulGateway);
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 0e7c321c690..2bab3e750ea 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -333,7 +333,72 @@
"properties" : {
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "type" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "description" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "caching" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
}
}
@@ -392,7 +457,72 @@
"properties" : {
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "type" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "description" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "caching" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
}
}
@@ -969,11 +1099,76 @@
},
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "type" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "description" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "caching" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
},
"stream-graph" : {
"type" : "object",
- "$ref" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
},
"pending-operators" : {
"type" : "integer"
@@ -2219,7 +2414,72 @@
"properties" : {
"plan" : {
"type" : "object",
- "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:RawJson"
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan",
+ "properties" : {
+ "jid" : {
+ "type" : "string"
+ },
+ "name" : {
+ "type" : "string"
+ },
+ "type" : {
+ "type" : "string"
+ },
+ "nodes" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "operator" : {
+ "type" : "string"
+ },
+ "parallelism" : {
+ "type" : "integer"
+ },
+ "operator_strategy" : {
+ "type" : "string"
+ },
+ "description" : {
+ "type" : "string"
+ },
+ "optimizer_properties" : {
+ "type" : "string"
+ },
+ "inputs" : {
+ "type" : "array",
+ "items" : {
+ "type" : "object",
+ "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:JobPlanInfo:Plan:Node:Input",
+ "properties" : {
+ "id" : {
+ "type" : "string"
+ },
+ "num" : {
+ "type" : "integer"
+ },
+ "exchange" : {
+ "type" : "string"
+ },
+ "ship_strategy" : {
+ "type" : "string"
+ },
+ "local_strategy" : {
+ "type" : "string"
+ },
+ "caching" : {
+ "type" : "string"
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/AccessExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/AccessExecutionGraph.java
index 1cac672a896..534b3ab5de0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/AccessExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/AccessExecutionGraph.java
@@ -26,6 +26,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointStatsSnapshot;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.TernaryBoolean;
@@ -41,11 +42,11 @@ import java.util.Optional;
*/
public interface AccessExecutionGraph extends JobStatusProvider {
/**
- * Returns the job plan as a JSON string.
+ * Returns the job plan as a JobPlanInfo.Plan.
*
- * @return job plan as a JSON string
+ * @return job plan as a JobPlanInfo.Plan
*/
- String getJsonPlan();
+ JobPlanInfo.Plan getPlan();
/**
* Returns the stream graph as a JSON string.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java
index f93eaa4a6ae..71935d9eedb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraph.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.VertexParallelismInformation;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
import org.apache.flink.util.OptionalFailure;
@@ -95,7 +96,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
@Nullable private final ErrorInfo failureCause;
// ------ Fields that are only relevant for archived execution graphs ------------
- private final String jsonPlan;
+ private final JobPlanInfo.Plan plan;
private final StringifiedAccumulatorResult[] archivedUserAccumulators;
private final ArchivedExecutionConfig archivedExecutionConfig;
private final boolean isStoppable;
@@ -126,7 +127,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
JobStatus state,
@Nullable JobType jobType,
@Nullable ErrorInfo failureCause,
- String jsonPlan,
+ JobPlanInfo.Plan plan,
StringifiedAccumulatorResult[] archivedUserAccumulators,
Map<String, SerializedValue<OptionalFailure<Object>>> serializedUserAccumulators,
ArchivedExecutionConfig executionConfig,
@@ -148,7 +149,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
this.state = Preconditions.checkNotNull(state);
this.jobType = jobType;
this.failureCause = failureCause;
- this.jsonPlan = Preconditions.checkNotNull(jsonPlan);
+ this.plan = Preconditions.checkNotNull(plan);
this.archivedUserAccumulators = Preconditions.checkNotNull(archivedUserAccumulators);
this.serializedUserAccumulators = Preconditions.checkNotNull(serializedUserAccumulators);
this.archivedExecutionConfig = Preconditions.checkNotNull(executionConfig);
@@ -166,8 +167,8 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
// --------------------------------------------------------------------------------------------
@Override
- public String getJsonPlan() {
- return jsonPlan;
+ public JobPlanInfo.Plan getPlan() {
+ return plan;
}
@Override
@@ -374,7 +375,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
statusOverride == null ? executionGraph.getState() : statusOverride,
executionGraph.getJobType(),
executionGraph.getFailureInfo(),
- executionGraph.getJsonPlan(),
+ executionGraph.getPlan(),
executionGraph.getAccumulatorResultsStringified(),
serializedUserAccumulators,
executionGraph.getArchivedExecutionConfig(),
@@ -473,7 +474,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
final long[] timestamps = new long[JobStatus.values().length];
timestamps[JobStatus.INITIALIZING.ordinal()] = initializationTimestamp;
- String jsonPlan = "{}";
+ JobPlanInfo.Plan plan = new JobPlanInfo.Plan("", "", "", new ArrayList<>());
ErrorInfo failureInfo = null;
if (throwable != null) {
@@ -493,7 +494,7 @@ public class ArchivedExecutionGraph implements AccessExecutionGraph, Serializabl
jobStatus,
jobType,
failureInfo,
- jsonPlan,
+ plan,
archivedUserAccumulators,
serializedUserAccumulators,
new ExecutionConfig().archive(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
index 2d551ec6dfd..345b8977d19 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -67,6 +67,7 @@ import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.operators.coordination.CoordinatorStore;
import org.apache.flink.runtime.operators.coordination.CoordinatorStoreImpl;
import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.DefaultVertexParallelismStore;
import org.apache.flink.runtime.scheduler.InternalFailuresListener;
import org.apache.flink.runtime.scheduler.SsgNetworkMemoryCalculationUtils;
@@ -280,7 +281,7 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
@Nullable private TernaryBoolean stateChangelogEnabled;
- private String jsonPlan;
+ private JobPlanInfo.Plan plan;
/** Shuffle master to register partitions for task deployment. */
private final ShuffleMaster<?> shuffleMaster;
@@ -613,8 +614,8 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
// --------------------------------------------------------------------------------------------
@Override
- public void setJsonPlan(String jsonPlan) {
- this.jsonPlan = jsonPlan;
+ public void setPlan(JobPlanInfo.Plan plan) {
+ this.plan = plan;
}
@Override
@@ -623,8 +624,8 @@ public class DefaultExecutionGraph implements ExecutionGraph, InternalExecutionG
}
@Override
- public String getJsonPlan() {
- return jsonPlan;
+ public JobPlanInfo.Plan getPlan() {
+ return plan;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
index 6b8db748def..5f3ede18ac3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphBuilder.java
@@ -44,6 +44,7 @@ import org.apache.flink.runtime.jobgraph.jsonplan.JsonPlanGenerator;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
import org.apache.flink.runtime.scheduler.adaptivebatch.ExecutionPlanSchedulingContext;
import org.apache.flink.runtime.shuffle.ShuffleMaster;
@@ -175,11 +176,11 @@ public class DefaultExecutionGraphBuilder {
// set the basic properties
try {
- executionGraph.setJsonPlan(JsonPlanGenerator.generatePlan(jobGraph));
+ executionGraph.setPlan(JsonPlanGenerator.generatePlan(jobGraph));
} catch (Throwable t) {
- log.warn("Cannot create JSON plan for job", t);
+ log.warn("Cannot create plan for job", t);
// give the graph an empty plan
- executionGraph.setJsonPlan("{}");
+ executionGraph.setPlan(new JobPlanInfo.Plan("", "", "", new ArrayList<>()));
}
initJobVerticesOnMaster(
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 9f48b908d37..cf7e23718a0 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
@@ -39,6 +39,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.InternalFailuresListener;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
import org.apache.flink.runtime.scheduler.strategy.SchedulingTopology;
@@ -99,7 +100,7 @@ public interface ExecutionGraph extends AccessExecutionGraph {
KvStateLocationRegistry getKvStateLocationRegistry();
- void setJsonPlan(String jsonPlan);
+ void setPlan(JobPlanInfo.Plan plan);
Configuration getJobConfiguration();
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 6005c324642..ebd0895bf9d 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
@@ -25,6 +25,7 @@ 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.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
import org.apache.flink.streaming.api.graph.StreamEdge;
import org.apache.flink.streaming.api.graph.StreamGraph;
@@ -36,6 +37,8 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator
import org.apache.commons.text.StringEscapeUtils;
import java.io.StringWriter;
+import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
@@ -45,7 +48,7 @@ public class JsonPlanGenerator {
private static final String NOT_SET = "";
private static final String EMPTY = "{}";
- public static String generatePlan(JobGraph jg) {
+ public static JobPlanInfo.Plan generatePlan(JobGraph jg) {
return generatePlan(
jg.getJobID(),
jg.getName(),
@@ -54,24 +57,14 @@ public class JsonPlanGenerator {
VertexParallelism.empty());
}
- public static String generatePlan(
+ public static JobPlanInfo.Plan generatePlan(
JobID jobID,
String jobName,
JobType jobType,
Iterable<JobVertex> vertices,
VertexParallelism vertexParallelism) {
try {
- final StringWriter writer = new StringWriter(1024);
-
- final JsonFactory factory = new JsonFactory();
- final JsonGenerator gen = factory.createGenerator(writer);
-
- // start of everything
- gen.writeStartObject();
- gen.writeStringField("jid", jobID.toString());
- gen.writeStringField("name", jobName);
- gen.writeStringField("type", jobType.name());
- gen.writeArrayFieldStart("nodes");
+ Collection<JobPlanInfo.Plan.Node> nodes = new ArrayList<>();
// info per vertex
for (JobVertex vertex : vertices) {
@@ -102,74 +95,56 @@ public class JsonPlanGenerator {
operatorDescr = StringEscapeUtils.escapeHtml4(operatorDescr);
operatorDescr = operatorDescr.replace("\n", "<br/>");
- gen.writeStartObject();
-
- // write the core properties
JobVertexID vertexID = vertex.getID();
- gen.writeStringField("id", vertexID.toString());
- gen.writeNumberField(
- "parallelism",
+ long parallelism =
vertexParallelism
.getParallelismOptional(vertexID)
- .orElse(vertex.getParallelism()));
- gen.writeStringField("operator", operator);
- gen.writeStringField("operator_strategy", operatorDescr);
- gen.writeStringField("description", description);
+ .orElse(vertex.getParallelism());
+ Collection<JobPlanInfo.Plan.Node.Input> inputs = new ArrayList<>();
if (!vertex.isInputVertex()) {
- // write the input edge properties
- gen.writeArrayFieldStart("inputs");
-
- List<JobEdge> inputs = vertex.getInputs();
- for (int inputNum = 0; inputNum < inputs.size(); inputNum++) {
- JobEdge edge = inputs.get(inputNum);
+ for (int inputNum = 0; inputNum < vertex.getInputs().size(); inputNum++) {
+ JobEdge edge = vertex.getInputs().get(inputNum);
if (edge.getSource() == null) {
continue;
}
-
JobVertex predecessor = edge.getSource().getProducer();
+ if (predecessor == null || predecessor.getID() == null) {
+ continue;
+ }
+ String inputId = predecessor.getID().toString();
+ if (edge.getSource().getResultType() == null
+ || edge.getSource().getResultType().name() == null) {
+ continue;
+ }
+ String exchange = edge.getSource().getResultType().name().toLowerCase();
String shipStrategy = edge.getShipStrategyName();
String preProcessingOperation = edge.getPreProcessingOperationName();
String operatorLevelCaching = edge.getOperatorLevelCachingDescription();
- gen.writeStartObject();
- gen.writeNumberField("num", inputNum);
- gen.writeStringField("id", predecessor.getID().toString());
-
- if (shipStrategy != null) {
- gen.writeStringField("ship_strategy", shipStrategy);
- }
- if (preProcessingOperation != null) {
- gen.writeStringField("local_strategy", preProcessingOperation);
- }
- if (operatorLevelCaching != null) {
- gen.writeStringField("caching", operatorLevelCaching);
- }
-
- gen.writeStringField(
- "exchange", edge.getSource().getResultType().name().toLowerCase());
-
- gen.writeEndObject();
+ inputs.add(
+ new JobPlanInfo.Plan.Node.Input(
+ inputId,
+ inputNum,
+ exchange,
+ shipStrategy,
+ preProcessingOperation,
+ operatorLevelCaching));
}
-
- gen.writeEndArray();
}
-
- // write the optimizer properties
- gen.writeFieldName("optimizer_properties");
- gen.writeRawValue(optimizerProps);
-
- gen.writeEndObject();
+ nodes.add(
+ new JobPlanInfo.Plan.Node(
+ vertexID.toString(),
+ operator,
+ parallelism,
+ operatorDescr,
+ description,
+ optimizerProps,
+ inputs));
}
- // end of everything
- gen.writeEndArray();
- gen.writeEndObject();
-
- gen.close();
-
- return writer.toString();
+ return new JobPlanInfo.Plan(jobID.toString(), jobName, jobType.name(), nodes);
} catch (Exception e) {
throw new RuntimeException("Failed to generate plan", e);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java
index 0ee703cfa51..dd657cd5af5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java
@@ -160,7 +160,7 @@ public class JobDetailsHandler
timestamps,
jobVertexInfos,
jobVerticesPerStateMap,
- new JobPlanInfo.RawJson(executionGraph.getJsonPlan()),
+ executionGraph.getPlan(),
streamGraphJson,
executionGraph.getPendingOperatorCount());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobPlanHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobPlanHandler.java
index f8f4c5fd5f7..f6a03e1a6ae 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobPlanHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobPlanHandler.java
@@ -75,6 +75,6 @@ public class JobPlanHandler
}
private static JobPlanInfo createJobPlanInfo(AccessExecutionGraph executionGraph) {
- return new JobPlanInfo(executionGraph.getJsonPlan());
+ return new JobPlanInfo(executionGraph.getPlan());
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobPlanInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobPlanInfo.java
index d82c9ec4657..e8082acfe3c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobPlanInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobPlanInfo.java
@@ -37,7 +37,13 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.jsonFormatVisitors.JsonFormatVisitorWrapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ser.std.StdSerializer;
+import io.swagger.v3.oas.annotations.media.Schema;
+
+import javax.annotation.Nullable;
+
import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collection;
import java.util.Objects;
/** Response type of the {@link JobPlanHandler}. */
@@ -46,20 +52,16 @@ public class JobPlanInfo implements ResponseBody {
private static final String FIELD_NAME_PLAN = "plan";
@JsonProperty(FIELD_NAME_PLAN)
- private final RawJson jsonPlan;
-
- public JobPlanInfo(String jsonPlan) {
- this(new RawJson(Preconditions.checkNotNull(jsonPlan)));
- }
+ private final Plan plan;
@JsonCreator
- public JobPlanInfo(@JsonProperty(FIELD_NAME_PLAN) RawJson jsonPlan) {
- this.jsonPlan = jsonPlan;
+ public JobPlanInfo(@JsonProperty(FIELD_NAME_PLAN) Plan plan) {
+ this.plan = plan;
}
@JsonIgnore
- public String getJsonPlan() {
- return jsonPlan.json;
+ public Plan getPlan() {
+ return plan;
}
@Override
@@ -71,17 +73,17 @@ public class JobPlanInfo implements ResponseBody {
return false;
}
JobPlanInfo that = (JobPlanInfo) o;
- return Objects.equals(jsonPlan, that.jsonPlan);
+ return Objects.equals(plan, that.plan);
}
@Override
public int hashCode() {
- return Objects.hash(jsonPlan);
+ return Objects.hash(plan);
}
@Override
public String toString() {
- return "JobPlanInfo{" + "jsonPlan=" + jsonPlan + '}';
+ return "JobPlanInfo{" + "plan=" + plan + '}';
}
/** Simple wrapper around a raw JSON string. */
@@ -164,4 +166,314 @@ public class JobPlanInfo implements ResponseBody {
}
}
}
+
+ // ---------------------------------------------------
+ // Static inner classes
+ // ---------------------------------------------------
+
+ /** An inner class for the plan reference. */
+ @Schema(name = "Plan")
+ public static final class Plan implements Serializable {
+ private static final String FIELD_NAME_JOB_ID = "jid";
+
+ @JsonProperty(FIELD_NAME_JOB_ID)
+ private final String jobId;
+
+ private static final String FIELD_NAME_NAME = "name";
+
+ @JsonProperty(FIELD_NAME_NAME)
+ private final String name;
+
+ private static final String FIELD_NAME_TYPE = "type";
+
+ @JsonProperty(FIELD_NAME_TYPE)
+ private final String type;
+
+ private static final String FIELD_NAME_NODES = "nodes";
+
+ @JsonProperty(FIELD_NAME_NODES)
+ private final Collection<Node> nodes;
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Plan that = (Plan) o;
+ return Objects.equals(jobId, that.jobId)
+ && Objects.equals(name, that.name)
+ && Objects.equals(type, that.type)
+ && Objects.equals(nodes, that.nodes);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(jobId, name, type, nodes);
+ }
+
+ @JsonCreator
+ public Plan(
+ @JsonProperty(FIELD_NAME_JOB_ID) String jobId,
+ @JsonProperty(FIELD_NAME_NAME) String name,
+ @JsonProperty(FIELD_NAME_TYPE) String type,
+ @JsonProperty(FIELD_NAME_NODES) Collection<Node> nodes) {
+ this.jobId = Preconditions.checkNotNull(jobId);
+ this.name = Preconditions.checkNotNull(name);
+ this.type = Preconditions.checkNotNull(type);
+ this.nodes = Preconditions.checkNotNull(nodes);
+ }
+
+ @JsonIgnore
+ public String getJobId() {
+ return jobId;
+ }
+
+ @JsonIgnore
+ public String getName() {
+ return name;
+ }
+
+ @JsonIgnore
+ public String getType() {
+ return type;
+ }
+
+ @JsonIgnore
+ public Collection<Node> getNodes() {
+ return nodes;
+ }
+
+ /** An inner class containing node (vertex) level info. */
+ @Schema(name = "PlanNode")
+ public static final class Node implements Serializable {
+ private static final String FIELD_NAME_ID = "id";
+
+ @JsonProperty(FIELD_NAME_ID)
+ private final String id;
+
+ private static final String FIELD_NAME_PARALLELISM = "parallelism";
+
+ @JsonProperty(FIELD_NAME_PARALLELISM)
+ private final long parallelism;
+
+ private static final String FIELD_NAME_OPERATOR = "operator";
+
+ @JsonProperty(FIELD_NAME_OPERATOR)
+ private final String operator;
+
+ private static final String FIELD_NAME_OPERATOR_STRATEGY = "operator_strategy";
+
+ @JsonProperty(FIELD_NAME_OPERATOR_STRATEGY)
+ private final String operatorStrategy;
+
+ private static final String FIELD_NAME_DESCRIPTION = "description";
+
+ @JsonProperty(FIELD_NAME_DESCRIPTION)
+ private final String description;
+
+ private static final String FIELD_NAME_OPTIMIZER_PROPERTIES = "optimizer_properties";
+
+ @JsonProperty(FIELD_NAME_OPTIMIZER_PROPERTIES)
+ private final String optimizerProperties;
+
+ private static final String FIELD_NAME_INPUTS = "inputs";
+
+ @JsonProperty(FIELD_NAME_INPUTS)
+ private final Collection<Input> inputs;
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Node that = (Node) o;
+ return Objects.equals(id, that.id)
+ && Objects.equals(operator, that.operator)
+ && Objects.equals(operatorStrategy, that.operatorStrategy)
+ && Objects.equals(description, that.description)
+ && Objects.equals(optimizerProperties, that.optimizerProperties)
+ && Objects.equals(inputs, that.inputs)
+ && parallelism == that.parallelism;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(
+ id,
+ operator,
+ parallelism,
+ operatorStrategy,
+ description,
+ optimizerProperties,
+ inputs);
+ }
+
+ @JsonCreator
+ public Node(
+ @JsonProperty(FIELD_NAME_ID) String id,
+ @JsonProperty(FIELD_NAME_OPERATOR) String operator,
+ @JsonProperty(FIELD_NAME_PARALLELISM) long parallelism,
+ @JsonProperty(FIELD_NAME_OPERATOR_STRATEGY) String operatorStrategy,
+ @JsonProperty(FIELD_NAME_DESCRIPTION) String description,
+ @JsonProperty(FIELD_NAME_OPTIMIZER_PROPERTIES) String optimizerProperties,
+ @JsonProperty(FIELD_NAME_INPUTS) Collection<Input> inputs) {
+ this.id = Preconditions.checkNotNull(id);
+ this.operator = Preconditions.checkNotNull(operator);
+ this.parallelism = Preconditions.checkNotNull(parallelism);
+ this.operatorStrategy = Preconditions.checkNotNull(operatorStrategy);
+ this.description = Preconditions.checkNotNull(description);
+ this.optimizerProperties = Preconditions.checkNotNull(optimizerProperties);
+ this.inputs = Preconditions.checkNotNull(inputs);
+ }
+
+ @JsonIgnore
+ public String getId() {
+ return id;
+ }
+
+ @JsonIgnore
+ public long getParallelism() {
+ return parallelism;
+ }
+
+ @JsonIgnore
+ public String getOperator() {
+ return operator;
+ }
+
+ @JsonIgnore
+ public String getOperatorStrategy() {
+ return operatorStrategy;
+ }
+
+ @JsonIgnore
+ public String getDescription() {
+ return description;
+ }
+
+ @JsonIgnore
+ public String getOptimizerProperties() {
+ return optimizerProperties;
+ }
+
+ @JsonIgnore
+ public Collection<Input> getInputs() {
+ return inputs;
+ }
+
+ /**
+ * An inner class containing information on what input nodes should be linked to this
+ * node.
+ */
+ @Schema(name = "Input")
+ public static final class Input implements Serializable {
+ private static final String FIELD_NAME_NUM = "num";
+
+ @JsonProperty(FIELD_NAME_NUM)
+ private final long num;
+
+ private static final String FIELD_NAME_ID = "id";
+
+ @JsonProperty(FIELD_NAME_ID)
+ private final String id;
+
+ private static final String FIELD_NAME_SHIP_STRATEGY = "ship_strategy";
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_SHIP_STRATEGY)
+ private final String shipStrategy;
+
+ private static final String FIELD_NAME_LOCAL_STRATEGY = "local_strategy";
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_LOCAL_STRATEGY)
+ private final String localStrategy;
+
+ private static final String FIELD_NAME_CACHING = "caching";
+
+ @Nullable
+ @JsonProperty(FIELD_NAME_CACHING)
+ private final String caching;
+
+ private static final String FIELD_NAME_EXCHANGE = "exchange";
+
+ @JsonProperty(FIELD_NAME_EXCHANGE)
+ private final String exchange;
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ Input that = (Input) o;
+ return Objects.equals(id, that.id)
+ && num == that.num
+ && Objects.equals(shipStrategy, that.shipStrategy)
+ && Objects.equals(localStrategy, that.localStrategy)
+ && Objects.equals(caching, that.caching)
+ && Objects.equals(exchange, that.exchange);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, num, shipStrategy, localStrategy, caching, exchange);
+ }
+
+ @JsonCreator
+ public Input(
+ @JsonProperty(FIELD_NAME_ID) String id,
+ @JsonProperty(FIELD_NAME_NUM) long num,
+ @JsonProperty(FIELD_NAME_EXCHANGE) String exchange,
+ @Nullable @JsonProperty(FIELD_NAME_SHIP_STRATEGY) String shipStrategy,
+ @Nullable @JsonProperty(FIELD_NAME_LOCAL_STRATEGY) String localStrategy,
+ @Nullable @JsonProperty(FIELD_NAME_CACHING) String caching) {
+ this.id = Preconditions.checkNotNull(id);
+ this.num = Preconditions.checkNotNull(num);
+ this.exchange = Preconditions.checkNotNull(exchange);
+ this.shipStrategy = shipStrategy;
+ this.localStrategy = localStrategy;
+ this.caching = caching;
+ }
+
+ @JsonIgnore
+ public long getNum() {
+ return num;
+ }
+
+ @JsonIgnore
+ public String getId() {
+ return id;
+ }
+
+ @JsonIgnore
+ public String getShipStrategy() {
+ return shipStrategy;
+ }
+
+ @JsonIgnore
+ public String getLocalStrategy() {
+ return localStrategy;
+ }
+
+ @JsonIgnore
+ public String getCaching() {
+ return caching;
+ }
+
+ @JsonIgnore
+ public String getExchange() {
+ return exchange;
+ }
+ }
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
index 250afd3a041..ff2330692cb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java
@@ -131,7 +131,7 @@ public class JobDetailsInfo implements ResponseBody {
private final Map<ExecutionState, Integer> jobVerticesPerState;
@JsonProperty(FIELD_NAME_JSON_PLAN)
- private final JobPlanInfo.RawJson jsonPlan;
+ private final JobPlanInfo.Plan plan;
@JsonProperty(FIELD_NAME_STREAM_GRAPH_JSON)
@JsonInclude(JsonInclude.Include.NON_NULL)
@@ -159,7 +159,7 @@ public class JobDetailsInfo implements ResponseBody {
Collection<JobVertexDetailsInfo> jobVertexInfos,
@JsonProperty(FIELD_NAME_JOB_VERTICES_PER_STATE)
Map<ExecutionState, Integer> jobVerticesPerState,
- @JsonProperty(FIELD_NAME_JSON_PLAN) JobPlanInfo.RawJson jsonPlan,
+ @JsonProperty(FIELD_NAME_JSON_PLAN) JobPlanInfo.Plan plan,
@JsonProperty(FIELD_NAME_STREAM_GRAPH_JSON) @Nullable
JobPlanInfo.RawJson streamGraphJson,
@JsonProperty(FIELD_NAME_PENDING_OPERATORS) int pendingOperators) {
@@ -176,7 +176,7 @@ public class JobDetailsInfo implements ResponseBody {
this.timestamps = Preconditions.checkNotNull(timestamps);
this.jobVertexInfos = Preconditions.checkNotNull(jobVertexInfos);
this.jobVerticesPerState = Preconditions.checkNotNull(jobVerticesPerState);
- this.jsonPlan = Preconditions.checkNotNull(jsonPlan);
+ this.plan = Preconditions.checkNotNull(plan);
this.streamGraphJson = streamGraphJson;
this.pendingOperators = pendingOperators;
}
@@ -203,7 +203,7 @@ public class JobDetailsInfo implements ResponseBody {
&& Objects.equals(timestamps, that.timestamps)
&& Objects.equals(jobVertexInfos, that.jobVertexInfos)
&& Objects.equals(jobVerticesPerState, that.jobVerticesPerState)
- && Objects.equals(jsonPlan, that.jsonPlan)
+ && Objects.equals(plan, that.plan)
&& Objects.equals(streamGraphJson, that.streamGraphJson)
&& Objects.equals(pendingOperators, that.pendingOperators);
}
@@ -224,7 +224,7 @@ public class JobDetailsInfo implements ResponseBody {
timestamps,
jobVertexInfos,
jobVerticesPerState,
- jsonPlan,
+ plan,
streamGraphJson,
pendingOperators);
}
@@ -295,8 +295,8 @@ public class JobDetailsInfo implements ResponseBody {
}
@JsonIgnore
- public String getJsonPlan() {
- return jsonPlan.toString();
+ public JobPlanInfo.Plan getPlan() {
+ return plan;
}
@JsonIgnore
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 44fa1a9b68b..021954bee54 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
@@ -27,6 +27,7 @@ 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.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.DefaultOperatorCoordinatorHandler;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.GlobalFailureHandler;
@@ -123,7 +124,7 @@ public class CreatingExecutionGraph extends StateWithoutExecutionGraph {
operatorCoordinatorHandlerFactory.create(executionGraph, context);
operatorCoordinatorHandler.initializeOperatorCoordinators(
context.getMainThreadExecutor());
- final String updatedPlan =
+ final JobPlanInfo.Plan updatedPlan =
JsonPlanGenerator.generatePlan(
executionGraph.getJobID(),
executionGraph.getJobName(),
@@ -135,7 +136,7 @@ public class CreatingExecutionGraph extends StateWithoutExecutionGraph {
.map(ExecutionJobVertex::getJobVertex)
.iterator(),
executionGraphWithVertexParallelism.getVertexParallelism());
- executionGraph.setJsonPlan(updatedPlan);
+ executionGraph.setPlan(updatedPlan);
executionGraph.transitionToRunning();
operatorCoordinatorHandler.startAllOperatorCoordinators();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
index dcd410d423c..c8e31f4cdab 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveBatchScheduler.java
@@ -64,6 +64,7 @@ import org.apache.flink.runtime.jobgraph.topology.DefaultLogicalTopology;
import org.apache.flink.runtime.jobgraph.topology.DefaultLogicalVertex;
import org.apache.flink.runtime.jobmaster.event.ExecutionJobVertexFinishedEvent;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.DefaultExecutionDeployer;
import org.apache.flink.runtime.scheduler.DefaultScheduler;
import org.apache.flink.runtime.scheduler.ExecutionGraphFactory;
@@ -274,7 +275,7 @@ public class AdaptiveBatchScheduler extends DefaultScheduler implements JobGraph
logicalTopology = DefaultLogicalTopology.fromJobGraph(getJobGraph());
// 4. update json plan
- getExecutionGraph().setJsonPlan(JsonPlanGenerator.generatePlan(getJobGraph()));
+ getExecutionGraph().setPlan(JsonPlanGenerator.generatePlan(getJobGraph()));
// 5. In broadcast join optimization, results might be written first with a hash
// method and then read with a broadcast method. Therefore, we need to update the
@@ -837,11 +838,11 @@ public class AdaptiveBatchScheduler extends DefaultScheduler implements JobGraph
// job vertices
jobVertex.getJobVertex().setDynamicParallelism(parallelism);
try {
- getExecutionGraph().setJsonPlan(JsonPlanGenerator.generatePlan(getJobGraph()));
+ getExecutionGraph().setPlan(JsonPlanGenerator.generatePlan(getJobGraph()));
} catch (Throwable t) {
- log.warn("Cannot create JSON plan for job", t);
+ log.warn("Cannot create plan for job", t);
// give the graph an empty plan
- getExecutionGraph().setJsonPlan("{}");
+ getExecutionGraph().setPlan(new JobPlanInfo.Plan("", "", "", new ArrayList<>()));
}
jobVertex.setParallelism(parallelism);
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java
index 6ff8992d615..a143827f750 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/ExecutionGraphInfoStoreTestUtils.java
@@ -121,8 +121,7 @@ public class ExecutionGraphInfoStoreTestUtils {
&& Objects.equals(
thisExecutionGraph.getJobName(), thatExecutionGraph.getJobName())
&& thisExecutionGraph.getState() == thatExecutionGraph.getState()
- && Objects.equals(
- thisExecutionGraph.getJsonPlan(), thatExecutionGraph.getJsonPlan())
+ && Objects.equals(thisExecutionGraph.getPlan(), thatExecutionGraph.getPlan())
&& Objects.equals(
thisExecutionGraph.getAccumulatorsSerialized(),
thatExecutionGraph.getAccumulatorsSerialized())
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
index 06e89c6c258..65bcf236cec 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionGraphTest.java
@@ -254,7 +254,7 @@ public class ArchivedExecutionGraphTest {
// -------------------------------------------------------------------------------------------------------------
// ExecutionGraph
// -------------------------------------------------------------------------------------------------------------
- assertThat(runtimeGraph.getJsonPlan()).isEqualTo(archivedGraph.getJsonPlan());
+ assertThat(runtimeGraph.getPlan()).isEqualTo(archivedGraph.getPlan());
assertThat(runtimeGraph.getJobID()).isEqualTo(archivedGraph.getJobID());
assertThat(runtimeGraph.getJobName()).isEqualTo(archivedGraph.getJobName());
assertThat(runtimeGraph.getState()).isEqualTo(archivedGraph.getState());
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonGeneratorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonGeneratorTest.java
index 3df0e283bd7..b31f6c6bc1d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonGeneratorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonGeneratorTest.java
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.graph.StreamEdge;
import org.apache.flink.streaming.api.graph.StreamGraph;
@@ -32,20 +33,18 @@ import org.apache.flink.streaming.api.graph.StreamNode;
import org.apache.flink.util.jackson.JacksonMapperFactory;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.TextNode;
import org.junit.Test;
import java.util.ArrayList;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@@ -116,29 +115,21 @@ public class JsonGeneratorTest {
sink1,
sink2);
- String plan = JsonPlanGenerator.generatePlan(jg);
+ JobPlanInfo.Plan plan = JsonPlanGenerator.generatePlan(jg);
assertNotNull(plan);
- // validate the produced JSON
- ObjectMapper m = JacksonMapperFactory.createObjectMapper();
- JsonNode rootNode = m.readTree(plan);
-
// core fields
- assertEquals(new TextNode(jg.getJobID().toString()), rootNode.get("jid"));
- assertEquals(new TextNode(jg.getName()), rootNode.get("name"));
- assertEquals(new TextNode(jg.getJobType().name()), rootNode.get("type"));
-
- assertTrue(rootNode.path("nodes").isArray());
+ assertEquals(jg.getJobID().toString(), plan.getJobId());
+ assertEquals(jg.getName(), plan.getName());
+ assertEquals(jg.getJobType().name(), plan.getType());
- for (Iterator<JsonNode> iter = rootNode.path("nodes").elements(); iter.hasNext(); ) {
- JsonNode next = iter.next();
+ assertThat(plan.getNodes()).isNotEmpty();
+ assertThat(plan.getNodes().size()).isEqualTo(9);
- JsonNode idNode = next.get("id");
- assertNotNull(idNode);
- assertTrue(idNode.isTextual());
- checkVertexExists(idNode.asText(), jg);
+ for (JobPlanInfo.Plan.Node node : plan.getNodes()) {
+ checkVertexExists(node.getId(), jg);
- String description = next.get("description").asText();
+ String description = node.getDescription();
assertTrue(
description.startsWith("source")
|| description.startsWith("sink")
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
index 3ed10dc0e12..6c7a0732b4e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionGraphBuilder.java
@@ -28,6 +28,7 @@ import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
import org.apache.flink.runtime.jobgraph.JobType;
import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.util.OptionalFailure;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.SerializedValue;
@@ -51,7 +52,7 @@ public class ArchivedExecutionGraphBuilder {
private long[] stateTimestamps;
private JobStatus state;
private ErrorInfo failureCause;
- private String jsonPlan;
+ private JobPlanInfo.Plan plan;
private StringifiedAccumulatorResult[] archivedUserAccumulators;
private ArchivedExecutionConfig archivedExecutionConfig;
private boolean isStoppable;
@@ -98,8 +99,8 @@ public class ArchivedExecutionGraphBuilder {
return this;
}
- public ArchivedExecutionGraphBuilder setJsonPlan(String jsonPlan) {
- this.jsonPlan = jsonPlan;
+ public ArchivedExecutionGraphBuilder setPlan(JobPlanInfo.Plan plan) {
+ this.plan = plan;
return this;
}
@@ -161,13 +162,9 @@ public class ArchivedExecutionGraphBuilder {
state != null ? state : JobStatus.FINISHED,
JobType.STREAMING,
failureCause,
- jsonPlan != null
- ? jsonPlan
- : "{\"jobid\":\""
- + jobID
- + "\", \"name\":\""
- + jobName
- + "\", \"nodes\":[]}",
+ plan != null
+ ? plan
+ : new JobPlanInfo.Plan(jobID.toString(), jobName, "", new ArrayList<>()),
archivedUserAccumulators != null
? archivedUserAccumulators
: new StringifiedAccumulatorResult[0],
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfoTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfoTest.java
index fa346bdfbad..c4361507f10 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfoTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfoTest.java
@@ -50,7 +50,7 @@ class JobDetailsInfoTest extends RestResponseMarshallingTestBase<JobDetailsInfo>
protected JobDetailsInfo getTestResponseInstance() throws Exception {
final Random random = new Random();
final int numJobVertexDetailsInfos = 4;
- final String jsonPlan = "{\"id\":\"1234\"}";
+ final JobPlanInfo.Plan plan = new JobPlanInfo.Plan("1234", "", "", new ArrayList<>());
final Map<JobStatus, Long> timestamps = new HashMap<>(JobStatus.values().length);
final Collection<JobDetailsInfo.JobVertexDetailsInfo> jobVertexInfos =
@@ -84,7 +84,7 @@ class JobDetailsInfoTest extends RestResponseMarshallingTestBase<JobDetailsInfo>
timestamps,
jobVertexInfos,
jobVerticesPerState,
- new JobPlanInfo.RawJson(jsonPlan),
+ plan,
null,
0);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index 54695b78b15..be8492e9a41 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -121,7 +121,6 @@ import org.apache.flink.util.FlinkException;
import org.apache.flink.util.IterableUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.apache.flink.util.jackson.JacksonMapperFactory;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -459,12 +458,7 @@ public class AdaptiveSchedulerTest {
assertThat(executionGraph.getJobVertex(JOB_VERTEX.getID()).getParallelism())
.isEqualTo(numAvailableSlots);
- assertThat(
- JacksonMapperFactory.createObjectMapper()
- .readTree(executionGraph.getJsonPlan())
- .get("nodes")
- .size())
- .isOne();
+ assertThat(executionGraph.getPlan().getNodes().size()).isOne();
}
@Test
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 becb10fdb8a..1c70ad7554f 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
@@ -56,6 +56,7 @@ import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.query.KvStateLocationRegistry;
+import org.apache.flink.runtime.rest.messages.JobPlanInfo;
import org.apache.flink.runtime.scheduler.InternalFailuresListener;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
import org.apache.flink.runtime.scheduler.exceptionhistory.TestingAccessExecution;
@@ -73,6 +74,7 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -183,8 +185,8 @@ class StateTrackingMockExecutionGraph implements ExecutionGraph {
}
@Override
- public String getJsonPlan() {
- return "";
+ public JobPlanInfo.Plan getPlan() {
+ return new JobPlanInfo.Plan("", "", "", new ArrayList<>());
}
@Override
@@ -193,7 +195,7 @@ class StateTrackingMockExecutionGraph implements ExecutionGraph {
}
@Override
- public void setJsonPlan(String jsonPlan) {}
+ public void setPlan(JobPlanInfo.Plan jsonPlan) {}
@Override
public JobID getJobID() {
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 4ef8c946c92..dd8865c2509 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
@@ -37,9 +37,6 @@ import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.util.RestartStrategyUtils;
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;
@@ -56,8 +53,6 @@ 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
@@ -140,10 +135,8 @@ public class ReactiveModeITCase extends TestLogger {
.get();
assertThat(
- OBJECT_MAPPER
- .readTree(archivedExecutionGraph.getJsonPlan())
- .findValues("parallelism"))
- .allMatch(n -> n.asInt() == initialParallelism);
+ archivedExecutionGraph.getPlan().getNodes().stream()
+ .allMatch(n -> n.getParallelism() == (long) initialParallelism));
// scale up to 2 TaskManagers:
miniClusterResource.getMiniCluster().startTaskManager();
@@ -162,10 +155,8 @@ public class ReactiveModeITCase extends TestLogger {
.get();
assertThat(
- OBJECT_MAPPER
- .readTree(archivedExecutionGraph.getJsonPlan())
- .findValues("parallelism"))
- .allMatch(n -> n.asInt() == rescaledParallelism);
+ archivedExecutionGraph.getPlan().getNodes().stream()
+ .allMatch(n -> n.getParallelism() == (long) rescaledParallelism));
}
@Test