[hotfix][scheduler] Migrate the Time to Duration for SlotPool in the …

…minimum scope
rohan-flutterint · Jul 31, 2024 · 82b628d · 82b628d
1 parent 76049d0
commit 82b628d
Show file tree

Hide file tree

Showing 10 changed files with 27 additions and 30 deletions.
diff --git a/.../src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java b/.../src/main/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridge.java
@@ -20,7 +20,6 @@
 
 import org.apache.flink.annotation.VisibleForTesting;
 import org.apache.flink.api.common.JobID;
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -282,7 +281,7 @@ public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
             @Nonnull SlotRequestId slotRequestId,
             @Nonnull ResourceProfile resourceProfile,
             @Nonnull Collection<AllocationID> preferredAllocations,
-            @Nullable Time timeout) {
+            @Nullable Duration timeout) {
         assertRunningInMainThread();
 
         log.debug(
@@ -318,20 +317,20 @@ public CompletableFuture<PhysicalSlot> requestNewAllocatedBatchSlot(
     }
 
     private CompletableFuture<PhysicalSlot> internalRequestNewSlot(
-            PendingRequest pendingRequest, @Nullable Time timeout) {
+            PendingRequest pendingRequest, @Nullable Duration timeout) {
         internalRequestNewAllocatedSlot(pendingRequest);
 
         if (timeout == null) {
             return pendingRequest.getSlotFuture();
         } else {
             return FutureUtils.orTimeout(
                             pendingRequest.getSlotFuture(),
-                            timeout.toMilliseconds(),
+                            timeout.toMillis(),
                             TimeUnit.MILLISECONDS,
                             componentMainThreadExecutor,
                             String.format(
                                     "Pending slot request %s timed out after %d ms.",
-                                    pendingRequest.getSlotRequestId(), timeout.toMilliseconds()))
+                                    pendingRequest.getSlotRequestId(), timeout.toMillis()))
                     .whenComplete(
                             (physicalSlot, throwable) -> {
                                 if (throwable instanceof TimeoutException) {

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -33,6 +32,7 @@
 
 import javax.annotation.Nullable;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.Optional;
@@ -154,7 +154,9 @@ Optional<PhysicalSlot> allocateAvailableSlot(
      * @return a newly allocated slot that was previously not available.
      */
     default CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
-            SlotRequestId slotRequestId, ResourceProfile resourceProfile, @Nullable Time timeout) {
+            SlotRequestId slotRequestId,
+            ResourceProfile resourceProfile,
+            @Nullable Duration timeout) {
         return requestNewAllocatedSlot(
                 slotRequestId, resourceProfile, Collections.emptyList(), timeout);
     }
@@ -175,7 +177,7 @@ CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
             SlotRequestId slotRequestId,
             ResourceProfile resourceProfile,
             Collection<AllocationID> preferredAllocations,
-            @Nullable Time timeout);
+            @Nullable Duration timeout);
 
     /**
      * Requests the allocation of a new batch slot from the resource manager. Unlike the normal

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -610,7 +610,7 @@ public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
                 @Nonnull SlotRequestId slotRequestId,
                 @Nonnull ResourceProfile resourceProfile,
                 @Nonnull Collection<AllocationID> preferredAllocations,
-                @Nullable Time timeout) {
+                @Nullable Duration timeout) {
             return new CompletableFuture<>();
         }
 

diff --git a/...st/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java b/...st/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeBuilder.java
@@ -41,7 +41,7 @@ public class DeclarativeSlotPoolBridgeBuilder {
 
     private JobID jobId = new JobID();
     private Duration batchSlotTimeout = JobManagerOptions.SLOT_IDLE_TIMEOUT.defaultValue();
-    private Duration idleSlotTimeout = TestingUtils.infiniteTime().toDuration();
+    private Duration idleSlotTimeout = TestingUtils.infiniteDuration();
     private Clock clock = SystemClock.getInstance();
     private Duration slotRequestMaxInterval = SLOT_REQUEST_MAX_INTERVAL.defaultValue();
     private ComponentMainThreadExecutor mainThreadExecutor = forMainThread();
@@ -101,7 +101,7 @@ public DeclarativeSlotPoolBridge build() {
                 jobId,
                 new DefaultDeclarativeSlotPoolFactory(),
                 clock,
-                TestingUtils.infiniteTime().toDuration(),
+                TestingUtils.infiniteDuration(),
                 idleSlotTimeout,
                 batchSlotTimeout,
                 requestSlotMatchingStrategy,
@@ -115,7 +115,7 @@ public DeclarativeSlotPoolBridge buildAndStart() throws Exception {
                         jobId,
                         new DefaultDeclarativeSlotPoolFactory(),
                         clock,
-                        TestingUtils.infiniteTime().toDuration(),
+                        TestingUtils.infiniteDuration(),
                         idleSlotTimeout,
                         batchSlotTimeout,
                         requestSlotMatchingStrategy,

diff --git a/...e/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java b/...e/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgePreferredAllocationsTest.java
@@ -52,9 +52,9 @@ void testDeclarativeSlotPoolTakesPreferredAllocationsIntoAccount() throws Except
                         new JobID(),
                         new DefaultDeclarativeSlotPoolFactory(),
                         SystemClock.getInstance(),
-                        TestingUtils.infiniteTime().toDuration(),
-                        TestingUtils.infiniteTime().toDuration(),
-                        TestingUtils.infiniteTime().toDuration(),
+                        TestingUtils.infiniteDuration(),
+                        TestingUtils.infiniteDuration(),
+                        TestingUtils.infiniteDuration(),
                         PreferredAllocationRequestSlotMatchingStrategy.INSTANCE,
                         Duration.ZERO,
                         forMainThread());

diff --git a/...ache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeRequestCompletionTest.java b/...ache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeRequestCompletionTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
@@ -32,6 +31,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -46,7 +46,7 @@
 /** Tests how the {@link DeclarativeSlotPoolBridge} completes slot requests. */
 class DeclarativeSlotPoolBridgeRequestCompletionTest {
 
-    private static final Time TIMEOUT = SlotPoolUtils.TIMEOUT;
+    private static final Duration TIMEOUT = SlotPoolUtils.TIMEOUT;
 
     private TestingResourceManagerGateway resourceManagerGateway;
 

diff --git a/...he/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeResourceDeclarationTest.java b/...he/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeResourceDeclarationTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -97,7 +96,7 @@ void testRequirementsIncreasedOnNewAllocation() throws Exception {
 
         // requesting the allocation of a new slot should increase the requirements
         declarativeSlotPoolBridge.requestNewAllocatedSlot(
-                new SlotRequestId(), ResourceProfile.UNKNOWN, Time.minutes(5));
+                new SlotRequestId(), ResourceProfile.UNKNOWN, Duration.ofMinutes(5));
         assertThat(requirementListener.getRequirements().getResourceCount(ResourceProfile.UNKNOWN))
                 .isOne();
     }
@@ -120,7 +119,7 @@ void testRequirementsDecreasedOnAllocationTimeout() throws Exception {
                                             declarativeSlotPoolBridge.requestNewAllocatedSlot(
                                                     new SlotRequestId(),
                                                     ResourceProfile.UNKNOWN,
-                                                    Time.milliseconds(5)),
+                                                    Duration.ofMillis(5)),
                                     mainThreadExecutor)
                             .get();
 

diff --git a/.../test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeTest.java b/.../test/java/org/apache/flink/runtime/jobmaster/slotpool/DeclarativeSlotPoolBridgeTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
@@ -95,7 +94,7 @@ void testNotEnoughResourcesAvailableFailsPendingRequests() throws Exception {
                                             declarativeSlotPoolBridge.requestNewAllocatedSlot(
                                                     slotRequestId,
                                                     ResourceProfile.UNKNOWN,
-                                                    Time.minutes(5)),
+                                                    Duration.ofMinutes(5)),
                                     componentMainThreadExecutor)
                             .get();
 
@@ -167,7 +166,7 @@ void testNoConcurrentModificationWhenSuspendingAndReleasingSlot() throws Excepti
                                                 declarativeSlotPoolBridge.requestNewAllocatedSlot(
                                                         slotRequestId,
                                                         ResourceProfile.UNKNOWN,
-                                                        Time.fromDuration(RPC_TIMEOUT));
+                                                        RPC_TIMEOUT);
                                         slotFuture.whenComplete(
                                                 (physicalSlot, throwable) -> {
                                                     if (throwable != null) {
@@ -199,9 +198,7 @@ void testAcceptingOfferedSlotsWithoutResourceManagerConnected() throws Exception
 
             final CompletableFuture<PhysicalSlot> slotFuture =
                     declarativeSlotPoolBridge.requestNewAllocatedSlot(
-                            new SlotRequestId(),
-                            ResourceProfile.UNKNOWN,
-                            Time.fromDuration(RPC_TIMEOUT));
+                            new SlotRequestId(), ResourceProfile.UNKNOWN, RPC_TIMEOUT);
 
             final LocalTaskManagerLocation localTaskManagerLocation =
                     new LocalTaskManagerLocation();

diff --git a/...e/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java b/...e/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.executiongraph.TestingComponentMainThreadExecutor;
 import org.apache.flink.runtime.jobmaster.SlotRequestId;
@@ -27,6 +26,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
@@ -36,7 +36,7 @@
 /** Tests for the {@link DeclarativeSlotPoolBridge} interactions. */
 class SlotPoolInteractionsTest {
 
-    private static final Time fastTimeout = Time.milliseconds(1L);
+    private static final Duration fastTimeout = Duration.ofMillis(1L);
 
     @RegisterExtension
     private static final TestingComponentMainThreadExecutor.Extension EXECUTOR_EXTENSION =

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolUtils.java
@@ -18,7 +18,6 @@
 
 package org.apache.flink.runtime.jobmaster.slotpool;
 
-import org.apache.flink.api.common.time.Time;
 import org.apache.flink.runtime.clusterframework.types.AllocationID;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -33,6 +32,7 @@
 import org.apache.flink.runtime.util.ResourceCounter;
 import org.apache.flink.util.FlinkException;
 
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
@@ -48,7 +48,7 @@
 /** Testing utility functions for the {@link SlotPool}. */
 public class SlotPoolUtils {
 
-    public static final Time TIMEOUT = Time.seconds(10L);
+    public static final Duration TIMEOUT = Duration.ofSeconds(10L);
 
     private SlotPoolUtils() {
         throw new UnsupportedOperationException("Cannot instantiate this class.");