[FLINK-12763][runtime] Requests slots with ResourceProfiles that are …

…converted from ResourceSpecs.

flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceProfile.java

@@ -325,7 +325,7 @@ public String toString() {
  '}';
  }
 
- static ResourceProfile fromResourceSpec(ResourceSpec resourceSpec, int networkMemory) {
+ public static ResourceProfile fromResourceSpec(ResourceSpec resourceSpec, int networkMemory) {
  Map<String, Resource> copiedExtendedResources = new HashMap<>(resourceSpec.getExtendedResources());
 
  return new ResourceProfile(

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/AccessExecutionJobVertex.java

@@ -18,6 +18,7 @@
 package org.apache.flink.runtime.executiongraph;
 
 import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 
@@ -46,6 +47,13 @@ public interface AccessExecutionJobVertex {
  */
  int getMaxParallelism();
 
+ /**
+ * Returns the resource profile for this job vertex.
+ *
+ * @return resource profile for this job vertex.
+ */
+ ResourceProfile getResourceProfile();
+
  /**
  * Returns the {@link JobVertexID} for this job vertex.
  *

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ArchivedExecutionJobVertex.java

@@ -18,6 +18,7 @@
 package org.apache.flink.runtime.executiongraph;
 
 import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
 
@@ -38,6 +39,8 @@ public class ArchivedExecutionJobVertex implements AccessExecutionJobVertex, Ser
 
  private final int maxParallelism;
 
+ private final ResourceProfile resourceProfile;
+
  private final StringifiedAccumulatorResult[] archivedUserAccumulators;
 
  public ArchivedExecutionJobVertex(ExecutionJobVertex jobVertex) {
@@ -52,6 +55,7 @@ public ArchivedExecutionJobVertex(ExecutionJobVertex jobVertex) {
  this.name = jobVertex.getJobVertex().getName();
  this.parallelism = jobVertex.getParallelism();
  this.maxParallelism = jobVertex.getMaxParallelism();
+ this.resourceProfile = jobVertex.getResourceProfile();
  }
 
  public ArchivedExecutionJobVertex(
@@ -60,12 +64,14 @@ public ArchivedExecutionJobVertex(
  String name,
  int parallelism,
  int maxParallelism,
+ ResourceProfile resourceProfile,
  StringifiedAccumulatorResult[] archivedUserAccumulators) {
  this.taskVertices = taskVertices;
  this.id = id;
  this.name = name;
  this.parallelism = parallelism;
  this.maxParallelism = maxParallelism;
+ this.resourceProfile = resourceProfile;
  this.archivedUserAccumulators = archivedUserAccumulators;
  }
 
@@ -88,6 +94,11 @@ public int getMaxParallelism() {
  return maxParallelism;
  }
 
+ @Override
+ public ResourceProfile getResourceProfile() {
+ return resourceProfile;
+ }
+
  @Override
  public JobVertexID getJobVertexId() {
  return id;

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java

@@ -31,7 +31,6 @@
 import org.apache.flink.runtime.checkpoint.JobManagerTaskRestore;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.clusterframework.types.SlotProfile;
 import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
 import org.apache.flink.runtime.concurrent.FutureUtils;
@@ -577,7 +576,7 @@ private CompletableFuture<LogicalSlot> allocateAndAssignSlotForExecution(
  slotRequestId,
  toSchedule,
  new SlotProfile(
- ResourceProfile.UNKNOWN,
+ vertex.getResourceProfile(),
  preferredLocations,
  previousAllocationIDs,
  allPreviousExecutionGraphAllocationIds),

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionJobVertex.java

@@ -35,6 +35,7 @@
 import org.apache.flink.runtime.blob.BlobWriter;
 import org.apache.flink.runtime.blob.PermanentBlobKey;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.concurrent.FutureUtils;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSet;
@@ -127,6 +128,8 @@ public class ExecutionJobVertex implements AccessExecutionJobVertex, Archiveable
 
  private int maxParallelism;
 
+ private final ResourceProfile resourceProfile;
+
  /**
  * Either store a serialized task information, which is for all sub tasks the same,
  * or the permanent blob key of the offloaded task information BLOB containing
@@ -193,6 +196,7 @@ public ExecutionJobVertex(
  }
 
  this.parallelism = numTaskVertices;
+ this.resourceProfile = ResourceProfile.fromResourceSpec(jobVertex.getMinResources(), 0);
 
  this.taskVertices = new ExecutionVertex[numTaskVertices];
  this.operatorIDs = Collections.unmodifiableList(jobVertex.getOperatorIDs());
@@ -334,6 +338,11 @@ public int getMaxParallelism() {
  return maxParallelism;
  }
 
+ @Override
+ public ResourceProfile getResourceProfile() {
+ return resourceProfile;
+ }
+
  public boolean isMaxParallelismConfigured() {
  return maxParallelismConfigured;
  }

flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java

@@ -28,6 +28,7 @@
 import org.apache.flink.core.io.InputSplitAssigner;
 import org.apache.flink.runtime.JobException;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
 import org.apache.flink.runtime.jobgraph.DistributionPattern;
@@ -227,6 +228,10 @@ public int getMaxParallelism() {
  return this.jobVertex.getMaxParallelism();
  }
 
+ public ResourceProfile getResourceProfile() {
+ return this.jobVertex.getResourceProfile();
+ }
+
  @Override
  public int getParallelSubtaskIndex() {
  return this.subTaskIndex;

flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/SubtaskExecutionAttemptDetailsHandlerTest.java

@@ -23,6 +23,7 @@
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.MetricOptions;
 import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.executiongraph.ArchivedExecution;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
@@ -108,6 +109,7 @@ public void testHandleRequest() throws Exception {
  "test",
  1,
  1,
+ ResourceProfile.UNKNOWN,
  emptyAccumulators);
 
  // Change some fields so we can make it different from other sub tasks.

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

@@ -19,6 +19,7 @@
 package org.apache.flink.runtime.rest.handler.legacy.utils;
 
 import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex;
 import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
@@ -78,6 +79,7 @@ public ArchivedExecutionJobVertex build() {
  name != null ? name : "task_" + RANDOM.nextInt(),
  parallelism,
  maxParallelism,
+ ResourceProfile.UNKNOWN,
  archivedUserAccumulators != null ? archivedUserAccumulators : new StringifiedAccumulatorResult[0]
  );
  }