Thesharing commented on a change in pull request #12917:
URL: https://github.com/apache/flink/pull/12917#discussion_r462825355
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolImpl.java
##########
@@ -83,4 +93,52 @@ boolean isBatchSlotRequestTimeoutCheckEnabled() {
public ResourceProfile getLastRequestedSlotResourceProfile() {
return lastRequestedSlotResourceProfile;
}
+
+ public void setReleaseSlotConsumer(Consumer<SlotRequestId>
releaseSlotConsumer) {
+ this.releaseSlotConsumer =
Preconditions.checkNotNull(releaseSlotConsumer);
+ }
+
+ public void
setTimeoutPendingSlotRequestConsumer(Consumer<SlotRequestId>
timeoutPendingSlotRequestConsumer) {
+ this.timeoutPendingSlotRequestConsumer =
Preconditions.checkNotNull(timeoutPendingSlotRequestConsumer);
+ }
+
+ @Override
+ public void releaseSlot(
+ @Nonnull SlotRequestId slotRequestId,
Review comment:
I forget this. I have removed this.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
##########
@@ -253,118 +191,7 @@ public void testExtraSlotsAreKept() throws Exception {
}
}
- /**
- * This case make sure when allocateSlot in ProviderAndOwner timeout,
- * it will automatically call cancelSlotAllocation as will inject
future.whenComplete in ProviderAndOwner.
- */
- @Test
- public void testProviderAndOwnerSlotAllocationTimeout() throws
Exception {
- final JobID jid = new JobID();
-
- try (TestingSlotPool pool = createTestingSlotPool(jid)) {
-
- final CompletableFuture<SlotRequestId>
releaseSlotFuture = new CompletableFuture<>();
-
-
pool.setReleaseSlotConsumer(releaseSlotFuture::complete);
-
- pool.start(JobMasterId.generate(), "foobar",
testMainThreadExecutor.getMainThreadExecutor());
- ResourceManagerGateway resourceManagerGateway = new
TestingResourceManagerGateway();
- pool.connectToResourceManager(resourceManagerGateway);
-
- Scheduler scheduler = new
SchedulerImpl(LocationPreferenceSlotSelectionStrategy.createDefault(), pool);
-
scheduler.start(testMainThreadExecutor.getMainThreadExecutor());
-
- // test the pending request is clear when timed out
- CompletableFuture<LogicalSlot> future =
testMainThreadExecutor.execute(() -> scheduler.allocateSlot(
- new DummyScheduledUnit(),
- SlotProfile.noRequirements(),
- fastTimeout));
- try {
- future.get();
- fail("We expected a TimeoutException.");
- } catch (ExecutionException e) {
-
assertTrue(ExceptionUtils.stripExecutionException(e) instanceof
TimeoutException);
- }
-
- // wait for the cancel call on the SlotPoolImpl
- releaseSlotFuture.get();
-
- assertEquals(0L, pool.getNumberOfPendingRequests());
- }
- }
-
- /**
- * Testing SlotPoolImpl which exposes internal state via some testing
methods.
- */
- private static final class TestingSlotPool extends SlotPoolImpl {
-
- private volatile Consumer<SlotRequestId> releaseSlotConsumer;
-
- private volatile Consumer<SlotRequestId>
timeoutPendingSlotRequestConsumer;
-
- public TestingSlotPool(
- JobID jobId,
- Clock clock,
- Time rpcTimeout,
- Time idleSlotTimeout,
- Time batchSlotTimeout) {
- super(
- jobId,
- clock,
- rpcTimeout,
- idleSlotTimeout,
- batchSlotTimeout);
-
- releaseSlotConsumer = null;
- timeoutPendingSlotRequestConsumer = null;
- }
-
- public void setReleaseSlotConsumer(Consumer<SlotRequestId>
releaseSlotConsumer) {
- this.releaseSlotConsumer =
Preconditions.checkNotNull(releaseSlotConsumer);
- }
-
- public void
setTimeoutPendingSlotRequestConsumer(Consumer<SlotRequestId>
timeoutPendingSlotRequestConsumer) {
- this.timeoutPendingSlotRequestConsumer =
Preconditions.checkNotNull(timeoutPendingSlotRequestConsumer);
- }
-
- @Override
- public void releaseSlot(
- @Nonnull SlotRequestId slotRequestId,
- @Nullable Throwable cause) {
- final Consumer<SlotRequestId>
currentReleaseSlotConsumer = releaseSlotConsumer;
-
- super.releaseSlot(slotRequestId, cause);
-
- if (currentReleaseSlotConsumer != null) {
-
currentReleaseSlotConsumer.accept(slotRequestId);
- }
- }
-
- @Override
- protected void timeoutPendingSlotRequest(SlotRequestId
slotRequestId) {
- final Consumer<SlotRequestId>
currentTimeoutPendingSlotRequestConsumer = timeoutPendingSlotRequestConsumer;
-
- if (currentTimeoutPendingSlotRequestConsumer != null) {
-
currentTimeoutPendingSlotRequestConsumer.accept(slotRequestId);
- }
-
- super.timeoutPendingSlotRequest(slotRequestId);
- }
-
- boolean containsAllocatedSlot(AllocationID allocationId) {
- return getAllocatedSlots().contains(allocationId);
- }
-
- boolean containsAvailableSlot(AllocationID allocationId) {
- return getAvailableSlots().contains(allocationId);
- }
-
- int getNumberOfPendingRequests() {
- return getPendingRequests().size();
- }
-
- int getNumberOfWaitingForResourceRequests() {
- return getWaitingForResourceManager().size();
- }
+ private TestingSlotPoolImpl createAndSetUpSlotPool(Boolean
connectToResourceManager) throws Exception {
+ return slotPoolBuilder.build(connectToResourceManager);
Review comment:
Yes. I fix this up in a new commit.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolImpl.java
##########
@@ -71,9 +81,9 @@ boolean isBatchSlotRequestTimeoutCheckEnabled() {
@Override
public CompletableFuture<PhysicalSlot> requestNewAllocatedSlot(
- final SlotRequestId slotRequestId,
- final ResourceProfile resourceProfile,
- @Nullable final Time timeout) {
+ final SlotRequestId slotRequestId,
+ final ResourceProfile resourceProfile,
+ @Nullable final Time timeout) {
Review comment:
Yes. Undone this change in a new commit.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolUtils.java
##########
@@ -58,6 +58,16 @@ private SlotPoolUtils() {
.thenCompose(Function.identity());
}
+ public static List<CompletableFuture<PhysicalSlot>>
requestNewAllocatedBatchSlots(
+ SlotPool slotPool,
+ ComponentMainThreadExecutor mainThreadExecutor,
+ List<ResourceProfile> resourceProfiles) {
Review comment:
Done.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBatchSlotRequestTest.java
##########
@@ -266,7 +269,10 @@ public void
testPendingBatchSlotRequestTimeoutAfterSlotRelease() throws Exceptio
}
}
- private void
advanceTimeAndTriggerCheckBatchSlotTimeout(TestingSlotPoolImpl slotPool,
ManualClock clock, Time batchSlotTimeout) {
+ private void advanceTimeAndTriggerCheckBatchSlotTimeout(
+ TestingSlotPoolImpl slotPool,
+ ManualClock clock,
+ Time batchSlotTimeout) {
Review comment:
Done.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SchedulerImplTest.java
##########
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.jobmaster.slotpool;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.SlotProfile;
+import
org.apache.flink.runtime.executiongraph.TestingComponentMainThreadExecutor;
+import
org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
+import org.apache.flink.runtime.jobmanager.scheduler.DummyScheduledUnit;
+import org.apache.flink.runtime.jobmaster.LogicalSlot;
+import org.apache.flink.runtime.jobmaster.SlotRequestId;
+import org.apache.flink.runtime.resourcemanager.SlotRequest;
+import
org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
+import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
+import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static
org.apache.flink.runtime.jobmaster.slotpool.AvailableSlotsTest.DEFAULT_TESTING_PROFILE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * Tests for the {@link SchedulerImpl}.
+ */
+public class SchedulerImplTest extends TestLogger {
+
+ private static final Time timeout = Time.seconds(1L);
Review comment:
Done.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolRequestCompletionTest.java
##########
@@ -129,22 +115,13 @@ private void runSlotRequestCompletionTest(
}
}
- private SlotPoolImpl setUpSlotPoolAndConnectToResourceManager() throws
Exception {
- final SlotPoolImpl slotPool = setUpSlotPool();
- connectToResourceManager(slotPool);
-
- return slotPool;
- }
-
private void connectToResourceManager(SlotPoolImpl slotPool) {
slotPool.connectToResourceManager(resourceManagerGateway);
}
- private SlotPoolImpl setUpSlotPool() throws Exception {
- final SlotPoolImpl slotPool = new TestingSlotPoolImpl(new
JobID());
-
- slotPool.start(JobMasterId.generate(), "foobar",
ComponentMainThreadExecutorServiceAdapter.forMainThread());
-
+ private TestingSlotPoolImpl
createAndSetUpSlotPoolWithoutResourceManager() throws Exception {
+ final TestingSlotPoolImpl slotPool = new
TestingSlotPoolImpl(new JobID());
Review comment:
Agreed and done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
