[FLINK-12607][rest] Expose maxParallelism of jobs and vertices

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -1468,6 +1468,9 @@
  "jid" : {
  "type" : "any"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "name" : {
  "type" : "string"
  },
@@ -1511,6 +1514,9 @@
  "id" : {
  "type" : "any"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "metrics" : {
  "type" : "object",
  "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:job:metrics:IOMetricsInfo",
@@ -3429,6 +3435,9 @@
  "id" : {
  "type" : "any"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "name" : {
  "type" : "string"
  },

flink-core/src/main/java/org/apache/flink/api/common/ArchivedExecutionConfig.java

@@ -32,11 +32,12 @@
 @Internal
 public class ArchivedExecutionConfig implements Serializable {
 
- private static final long serialVersionUID = 2126156250920316528L;
+ private static final long serialVersionUID = 2907040336948181163L;
 
  private final String executionMode;
  private final String restartStrategyDescription;
  private final int parallelism;
+ private final int maxParallelism;
  private final boolean objectReuseEnabled;
  private final Map<String, String> globalJobParameters;
 
@@ -47,6 +48,7 @@ public ArchivedExecutionConfig(ExecutionConfig ec) {
  } else {
  restartStrategyDescription = "default";
  }
+ maxParallelism = ec.getMaxParallelism();
  parallelism = ec.getParallelism();
  objectReuseEnabled = ec.isObjectReuseEnabled();
  if (ec.getGlobalJobParameters() != null && ec.getGlobalJobParameters().toMap() != null) {
@@ -59,11 +61,13 @@ public ArchivedExecutionConfig(ExecutionConfig ec) {
  public ArchivedExecutionConfig(
  String executionMode,
  String restartStrategyDescription,
+ int maxParallelism,
  int parallelism,
  boolean objectReuseEnabled,
  Map<String, String> globalJobParameters) {
  this.executionMode = executionMode;
  this.restartStrategyDescription = restartStrategyDescription;
+ this.maxParallelism = maxParallelism;
  this.parallelism = parallelism;
  this.objectReuseEnabled = objectReuseEnabled;
  this.globalJobParameters = globalJobParameters;
@@ -77,6 +81,10 @@ public String getRestartStrategyDescription() {
  return restartStrategyDescription;
  }
 
+ public int getMaxParallelism() {
+ return maxParallelism;
+ }
+
  public int getParallelism() {
  return parallelism;
  }

flink-runtime-web/src/test/resources/rest_api_v1.snapshot

@@ -708,6 +708,9 @@
  "duration" : {
  "type" : "integer"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "now" : {
  "type" : "integer"
  },
@@ -729,6 +732,9 @@
  "name" : {
  "type" : "string"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "parallelism" : {
  "type" : "integer"
  },
@@ -1887,6 +1893,9 @@
  "parallelism" : {
  "type" : "integer"
  },
+ "maxParallelism" : {
+ "type" : "integer"
+ },
  "now" : {
  "type" : "integer"
  },

flink-runtime-web/web-dashboard/src/app/interfaces/job-detail.ts

@@ -50,6 +50,7 @@ export interface JobDetailInterface {
  'start-time': number;
  'end-time': number;
  duration: number;
+ maxParallelism: number;
  now: number;
  timestamps: TimestampsStatus;
  vertices: VerticesItemInterface[];
@@ -80,6 +81,7 @@ export interface VerticesItemInterface {
  id: string;
  name: string;
  parallelism: number;
+ maxParallelism: number;
  status: string;
  'start-time': number;
  'end-time': number;

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java

@@ -148,6 +148,7 @@ private static JobDetailsInfo createJobDetailsInfo(
  startTime,
  endTime,
  duration,
+ executionGraph.getArchivedExecutionConfig().getMaxParallelism(),
  now,
  timestamps,
  jobVertexInfos,
@@ -223,6 +224,7 @@ private static JobDetailsInfo.JobVertexDetailsInfo createJobVertexDetailsInfo(
  return new JobDetailsInfo.JobVertexDetailsInfo(
  ejv.getJobVertexId(),
  ejv.getName(),
+ ejv.getMaxParallelism(),
  ejv.getParallelism(),
  jobVertexState,
  startTime,

flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexDetailsHandler.java

@@ -130,6 +130,7 @@ private static JobVertexDetailsInfo createJobVertexDetailsInfo(
  jobVertex.getJobVertexId(),
  jobVertex.getName(),
  jobVertex.getParallelism(),
+ jobVertex.getMaxParallelism(),
  now,
  subtasks);
  }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexDetailsInfo.java

@@ -41,6 +41,7 @@ public class JobVertexDetailsInfo implements ResponseBody {
  public static final String FIELD_NAME_VERTEX_ID = "id";
  public static final String FIELD_NAME_VERTEX_NAME = "name";
  public static final String FIELD_NAME_PARALLELISM = "parallelism";
+ public static final String FIELD_NAME_MAX_PARALLELISM = "maxParallelism";
  public static final String FIELD_NAME_NOW = "now";
  public static final String FIELD_NAME_SUBTASKS = "subtasks";
 
@@ -54,6 +55,9 @@ public class JobVertexDetailsInfo implements ResponseBody {
  @JsonProperty(FIELD_NAME_PARALLELISM)
  private final int parallelism;
 
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM)
+ private final int maxParallelism;
+
  @JsonProperty(FIELD_NAME_NOW)
  private final long now;
 
@@ -67,11 +71,13 @@ public JobVertexDetailsInfo(
  JobVertexID id,
  @JsonProperty(FIELD_NAME_VERTEX_NAME) String name,
  @JsonProperty(FIELD_NAME_PARALLELISM) int parallelism,
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM) int maxParallelism,
  @JsonProperty(FIELD_NAME_NOW) long now,
  @JsonProperty(FIELD_NAME_SUBTASKS) List<SubtaskExecutionAttemptDetailsInfo> subtasks) {
  this.id = checkNotNull(id);
  this.name = checkNotNull(name);
  this.parallelism = parallelism;
+ this.maxParallelism = maxParallelism;
  this.now = now;
  this.subtasks = checkNotNull(subtasks);
  }
@@ -95,12 +101,13 @@ public boolean equals(Object o) {
  return Objects.equals(id, that.id)
  && Objects.equals(name, that.name)
  && parallelism == that.parallelism
+ && maxParallelism == that.maxParallelism
  && now == that.now
  && Objects.equals(subtasks, that.subtasks);
  }
 
  @Override
  public int hashCode() {
- return Objects.hash(id, name, parallelism, now, subtasks);
+ return Objects.hash(id, name, parallelism, maxParallelism, now, subtasks);
  }
 }

flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfo.java

@@ -59,6 +59,8 @@ public class JobDetailsInfo implements ResponseBody {
 
  public static final String FIELD_NAME_DURATION = "duration";
 
+ public static final String FIELD_NAME_MAX_PARALLELISM = "maxParallelism";
+
  // TODO: For what do we need this???
  public static final String FIELD_NAME_NOW = "now";
 
@@ -92,6 +94,9 @@ public class JobDetailsInfo implements ResponseBody {
  @JsonProperty(FIELD_NAME_DURATION)
  private final long duration;
 
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM)
+ private final long maxParallelism;
+
  @JsonProperty(FIELD_NAME_NOW)
  private final long now;
 
@@ -118,6 +123,7 @@ public JobDetailsInfo(
  @JsonProperty(FIELD_NAME_START_TIME) long startTime,
  @JsonProperty(FIELD_NAME_END_TIME) long endTime,
  @JsonProperty(FIELD_NAME_DURATION) long duration,
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM) long maxParallelism,
  @JsonProperty(FIELD_NAME_NOW) long now,
  @JsonProperty(FIELD_NAME_TIMESTAMPS) Map<JobStatus, Long> timestamps,
  @JsonProperty(FIELD_NAME_JOB_VERTEX_INFOS)
@@ -133,6 +139,7 @@ public JobDetailsInfo(
  this.startTime = startTime;
  this.endTime = endTime;
  this.duration = duration;
+ this.maxParallelism = maxParallelism;
  this.now = now;
  this.timestamps = Preconditions.checkNotNull(timestamps);
  this.jobVertexInfos = Preconditions.checkNotNull(jobVertexInfos);
@@ -153,6 +160,7 @@ public boolean equals(Object o) {
  && startTime == that.startTime
  && endTime == that.endTime
  && duration == that.duration
+ && maxParallelism == that.maxParallelism
  && now == that.now
  && Objects.equals(jobId, that.jobId)
  && Objects.equals(name, that.name)
@@ -173,6 +181,7 @@ public int hashCode() {
  startTime,
  endTime,
  duration,
+ maxParallelism,
  now,
  timestamps,
  jobVertexInfos,
@@ -210,6 +219,11 @@ public long getEndTime() {
  return endTime;
  }
 
+ @JsonIgnore
+ public long getMaxParallelism() {
+ return maxParallelism;
+ }
+
  @JsonIgnore
  public long getDuration() {
  return duration;
@@ -251,6 +265,8 @@ public static final class JobVertexDetailsInfo {
 
  public static final String FIELD_NAME_JOB_VERTEX_NAME = "name";
 
+ public static final String FIELD_NAME_MAX_PARALLELISM = "maxParallelism";
+
  public static final String FIELD_NAME_PARALLELISM = "parallelism";
 
  public static final String FIELD_NAME_JOB_VERTEX_STATE = "status";
@@ -272,6 +288,9 @@ public static final class JobVertexDetailsInfo {
  @JsonProperty(FIELD_NAME_JOB_VERTEX_NAME)
  private final String name;
 
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM)
+ private final int maxParallelism;
+
  @JsonProperty(FIELD_NAME_PARALLELISM)
  private final int parallelism;
 
@@ -299,6 +318,7 @@ public JobVertexDetailsInfo(
  @JsonProperty(FIELD_NAME_JOB_VERTEX_ID)
  JobVertexID jobVertexID,
  @JsonProperty(FIELD_NAME_JOB_VERTEX_NAME) String name,
+ @JsonProperty(FIELD_NAME_MAX_PARALLELISM) int maxParallelism,
  @JsonProperty(FIELD_NAME_PARALLELISM) int parallelism,
  @JsonProperty(FIELD_NAME_JOB_VERTEX_STATE) ExecutionState executionState,
  @JsonProperty(FIELD_NAME_JOB_VERTEX_START_TIME) long startTime,
@@ -309,6 +329,7 @@ public JobVertexDetailsInfo(
  @JsonProperty(FIELD_NAME_JOB_VERTEX_METRICS) IOMetricsInfo jobVertexMetrics) {
  this.jobVertexID = Preconditions.checkNotNull(jobVertexID);
  this.name = Preconditions.checkNotNull(name);
+ this.maxParallelism = maxParallelism;
  this.parallelism = parallelism;
  this.executionState = Preconditions.checkNotNull(executionState);
  this.startTime = startTime;
@@ -328,6 +349,11 @@ public String getName() {
  return name;
  }
 
+ @JsonIgnore
+ public int getMaxParallelism() {
+ return maxParallelism;
+ }
+
  @JsonIgnore
  public int getParallelism() {
  return parallelism;
@@ -372,7 +398,8 @@ public boolean equals(Object o) {
  return false;
  }
  JobVertexDetailsInfo that = (JobVertexDetailsInfo) o;
- return parallelism == that.parallelism
+ return maxParallelism == that.maxParallelism
+ && parallelism == that.parallelism
  && startTime == that.startTime
  && endTime == that.endTime
  && duration == that.duration
@@ -388,6 +415,7 @@ public int hashCode() {
  return Objects.hash(
  jobVertexID,
  name,
+ maxParallelism,
  parallelism,
  executionState,
  startTime,

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/utils/ArchivedExecutionConfigBuilder.java

@@ -28,6 +28,7 @@
 public class ArchivedExecutionConfigBuilder {
  private String executionMode;
  private String restartStrategyDescription;
+ private int maxParallelism;
  private int parallelism;
  private boolean objectReuseEnabled;
  private Map<String, String> globalJobParameters;
@@ -48,6 +49,11 @@ public ArchivedExecutionConfigBuilder setParallelism(int parallelism) {
  return this;
  }
 
+ public ArchivedExecutionConfigBuilder setMaxParallelism(int parallelism) {
+ this.maxParallelism = maxParallelism;
+ return this;
+ }
+
  public ArchivedExecutionConfigBuilder setObjectReuseEnabled(boolean objectReuseEnabled) {
  this.objectReuseEnabled = objectReuseEnabled;
  return this;
@@ -63,6 +69,7 @@ public ArchivedExecutionConfig build() {
  return new ArchivedExecutionConfig(
  executionMode != null ? executionMode : ExecutionMode.PIPELINED.name(),
  restartStrategyDescription != null ? restartStrategyDescription : "default",
+ maxParallelism,
  parallelism,
  objectReuseEnabled,
  globalJobParameters != null

flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/JobVertexDetailsInfoTest.java

@@ -83,10 +83,12 @@ protected JobVertexDetailsInfo getTestResponseInstance() throws Exception {
  jobVertexMetrics,
  "taskmanagerId3"));
 
+ int parallelism = 1 + (random.nextInt() / 3);
  return new JobVertexDetailsInfo(
  new JobVertexID(),
  "jobVertex" + random.nextLong(),
- random.nextInt(),
+ parallelism,
+ 2 * parallelism,
  System.currentTimeMillis(),
  vertexTaskDetailList);
  }

flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobDetailsInfoTest.java

@@ -71,6 +71,7 @@ protected JobDetailsInfo getTestResponseInstance() throws Exception {
  1L,
  2L,
  1L,
+ 8888L,
  1984L,
  timestamps,
  jobVertexInfos,
@@ -96,10 +97,12 @@ private JobDetailsInfo.JobVertexDetailsInfo createJobVertexDetailsInfo(Random ra
  tasksPerState.put(executionState, random.nextInt());
  }
 
+ int parallelism = 1 + (random.nextInt() / 3);
  return new JobDetailsInfo.JobVertexDetailsInfo(
  new JobVertexID(),
  "jobVertex" + random.nextLong(),
- random.nextInt(),
+ 2 * parallelism,
+ parallelism,
  ExecutionState.values()[random.nextInt(ExecutionState.values().length)],
  random.nextLong(),
  random.nextLong(),