[FLINK-21303][coordination] Remove LogicalSlot#getPhysicalSlotNumber

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptor.java

@@ -128,9 +128,6 @@ public Offloaded(PermanentBlobKey serializedValueKey) {
     /** The list of consumed intermediate result partitions. */
     private final List<InputGateDeploymentDescriptor> inputGates;
 
-    /** Slot number to run the sub task in on the target machine. */
-    private final int targetSlotNumber;
-
     /** Information to restore the task. This can be null if there is no state to restore. */
     @Nullable private final JobManagerTaskRestore taskRestore;
 
@@ -142,7 +139,6 @@ public TaskDeploymentDescriptor(
             AllocationID allocationId,
             int subtaskIndex,
             int attemptNumber,
-            int targetSlotNumber,
             @Nullable JobManagerTaskRestore taskRestore,
             List<ResultPartitionDeploymentDescriptor> resultPartitionDeploymentDescriptors,
             List<InputGateDeploymentDescriptor> inputGateDeploymentDescriptors) {
@@ -161,10 +157,6 @@ public TaskDeploymentDescriptor(
         Preconditions.checkArgument(0 <= attemptNumber, "The attempt number must be positive.");
         this.attemptNumber = attemptNumber;
 
-        Preconditions.checkArgument(
-                0 <= targetSlotNumber, "The target slot number must be positive.");
-        this.targetSlotNumber = targetSlotNumber;
-
         this.taskRestore = taskRestore;
 
         this.producedPartitions = Preconditions.checkNotNull(resultPartitionDeploymentDescriptors);
@@ -234,15 +226,6 @@ public int getAttemptNumber() {
         return attemptNumber;
     }
 
-    /**
-     * Gets the number of the slot into which the task is to be deployed.
-     *
-     * @return The number of the target slot.
-     */
-    public int getTargetSlotNumber() {
-        return targetSlotNumber;
-    }
-
     public List<ResultPartitionDeploymentDescriptor> getProducedPartitions() {
         return producedPartitions;
     }

flink-runtime/src/main/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactory.java

@@ -85,7 +85,6 @@ private TaskDeploymentDescriptorFactory(
 
     public TaskDeploymentDescriptor createDeploymentDescriptor(
             AllocationID allocationID,
-            int targetSlotNumber,
             @Nullable JobManagerTaskRestore taskRestore,
             Collection<ResultPartitionDeploymentDescriptor> producedPartitions) {
         return new TaskDeploymentDescriptor(
@@ -96,7 +95,6 @@ public TaskDeploymentDescriptor createDeploymentDescriptor(
                 allocationID,
                 subtaskIndex,
                 attemptNumber,
-                targetSlotNumber,
                 taskRestore,
                 new ArrayList<>(producedPartitions),
                 createInputGateDeploymentDescriptors());

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

@@ -570,7 +570,6 @@ public void deploy() throws JobException {
                     TaskDeploymentDescriptorFactory.fromExecutionVertex(vertex, attemptNumber)
                             .createDeploymentDescriptor(
                                     slot.getAllocationId(),
-                                    slot.getPhysicalSlotNumber(),
                                     taskRestore,
                                     producedPartitions.values());
 

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/LogicalSlot.java

@@ -112,14 +112,6 @@ default CompletableFuture<?> releaseSlot() {
      */
     CompletableFuture<?> releaseSlot(@Nullable Throwable cause);
 
-    /**
-     * Gets the slot number on the TaskManager. Multiple logical slots can share the same slot
-     * number.
-     *
-     * @return slot number
-     */
-    int getPhysicalSlotNumber();
-
     /**
      * Gets the allocation id of this slot. Multiple logical slots can share the same allocation id.
      *

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SingleLogicalSlot.java

@@ -134,11 +134,6 @@ public CompletableFuture<?> releaseSlot(@Nullable Throwable cause) {
         return releaseFuture;
     }
 
-    @Override
-    public int getPhysicalSlotNumber() {
-        return slotContext.getPhysicalSlotNumber();
-    }
-
     @Override
     public AllocationID getAllocationId() {
         return slotContext.getAllocationId();

flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java

@@ -646,7 +646,6 @@ public CompletableFuture<Acknowledge> submitTask(
                             tdd.getAttemptNumber(),
                             tdd.getProducedPartitions(),
                             tdd.getInputGates(),
-                            tdd.getTargetSlotNumber(),
                             memoryManager,
                             taskExecutorServices.getIOManager(),
                             taskExecutorServices.getShuffleEnvironment(),

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java

@@ -304,7 +304,6 @@ public Task(
             int attemptNumber,
             List<ResultPartitionDeploymentDescriptor> resultPartitionDeploymentDescriptors,
             List<InputGateDeploymentDescriptor> inputGateDeploymentDescriptors,
-            int targetSlotNumber,
             MemoryManager memManager,
             IOManager ioManager,
             ShuffleEnvironment<?, ?> shuffleEnvironment,
@@ -331,8 +330,6 @@ public Task(
 
         Preconditions.checkArgument(0 <= subtaskIndex, "The subtask index must be positive.");
         Preconditions.checkArgument(0 <= attemptNumber, "The attempt number must be positive.");
-        Preconditions.checkArgument(
-                0 <= targetSlotNumber, "The target slot number must be positive.");
 
         this.taskInfo =
                 new TaskInfo(

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorBuilder.java

@@ -48,7 +48,6 @@ public class TaskDeploymentDescriptorBuilder {
     private int attemptNumber;
     private List<ResultPartitionDeploymentDescriptor> producedPartitions;
     private List<InputGateDeploymentDescriptor> inputGates;
-    private int targetSlotNumber;
 
     @Nullable private JobManagerTaskRestore taskRestore;
 
@@ -74,7 +73,6 @@ private TaskDeploymentDescriptorBuilder(JobID jobId, String invokableClassName)
         this.attemptNumber = 0;
         this.producedPartitions = Collections.emptyList();
         this.inputGates = Collections.emptyList();
-        this.targetSlotNumber = 0;
         this.taskRestore = null;
     }
 
@@ -127,11 +125,6 @@ public TaskDeploymentDescriptorBuilder setInputGates(
         return this;
     }
 
-    public TaskDeploymentDescriptorBuilder setTargetSlotNumber(int targetSlotNumber) {
-        this.targetSlotNumber = targetSlotNumber;
-        return this;
-    }
-
     public TaskDeploymentDescriptorBuilder setTaskRestore(
             @Nullable JobManagerTaskRestore taskRestore) {
         this.taskRestore = taskRestore;
@@ -147,7 +140,6 @@ public TaskDeploymentDescriptor build() {
                 allocationId,
                 subtaskIndex,
                 attemptNumber,
-                targetSlotNumber,
                 taskRestore,
                 producedPartitions,
                 inputGates);

flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorTest.java

@@ -72,7 +72,6 @@ public class TaskDeploymentDescriptorTest extends TestLogger {
             new ArrayList<InputGateDeploymentDescriptor>(0);
     private static final List<PermanentBlobKey> requiredJars = new ArrayList<>(0);
     private static final List<URL> requiredClasspaths = new ArrayList<>(0);
-    private static final int targetSlotNumber = 47;
     private static final TaskStateSnapshot taskStateHandles = new TaskStateSnapshot();
     private static final JobManagerTaskRestore taskRestore =
             new JobManagerTaskRestore(1L, taskStateHandles);
@@ -123,7 +122,6 @@ public void testSerialization() throws Exception {
         assertEquals(orig.getAllocationId(), copy.getAllocationId());
         assertEquals(orig.getSubtaskIndex(), copy.getSubtaskIndex());
         assertEquals(orig.getAttemptNumber(), copy.getAttemptNumber());
-        assertEquals(orig.getTargetSlotNumber(), copy.getTargetSlotNumber());
         assertEquals(
                 orig.getTaskRestore().getRestoreCheckpointId(),
                 copy.getTaskRestore().getRestoreCheckpointId());
@@ -165,7 +163,6 @@ private TaskDeploymentDescriptor createTaskDeploymentDescriptor(
                 allocationId,
                 indexInSubtaskGroup,
                 attemptNumber,
-                targetSlotNumber,
                 taskRestore,
                 producedResults,
                 inputGates);

flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexDeploymentTest.java

@@ -305,7 +305,6 @@ public void testTddProducedPartitionsLazyScheduling() throws Exception {
             TaskDeploymentDescriptor tdd =
                     tddFactory.createDeploymentDescriptor(
                             new AllocationID(),
-                            0,
                             null,
                             Execution.registerProducedPartitions(
                                             vertex,

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingLogicalSlot.java

@@ -38,8 +38,6 @@ public class TestingLogicalSlot implements LogicalSlot {
 
     private final AtomicReference<Payload> payloadReference;
 
-    private final int slotNumber;
-
     private final CompletableFuture<?> releaseFuture;
 
     private final boolean automaticallyCompleteReleaseFuture;
@@ -55,7 +53,6 @@ public class TestingLogicalSlot implements LogicalSlot {
     TestingLogicalSlot(
             TaskManagerLocation taskManagerLocation,
             TaskManagerGateway taskManagerGateway,
-            int slotNumber,
             AllocationID allocationId,
             SlotRequestId slotRequestId,
             boolean automaticallyCompleteReleaseFuture,
@@ -64,7 +61,6 @@ public class TestingLogicalSlot implements LogicalSlot {
         this.taskManagerLocation = Preconditions.checkNotNull(taskManagerLocation);
         this.taskManagerGateway = Preconditions.checkNotNull(taskManagerGateway);
         this.payloadReference = new AtomicReference<>();
-        this.slotNumber = slotNumber;
         this.allocationId = Preconditions.checkNotNull(allocationId);
         this.slotRequestId = Preconditions.checkNotNull(slotRequestId);
         this.releaseFuture = new CompletableFuture<>();
@@ -121,11 +117,6 @@ public CompletableFuture<?> releaseSlot(@Nullable Throwable cause) {
         return releaseFuture;
     }
 
-    @Override
-    public int getPhysicalSlotNumber() {
-        return slotNumber;
-    }
-
     @Override
     public AllocationID getAllocationId() {
         return allocationId;

flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingLogicalSlotBuilder.java

@@ -29,7 +29,6 @@
 public class TestingLogicalSlotBuilder {
     private TaskManagerGateway taskManagerGateway = new SimpleAckingTaskManagerGateway();
     private TaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
-    private int slotNumber = 0;
     private AllocationID allocationId = new AllocationID();
     private SlotRequestId slotRequestId = new SlotRequestId();
     private SlotOwner slotOwner = new DummySlotOwner();
@@ -46,11 +45,6 @@ public TestingLogicalSlotBuilder setTaskManagerLocation(
         return this;
     }
 
-    public TestingLogicalSlotBuilder setSlotNumber(int slotNumber) {
-        this.slotNumber = slotNumber;
-        return this;
-    }
-
     public TestingLogicalSlotBuilder setAllocationId(AllocationID allocationId) {
         this.allocationId = allocationId;
         return this;
@@ -76,7 +70,6 @@ public TestingLogicalSlot createTestingLogicalSlot() {
         return new TestingLogicalSlot(
                 taskManagerLocation,
                 taskManagerGateway,
-                slotNumber,
                 allocationId,
                 slotRequestId,
                 automaticallyCompleteReleaseFuture,

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/SharedSlotTest.java

@@ -88,14 +88,9 @@ public void testLogicalSlotAllocation() {
         AllocationID allocationId = new AllocationID();
         LocalTaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
         SimpleAckingTaskManagerGateway taskManagerGateway = new SimpleAckingTaskManagerGateway();
-        int physicalSlotNumber = 3;
         slotContextFuture.complete(
                 new TestingPhysicalSlot(
-                        allocationId,
-                        taskManagerLocation,
-                        physicalSlotNumber,
-                        taskManagerGateway,
-                        RP));
+                        allocationId, taskManagerLocation, 3, taskManagerGateway, RP));
 
         assertThat(sharedSlot.isEmpty(), is(false));
         assertThat(released.isDone(), is(false));
@@ -104,7 +99,6 @@ public void testLogicalSlotAllocation() {
         assertThat(logicalSlot.getAllocationId(), is(allocationId));
         assertThat(logicalSlot.getTaskManagerLocation(), is(taskManagerLocation));
         assertThat(logicalSlot.getTaskManagerGateway(), is(taskManagerGateway));
-        assertThat(logicalSlot.getPhysicalSlotNumber(), is(physicalSlotNumber));
         assertThat(logicalSlot.getLocality(), is(Locality.UNKNOWN));
     }
 

flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/declarative/allocator/SharedSlotTest.java

@@ -66,7 +66,6 @@ public void testAllocateLogicalSlot() {
         assertThat(logicalSlot.getAllocationId(), equalTo(physicalSlot.getAllocationId()));
         assertThat(logicalSlot.getLocality(), is(Locality.UNKNOWN));
         assertThat(logicalSlot.getPayload(), nullValue());
-        assertThat(logicalSlot.getPhysicalSlotNumber(), is(physicalSlot.getPhysicalSlotNumber()));
         assertThat(
                 logicalSlot.getTaskManagerLocation(),
                 equalTo(physicalSlot.getTaskManagerLocation()));

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorOperatorEventHandlingTest.java

@@ -182,8 +182,7 @@ private static TaskDeploymentDescriptor createTaskDeploymentDescriptor(
                 Collections.emptyList(),
                 Collections.emptyList(),
                 Collections.emptyList(),
-                Collections.emptyList(),
-                0);
+                Collections.emptyList());
     }
 
     // ------------------------------------------------------------------------

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorPartitionLifecycleTest.java

@@ -428,8 +428,7 @@ private void internalTestPartitionRelease(
                         Collections.singletonList(taskResultPartitionDescriptor),
                         Collections.emptyList(),
                         Collections.emptyList(),
-                        Collections.emptyList(),
-                        0);
+                        Collections.emptyList());
 
         final TaskSlotTable<Task> taskSlotTable = createTaskSlotTable();
 

flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorSubmissionTest.java

@@ -691,8 +691,7 @@ private TaskDeploymentDescriptor createTestTaskDeploymentDescriptor(
                 producedPartitions,
                 inputGates,
                 Collections.emptyList(),
-                Collections.emptyList(),
-                0);
+                Collections.emptyList());
     }
 
     static TaskDeploymentDescriptor createTaskDeploymentDescriptor(
@@ -711,8 +710,7 @@ static TaskDeploymentDescriptor createTaskDeploymentDescriptor(
             List<ResultPartitionDeploymentDescriptor> producedPartitions,
             List<InputGateDeploymentDescriptor> inputGates,
             Collection<PermanentBlobKey> requiredJarFiles,
-            Collection<URL> requiredClasspaths,
-            int targetSlotNumber)
+            Collection<URL> requiredClasspaths)
             throws IOException {
 
         JobInformation jobInformation =
@@ -746,7 +744,6 @@ static TaskDeploymentDescriptor createTaskDeploymentDescriptor(
                 new AllocationID(),
                 subtaskIndex,
                 attemptNumber,
-                targetSlotNumber,
                 null,
                 producedPartitions,
                 inputGates);

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

@@ -210,7 +210,6 @@ private Task createTask(Class<? extends AbstractInvokable> invokableClass) throw
                 0,
                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                0,
                 mock(MemoryManager.class),
                 mock(IOManager.class),
                 shuffleEnvironment,

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TestTaskBuilder.java

@@ -216,7 +216,6 @@ public Task build() throws Exception {
                 0,
                 resultPartitions,
                 inputGates,
-                0,
                 MemoryManagerBuilder.newBuilder().setMemorySize(1024 * 1024).build(),
                 mock(IOManager.class),
                 shuffleEnvironment,

flink-runtime/src/test/java/org/apache/flink/runtime/util/JvmExitOnFatalErrorTest.java

@@ -216,7 +216,6 @@ public static void main(String[] args) throws Exception {
                                 0, // attemptNumber
                                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                                0, // targetSlotNumber
                                 memoryManager,
                                 ioManager,
                                 shuffleEnvironment,

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/InterruptSensitiveRestoreTest.java

@@ -269,7 +269,6 @@ private static Task createTask(
                 0,
                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                0,
                 mock(MemoryManager.class),
                 mock(IOManager.class),
                 shuffleEnvironment,

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskSystemExitTest.java

@@ -187,7 +187,6 @@ private Task createSystemExitTask(final String invokableClassName, StreamOperato
                 0,
                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                0,
                 MemoryManagerBuilder.newBuilder().setMemorySize(32L * 1024L).build(),
                 new IOManagerAsync(),
                 shuffleEnvironment,

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTerminationTest.java

@@ -168,7 +168,6 @@ public void testConcurrentAsyncCheckpointCannotFailFinishedStreamTask() throws E
                         0,
                         Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                         Collections.<InputGateDeploymentDescriptor>emptyList(),
-                        0,
                         MemoryManagerBuilder.newBuilder().setMemorySize(32L * 1024L).build(),
                         new IOManagerAsync(),
                         shuffleEnvironment,

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SynchronousCheckpointITCase.java

@@ -262,7 +262,6 @@ private Task createTask(Class<? extends AbstractInvokable> invokableClass) throw
                 0,
                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                0,
                 mock(MemoryManager.class),
                 mock(IOManager.class),
                 shuffleEnvironment,

flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/TaskCheckpointingBehaviourTest.java

@@ -222,7 +222,6 @@ private static Task createTask(
                 0,
                 Collections.<ResultPartitionDeploymentDescriptor>emptyList(),
                 Collections.<InputGateDeploymentDescriptor>emptyList(),
-                0,
                 mock(MemoryManager.class),
                 mock(IOManager.class),
                 shuffleEnvironment,