zhuzhurk commented on a change in pull request #12917:
URL: https://github.com/apache/flink/pull/12917#discussion_r462712808
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTestBase.java
##########
@@ -0,0 +1,74 @@
+package org.apache.flink.runtime.jobmaster.slotpool;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
+import org.apache.flink.runtime.jobmaster.SlotRequestId;
+import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
+import
org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Before;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Test base for {@link SlotPool} related test cases.
+ */
+public abstract class SlotPoolTestBase extends TestLogger {
+ protected static final Time TIMEOUT = Time.seconds(10L);
+
+ protected final ComponentMainThreadExecutor mainThreadExecutor =
+ ComponentMainThreadExecutorServiceAdapter.forMainThread();
+
+ protected TestingResourceManagerGateway resourceManagerGateway;
+ protected SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setup() throws Exception {
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(mainThreadExecutor).setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ protected TestingSlotPoolImpl createAndSetUpSlotPool() throws Exception
{
Review comment:
looks to me all the `createAndSetUpSlotPool(..)` methods can be package
private?
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImplTest.java
##########
@@ -224,48 +182,27 @@ public void testAllocateWithFreeSlot() throws Exception {
final CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
- try (SlotPoolImpl slotPool = createSlotPoolImpl()) {
- setupSlotPool(slotPool, resourceManagerGateway,
mainThreadExecutor);
- Scheduler scheduler = setupScheduler(slotPool,
mainThreadExecutor);
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
slotPool.registerTaskManager(taskManagerLocation.getResourceID());
- CompletableFuture<LogicalSlot> future1 =
scheduler.allocateSlot(
- new SlotRequestId(),
- new DummyScheduledUnit(),
- SlotProfile.noLocality(DEFAULT_TESTING_PROFILE),
- timeout);
- assertFalse(future1.isDone());
-
- final SlotRequest slotRequest =
slotRequestFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
-
+ AllocationID allocationID = new AllocationID();
Review comment:
nit: `allocationId`
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImplTest.java
##########
@@ -591,17 +501,19 @@ public void testDiscardIdleSlotIfReleasingFailed() throws
Exception {
assertThat(slotPool.offerSlot(taskManagerLocation,
taskManagerGateway, slotToExpire), Matchers.is(true));
- clock.advanceTime(timeout.toMilliseconds() + 1,
TimeUnit.MILLISECONDS);
+ clock.advanceTime(TIMEOUT.toMilliseconds() + 1,
TimeUnit.MILLISECONDS);
slotPool.triggerCheckIdleSlot();
freeSlotLatch.await();
- CompletableFuture<LogicalSlot> allocatedSlotFuture =
allocateSlot(scheduler, new SlotRequestId());
+ final CompletableFuture<PhysicalSlot> future =
requestNewAllocatedSlot(
Review comment:
nit: `allocatedSlotFuture`
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImplTest.java
##########
@@ -676,18 +586,18 @@ public void testFreeFailedSlots() throws Exception {
*/
@Test
public void testCreateAllocatedSlotReport() throws Exception {
+ final JobID jobId = new JobID();
- try (SlotPoolImpl slotPool = createSlotPoolImpl()) {
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool(jobId)) {
final ArrayBlockingQueue<AllocationID> allocationIds =
new ArrayBlockingQueue<>(1);
resourceManagerGateway.setRequestSlotConsumer(
- slotRequest ->
allocationIds.offer(slotRequest.getAllocationId()));
-
- setupSlotPool(slotPool, resourceManagerGateway,
mainThreadExecutor);
- Scheduler scheduler = setupScheduler(slotPool,
mainThreadExecutor);
+ slotRequest ->
allocationIds.offer(slotRequest.getAllocationId()));
- final SlotRequestId slotRequestId = new SlotRequestId();
- final CompletableFuture<LogicalSlot> slotRequestFuture
= allocateSlot(scheduler, slotRequestId);
+ final CompletableFuture<PhysicalSlot> slotRequestFuture
= requestNewAllocatedSlot(
+ slotPool,
+ new SlotRequestId()
+ );
Review comment:
this right bracket should be at the end of the previous line.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBuilder.java
##########
@@ -35,9 +36,10 @@
*/
public class SlotPoolBuilder {
- private ComponentMainThreadExecutor componentMainThreadExecutor;
- private ResourceManagerGateway resourceManagerGateway = new
TestingResourceManagerGateway();
- private Time batchSlotTimeout = Time.milliseconds(2L);
+ private final ComponentMainThreadExecutor componentMainThreadExecutor;
+ private ResourceManagerGateway resourceManagerGateway;
+ private Time batchSlotTimeout =
Time.milliseconds(JobManagerOptions.SLOT_IDLE_TIMEOUT.defaultValue());
Review comment:
I see. Thanks for the explanation.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImplTest.java
##########
@@ -776,7 +686,9 @@ public void testOrphanedAllocationIsCanceledIfNotRemapped()
throws Exception {
final List<AllocationID> canceledAllocations = new
ArrayList<>();
resourceManagerGateway.setCancelSlotConsumer(canceledAllocations::add);
- try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ try (SlotPoolImpl slotPool =
createAndSetUpSlotPool(resourceManagerGateway)) {
+
slotPool.registerTaskManager(taskManagerLocation.getResourceID());
Review comment:
`registerTaskManager` is not need because the following
`registerAndOfferSlot` will do it.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImplTest.java
##########
@@ -87,56 +77,38 @@
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
/**
* Tests for the {@link SlotPoolImpl}.
*/
-public class SlotPoolImplTest extends TestLogger {
-
- private final Time timeout = Time.seconds(10L);
-
- private JobID jobId;
+public class SlotPoolImplTest extends SlotPoolTestBase {
private TaskManagerLocation taskManagerLocation;
-
private SimpleAckingTaskManagerGateway taskManagerGateway;
- private TestingResourceManagerGateway resourceManagerGateway;
-
- private ComponentMainThreadExecutor mainThreadExecutor =
- ComponentMainThreadExecutorServiceAdapter.forMainThread();
-
@Before
- public void setUp() throws Exception {
- this.jobId = new JobID();
-
+ @Override
+ public void setup() throws Exception {
taskManagerLocation = new LocalTaskManagerLocation();
taskManagerGateway = new SimpleAckingTaskManagerGateway();
- resourceManagerGateway = new TestingResourceManagerGateway();
+ super.setup();
Review comment:
It's better to invoke `super.setup();` first in case it overrides any
actions in the sub-class.
##########
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:
I guess this change is unintentional?
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/TestingSlotPoolImpl.java
##########
@@ -39,6 +43,10 @@
private ResourceProfile lastRequestedSlotResourceProfile;
+ private volatile Consumer<SlotRequestId> releaseSlotConsumer;
Review comment:
Could we have a separate commit to merge
`SlotPoolInteractionsTest#TestingSlotPool` with `TestingSlotPoolImpl`?
It can be easier to reasoning which methods are ported instead of being
newly added.
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
Review comment:
Could we add this test in a separate commit and elaborate in the commit
message for the purpose of it?
##########
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:
the method `slotPoolBuilder.build(connectToResourceManager)` no longer
exists
##########
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:
can we use `SlotPoolBuilder` to do this by setting
`resourceManagerGateway` to `null`?
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
+
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
+
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ testMainThreadExecutor.execute(() ->
slotPool.registerTaskManager(taskManagerLocation.getResourceID()));
+
+ Scheduler scheduler = new
SchedulerImpl(LocationPreferenceSlotSelectionStrategy.createDefault(),
slotPool);
Review comment:
can be final. also applies some other variables below.
##########
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:
timeout -> TIMEOUT
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
+
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
+
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ testMainThreadExecutor.execute(() ->
slotPool.registerTaskManager(taskManagerLocation.getResourceID()));
+
+ Scheduler scheduler = new
SchedulerImpl(LocationPreferenceSlotSelectionStrategy.createDefault(),
slotPool);
+
scheduler.start(testMainThreadExecutor.getMainThreadExecutor());
+
+ SlotRequestId requestId = new SlotRequestId();
+ CompletableFuture<LogicalSlot> future =
testMainThreadExecutor.execute(() -> scheduler.allocateSlot(
+ requestId,
+ new DummyScheduledUnit(),
+ SlotProfile.noRequirements(),
+ timeout));
+ assertFalse(future.isDone());
+
+ final SlotRequest slotRequest =
slotRequestFuture.get(timeout.toMilliseconds(), TimeUnit.MILLISECONDS);
+
+ final SlotOffer slotOffer = new SlotOffer(
+ slotRequest.getAllocationId(),
+ 0,
+ DEFAULT_TESTING_PROFILE);
+
+ assertTrue(testMainThreadExecutor.execute(() ->
slotPool.offerSlot(
Review comment:
There is no need to verify this offer. IIRC, it is already covered in
`SlotPoolImplTest`.
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
+
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
+
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ testMainThreadExecutor.execute(() ->
slotPool.registerTaskManager(taskManagerLocation.getResourceID()));
Review comment:
I'd prefer to do it right before offerSlot
##########
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:
indentation or an empty line is needed.
##########
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:
indentation or an empty line is needed
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
+
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
+
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ testMainThreadExecutor.execute(() ->
slotPool.registerTaskManager(taskManagerLocation.getResourceID()));
+
+ Scheduler scheduler = new
SchedulerImpl(LocationPreferenceSlotSelectionStrategy.createDefault(),
slotPool);
+
scheduler.start(testMainThreadExecutor.getMainThreadExecutor());
+
+ SlotRequestId requestId = new SlotRequestId();
+ CompletableFuture<LogicalSlot> future =
testMainThreadExecutor.execute(() -> scheduler.allocateSlot(
+ requestId,
Review comment:
looks to me `requestId` is not further used.
This allocation code is used in both test cases. Could we have an
`allocateSlot()` method?
##########
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);
+
+ @ClassRule
+ public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
+ new TestingComponentMainThreadExecutor.Resource(10L);
+
+ private final TestingComponentMainThreadExecutor testMainThreadExecutor
=
+ EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+
+ private TaskManagerLocation taskManagerLocation;
+ private SimpleAckingTaskManagerGateway taskManagerGateway;
+ private TestingResourceManagerGateway resourceManagerGateway;
+ private SlotPoolBuilder slotPoolBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ taskManagerLocation = new LocalTaskManagerLocation();
+ taskManagerGateway = new SimpleAckingTaskManagerGateway();
+ resourceManagerGateway = new TestingResourceManagerGateway();
+ slotPoolBuilder = new
SlotPoolBuilder(testMainThreadExecutor.getMainThreadExecutor())
+ .setResourceManagerGateway(resourceManagerGateway);
+ }
+
+ @Test
+ public void testAllocateSlot() throws Exception {
+ CompletableFuture<SlotRequest> slotRequestFuture = new
CompletableFuture<>();
+
resourceManagerGateway.setRequestSlotConsumer(slotRequestFuture::complete);
+
+ try (SlotPoolImpl slotPool = createAndSetUpSlotPool()) {
+ testMainThreadExecutor.execute(() ->
slotPool.registerTaskManager(taskManagerLocation.getResourceID()));
+
+ Scheduler scheduler = new
SchedulerImpl(LocationPreferenceSlotSelectionStrategy.createDefault(),
slotPool);
+
scheduler.start(testMainThreadExecutor.getMainThreadExecutor());
Review comment:
could we have a `createAndSetupScheduler()` to deduplicate code?
----------------------------------------------------------------
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:
us...@infra.apache.org
