reswqa commented on code in PR #19840:
URL: https://github.com/apache/flink/pull/19840#discussion_r899809302
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java:
##########
@@ -752,4 +753,76 @@ public void
testAllocationUpdatesIgnoredIfSlotMarkedAsAllocatedAfterSlotReport()
System.setSecurityManager(null);
}
+
+ @Test
+ public void testRequirementLongDelayOnlyTakeEffectOnce() throws Exception {
+ final CompletableFuture<Void> allocationIdFuture1 = new
CompletableFuture<>();
+ final CompletableFuture<Void> allocationIdFuture2 = new
CompletableFuture<>();
+ final TaskExecutorGateway taskExecutorGateway1 =
+ new TestingTaskExecutorGatewayBuilder()
+ .setRequestSlotFunction(
+ tuple6 -> {
+ allocationIdFuture1.complete(null);
+ return
CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .createTestingTaskExecutorGateway();
+ final TaskExecutorGateway taskExecutorGateway2 =
+ new TestingTaskExecutorGatewayBuilder()
+ .setRequestSlotFunction(
+ tuple6 -> {
+ allocationIdFuture2.complete(null);
+ return
CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .createTestingTaskExecutorGateway();
+ final TaskExecutorConnection taskExecutorConnection1 =
+ new TaskExecutorConnection(ResourceID.generate(),
taskExecutorGateway1);
+ final TaskExecutorConnection taskExecutorConnection2 =
+ new TaskExecutorConnection(ResourceID.generate(),
taskExecutorGateway2);
+ new Context() {
+ {
+ runTest(
+ () -> {
+ final CompletableFuture<Boolean>
registerTaskManagerFuture1 =
+ new CompletableFuture<>();
+ final CompletableFuture<Boolean>
registerTaskManagerFuture2 =
+ new CompletableFuture<>();
+ runInMainThread(
+ () -> {
+
getSlotManager().enlargeRequirementsCheckDelayOnce();
+ getSlotManager()
+ .processResourceRequirements(
+
createResourceRequirements(
+ new JobID(),
+ 2,
+
DEFAULT_SLOT_RESOURCE_PROFILE));
+ registerTaskManagerFuture1.complete(
+ getSlotManager()
+ .registerTaskManager(
+
taskExecutorConnection1,
+ new
SlotReport(),
+
DEFAULT_SLOT_RESOURCE_PROFILE,
+
DEFAULT_SLOT_RESOURCE_PROFILE));
+ });
+ assertThat(
+
assertFutureCompleteAndReturn(registerTaskManagerFuture1),
+ is(true));
+ assertFutureNotComplete(allocationIdFuture1);
+ allocationIdFuture1.get(200,
TimeUnit.MILLISECONDS);
Review Comment:
Just a small suggestion: for newly introduced tests we should use assertj
then the code will look like this
```Assertions.assertThat(allocationIdFuture1).succeedsWithin(FUTURE_TIMEOUT_SECOND,
TimeUnit.SECONDS)```
Of course, it would be better if the entire test class could be migrated to
junit5 by the way. The entire runtime is a very large module. Migrating to
junit5 & assertj requires a lot of work. If the new test does not do this, it
will be more difficult to migrate later.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
