This is an automated email from the ASF dual-hosted git repository.
jiabaosun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new beb0b167bdc [FLINK-32843][JUnit5 Migration] Migrate the jobmaster
package of flink-runtime module to JUnit5
beb0b167bdc is described below
commit beb0b167bdcf95f27be87a214a69a174fd49d256
Author: Yuepeng Pan <[email protected]>
AuthorDate: Mon May 6 09:13:54 2024 +0800
[FLINK-32843][JUnit5 Migration] Migrate the jobmaster package of
flink-runtime module to JUnit5
---
.../TestingComponentMainThreadExecutor.java | 36 -------
.../DefaultExecutionDeploymentReconcilerTest.java | 39 ++++---
.../DefaultExecutionDeploymentTrackerTest.java | 46 ++++-----
.../DefaultJobMasterServiceProcessTest.java | 4 +-
...DefaultSlotPoolServiceSchedulerFactoryTest.java | 10 +-
.../runtime/jobmaster/JobExecutionITCase.java | 12 +--
.../jobmaster/JobIntermediateDatasetReuseTest.java | 14 ++-
.../jobmaster/JobManagerRunnerResultTest.java | 38 ++++---
.../jobmaster/JobManagerSharedServicesTest.java | 25 +++--
...asterExecutionDeploymentReconciliationTest.java | 77 +++++++-------
.../jobmaster/JobMasterPartitionReleaseTest.java | 72 ++++++-------
.../jobmaster/JobMasterQueryableStateTest.java | 113 ++++++++++-----------
.../runtime/jobmaster/JobMasterSchedulerTest.java | 33 +++---
.../flink/runtime/jobmaster/JobRecoveryITCase.java | 26 +++--
.../flink/runtime/jobmaster/JobResultTest.java | 84 +++++++--------
.../AbstractServiceConnectionManagerTest.java | 31 +++---
.../slotpool/AllocatedSlotOccupationTest.java | 17 ++--
...rceRequirementServiceConnectionManagerTest.java | 39 ++++---
...rce.java => PhysicalSlotProviderExtension.java} | 14 +--
...erImplWithDefaultSlotSelectionStrategyTest.java | 46 ++++-----
...lSlotProviderImplWithSpreadOutStrategyTest.java | 73 +++++++------
.../PhysicalSlotRequestBulkCheckerImplTest.java | 100 +++++++++---------
.../PhysicalSlotRequestBulkWithTimestampTest.java | 16 ++-
...dAllocationRequestSlotMatchingStrategyTest.java | 4 +-
.../jobmaster/slotpool/SingleLogicalSlotTest.java | 63 ++++++------
.../slotpool/SlotPoolBatchSlotRequestTest.java | 67 ++++++------
.../slotpool/SlotPoolInteractionsTest.java | 55 +++++-----
.../runtime/jobmaster/slotpool/SlotPoolUtils.java | 5 +-
28 files changed, 520 insertions(+), 639 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
index c767fc14d1b..bc88c289da4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TestingComponentMainThreadExecutor.java
@@ -28,7 +28,6 @@ import org.apache.flink.util.function.ThrowingRunnable;
import org.junit.jupiter.api.extension.AfterAllCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
-import org.junit.rules.ExternalResource;
import javax.annotation.Nonnull;
@@ -72,41 +71,6 @@ public class TestingComponentMainThreadExecutor {
return mainThreadExecutor;
}
- /** Test resource for convenience. */
- public static class Resource extends ExternalResource {
-
- private long shutdownTimeoutMillis;
- private TestingComponentMainThreadExecutor
componentMainThreadTestExecutor;
- private ScheduledExecutorService innerExecutorService;
-
- public Resource() {
- this(500L);
- }
-
- public Resource(long shutdownTimeoutMillis) {
- this.shutdownTimeoutMillis = shutdownTimeoutMillis;
- }
-
- @Override
- protected void before() {
- this.innerExecutorService =
Executors.newSingleThreadScheduledExecutor();
- this.componentMainThreadTestExecutor =
- new TestingComponentMainThreadExecutor(
-
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
- innerExecutorService));
- }
-
- @Override
- protected void after() {
- ExecutorUtils.gracefulShutdown(
- shutdownTimeoutMillis, TimeUnit.MILLISECONDS,
innerExecutorService);
- }
-
- public TestingComponentMainThreadExecutor
getComponentMainThreadTestExecutor() {
- return componentMainThreadTestExecutor;
- }
- }
-
/** Test extension for convenience. */
public static class Extension implements BeforeAllCallback,
AfterAllCallback {
private final long shutdownTimeoutMillis;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentReconcilerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentReconcilerTest.java
index e9fbe3784d9..390056e4a5d 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentReconcilerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentReconcilerTest.java
@@ -20,9 +20,8 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.taskexecutor.ExecutionDeploymentReport;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
@@ -34,15 +33,13 @@ import java.util.stream.Stream;
import static
org.apache.flink.runtime.clusterframework.types.ResourceID.generate;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.core.IsCollectionContaining.hasItem;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link DefaultExecutionDeploymentReconciler}. */
-public class DefaultExecutionDeploymentReconcilerTest extends TestLogger {
+class DefaultExecutionDeploymentReconcilerTest {
@Test
- public void testMatchingDeployments() {
+ void testMatchingDeployments() {
TestingExecutionDeploymentReconciliationHandler handler =
new TestingExecutionDeploymentReconciliationHandler();
@@ -57,12 +54,12 @@ public class DefaultExecutionDeploymentReconcilerTest
extends TestLogger {
new
ExecutionDeploymentReport(Collections.singleton(attemptId)),
Collections.singletonMap(attemptId,
ExecutionDeploymentState.DEPLOYED));
- assertThat(handler.getMissingExecutions(), empty());
- assertThat(handler.getUnknownExecutions(), empty());
+ assertThat(handler.getMissingExecutions()).isEmpty();
+ assertThat(handler.getUnknownExecutions()).isEmpty();
}
@Test
- public void testMissingDeployments() {
+ void testMissingDeployments() {
TestingExecutionDeploymentReconciliationHandler handler =
new TestingExecutionDeploymentReconciliationHandler();
@@ -77,12 +74,12 @@ public class DefaultExecutionDeploymentReconcilerTest
extends TestLogger {
new ExecutionDeploymentReport(Collections.emptySet()),
Collections.singletonMap(attemptId,
ExecutionDeploymentState.DEPLOYED));
- assertThat(handler.getUnknownExecutions(), empty());
- assertThat(handler.getMissingExecutions(), hasItem(attemptId));
+ assertThat(handler.getUnknownExecutions()).isEmpty();
+ assertThat(handler.getMissingExecutions()).contains(attemptId);
}
@Test
- public void testUnknownDeployments() {
+ void testUnknownDeployments() {
TestingExecutionDeploymentReconciliationHandler handler =
new TestingExecutionDeploymentReconciliationHandler();
@@ -97,12 +94,12 @@ public class DefaultExecutionDeploymentReconcilerTest
extends TestLogger {
new
ExecutionDeploymentReport(Collections.singleton(attemptId)),
Collections.emptyMap());
- assertThat(handler.getMissingExecutions(), empty());
- assertThat(handler.getUnknownExecutions(), hasItem(attemptId));
+ assertThat(handler.getMissingExecutions()).isEmpty();
+ assertThat(handler.getUnknownExecutions()).contains(attemptId);
}
@Test
- public void testMissingAndUnknownDeployments() {
+ void testMissingAndUnknownDeployments() {
TestingExecutionDeploymentReconciliationHandler handler =
new TestingExecutionDeploymentReconciliationHandler();
@@ -120,12 +117,12 @@ public class DefaultExecutionDeploymentReconcilerTest
extends TestLogger {
Stream.of(missingId, matchingId)
.collect(Collectors.toMap(x -> x, x ->
ExecutionDeploymentState.DEPLOYED)));
- assertThat(handler.getMissingExecutions(), hasItem(missingId));
- assertThat(handler.getUnknownExecutions(), hasItem(unknownId));
+ assertThat(handler.getMissingExecutions()).contains(missingId);
+ assertThat(handler.getUnknownExecutions()).contains(unknownId);
}
@Test
- public void testPendingDeployments() {
+ void testPendingDeployments() {
TestingExecutionDeploymentReconciliationHandler handler =
new TestingExecutionDeploymentReconciliationHandler();
@@ -143,8 +140,8 @@ public class DefaultExecutionDeploymentReconcilerTest
extends TestLogger {
Stream.of(matchingId, missingId)
.collect(Collectors.toMap(x -> x, x ->
ExecutionDeploymentState.PENDING)));
- assertThat(handler.getMissingExecutions(), empty());
- assertThat(handler.getUnknownExecutions(), hasItem(unknownId));
+ assertThat(handler.getMissingExecutions()).isEmpty();
+ assertThat(handler.getUnknownExecutions()).contains(unknownId);
}
private static class TestingExecutionDeploymentReconciliationHandler
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentTrackerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentTrackerTest.java
index 98691d0769b..528be05a848 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentTrackerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultExecutionDeploymentTrackerTest.java
@@ -19,42 +19,34 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.hamcrest.Matchers.allOf;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.hasEntry;
-import static org.hamcrest.Matchers.hasKey;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link DefaultExecutionDeploymentTracker}. */
-public class DefaultExecutionDeploymentTrackerTest extends TestLogger {
+class DefaultExecutionDeploymentTrackerTest {
@Test
- public void testStartTracking() {
+ void testStartTracking() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
final ExecutionAttemptID attemptId1 = createExecutionAttemptId();
final ResourceID resourceId1 = ResourceID.generate();
tracker.startTrackingPendingDeploymentOf(attemptId1, resourceId1);
- assertThat(
- tracker.getExecutionsOn(resourceId1),
- hasEntry(attemptId1, ExecutionDeploymentState.PENDING));
+ assertThat(tracker.getExecutionsOn(resourceId1))
+ .containsEntry(attemptId1, ExecutionDeploymentState.PENDING);
tracker.completeDeploymentOf(attemptId1);
- assertThat(
- tracker.getExecutionsOn(resourceId1),
- hasEntry(attemptId1, ExecutionDeploymentState.DEPLOYED));
+ assertThat(tracker.getExecutionsOn(resourceId1))
+ .containsEntry(attemptId1, ExecutionDeploymentState.DEPLOYED);
}
@Test
- public void testStopTrackingCompletedDeployment() {
+ void testStopTrackingCompletedDeployment() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
final ExecutionAttemptID attemptId1 = createExecutionAttemptId();
@@ -65,11 +57,11 @@ public class DefaultExecutionDeploymentTrackerTest extends
TestLogger {
tracker.stopTrackingDeploymentOf(attemptId1);
- assertThat(tracker.getExecutionsOn(resourceId1).entrySet(), empty());
+ assertThat(tracker.getExecutionsOn(resourceId1).entrySet()).isEmpty();
}
@Test
- public void testStopTrackingPendingDeployment() {
+ void testStopTrackingPendingDeployment() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
final ExecutionAttemptID attemptId1 = createExecutionAttemptId();
@@ -78,11 +70,11 @@ public class DefaultExecutionDeploymentTrackerTest extends
TestLogger {
tracker.stopTrackingDeploymentOf(attemptId1);
- assertThat(tracker.getExecutionsOn(resourceId1).entrySet(), empty());
+ assertThat(tracker.getExecutionsOn(resourceId1).entrySet()).isEmpty();
}
@Test
- public void testStopTrackingDoesNotAffectOtherIds() {
+ void testStopTrackingDoesNotAffectOtherIds() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
final ExecutionAttemptID attemptId1 = createExecutionAttemptId();
@@ -92,18 +84,18 @@ public class DefaultExecutionDeploymentTrackerTest extends
TestLogger {
tracker.stopTrackingDeploymentOf(createExecutionAttemptId());
- assertThat(tracker.getExecutionsOn(resourceId1), hasKey(attemptId1));
+
assertThat(tracker.getExecutionsOn(resourceId1)).containsKey(attemptId1);
}
@Test
- public void testCompleteDeploymentUnknownExecutionDoesNotThrowException() {
+ void testCompleteDeploymentUnknownExecutionDoesNotThrowException() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
tracker.completeDeploymentOf(createExecutionAttemptId());
}
@Test
- public void testStopTrackingUnknownExecutionDoesNotThrowException() {
+ void testStopTrackingUnknownExecutionDoesNotThrowException() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
final ExecutionAttemptID attemptId2 = createExecutionAttemptId();
@@ -111,11 +103,9 @@ public class DefaultExecutionDeploymentTrackerTest extends
TestLogger {
}
@Test
- public void testGetExecutionsReturnsEmptySetForUnknownHost() {
+ void testGetExecutionsReturnsEmptySetForUnknownHost() {
final DefaultExecutionDeploymentTracker tracker = new
DefaultExecutionDeploymentTracker();
- assertThat(
- tracker.getExecutionsOn(ResourceID.generate()).entrySet(),
- allOf(notNullValue(), empty()));
+
assertThat(tracker.getExecutionsOn(ResourceID.generate()).entrySet()).isEmpty();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java
index 9d1addca44c..fbfffd6e6d9 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultJobMasterServiceProcessTest.java
@@ -205,7 +205,7 @@ class DefaultJobMasterServiceProcessTest {
}
@Test
- void testLeaderAddressGetsForwarded() throws Exception {
+ void testLeaderAddressGetsForwarded() {
final CompletableFuture<JobMasterService> jobMasterServiceFuture =
new CompletableFuture<>();
DefaultJobMasterServiceProcess serviceProcess =
createTestInstance(jobMasterServiceFuture);
@@ -249,7 +249,7 @@ class DefaultJobMasterServiceProcessTest {
}
@Test
- void testSuccessOnTerminalState() throws Exception {
+ void testSuccessOnTerminalState() {
final CompletableFuture<JobMasterService> jobMasterServiceFuture =
new CompletableFuture<>();
DefaultJobMasterServiceProcess serviceProcess =
createTestInstance(jobMasterServiceFuture);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
index bdc42e2d5cb..fc651225bb6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/DefaultSlotPoolServiceSchedulerFactoryTest.java
@@ -43,10 +43,10 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link DefaultSlotPoolServiceSchedulerFactory}. */
@ExtendWith(TestLoggerExtension.class)
-public class DefaultSlotPoolServiceSchedulerFactoryTest {
+class DefaultSlotPoolServiceSchedulerFactoryTest {
@Test
- public void testFallsBackToDefaultSchedulerIfAdaptiveSchedulerInBatchJob()
{
+ void testFallsBackToDefaultSchedulerIfAdaptiveSchedulerInBatchJob() {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.SCHEDULER,
JobManagerOptions.SchedulerType.Adaptive);
@@ -61,7 +61,7 @@ public class DefaultSlotPoolServiceSchedulerFactoryTest {
}
@Test
- public void testAdaptiveSchedulerForReactiveMode() {
+ void testAdaptiveSchedulerForReactiveMode() {
final Configuration configuration = new Configuration();
configuration.set(JobManagerOptions.SCHEDULER_MODE,
SchedulerExecutionMode.REACTIVE);
@@ -76,7 +76,7 @@ public class DefaultSlotPoolServiceSchedulerFactoryTest {
}
@Test
- public void testFallBackSchedulerWithAdaptiveSchedulerTestProperty() {
+ void testFallBackSchedulerWithAdaptiveSchedulerTestProperty() {
String propertyValue = saveAdaptiveSchedulerTestPropertyValue();
System.setProperty("flink.tests.enable-adaptive-scheduler", "true");
@@ -102,7 +102,7 @@ public class DefaultSlotPoolServiceSchedulerFactoryTest {
}
@Test
- public void testFallBackSchedulerWithoutAdaptiveSchedulerTestProperty() {
+ void testFallBackSchedulerWithoutAdaptiveSchedulerTestProperty() {
String propertyValue = saveAdaptiveSchedulerTestPropertyValue();
System.clearProperty("flink.tests.enable-adaptive-scheduler");
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobExecutionITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobExecutionITCase.java
index 283fe08d24a..f649f6cc6ed 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobExecutionITCase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobExecutionITCase.java
@@ -26,17 +26,15 @@ import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.minicluster.TestingMiniCluster;
import org.apache.flink.runtime.minicluster.TestingMiniClusterConfiguration;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.concurrent.CompletableFuture;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Integration tests for job scheduling. */
-public class JobExecutionITCase extends TestLogger {
+class JobExecutionITCase {
/**
* Tests that tasks with a co-location constraint are scheduled in the
same slots. In fact it
@@ -44,7 +42,7 @@ public class JobExecutionITCase extends TestLogger {
* constraint is deactivated.
*/
@Test
- public void testCoLocationConstraintJobExecution() throws Exception {
+ void testCoLocationConstraintJobExecution() throws Exception {
final int numSlotsPerTaskExecutor = 1;
final int numTaskExecutors = 3;
final int parallelism = numTaskExecutors * numSlotsPerTaskExecutor;
@@ -66,7 +64,7 @@ public class JobExecutionITCase extends TestLogger {
final CompletableFuture<JobResult> jobResultFuture =
miniCluster.requestJobResult(jobGraph.getJobID());
- assertThat(jobResultFuture.get().isSuccess(), is(true));
+ assertThat(jobResultFuture.get().isSuccess()).isTrue();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobIntermediateDatasetReuseTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobIntermediateDatasetReuseTest.java
index a0ce9132e3f..e69f5b133b0 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobIntermediateDatasetReuseTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobIntermediateDatasetReuseTest.java
@@ -44,26 +44,25 @@ import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
/** Integration tests for reusing persisted intermediate dataset */
-public class JobIntermediateDatasetReuseTest {
+class JobIntermediateDatasetReuseTest {
private static final Logger LOG =
LoggerFactory.getLogger(JobIntermediateDatasetReuseTest.class);
@Test
- public void testClusterPartitionReuse() throws Exception {
+ void testClusterPartitionReuse() throws Exception {
internalTestClusterPartitionReuse(
1, 1, jobResult ->
Assertions.assertThat(jobResult.isSuccess()).isTrue());
}
@Test
- public void testClusterPartitionReuseMultipleParallelism() throws
Exception {
+ void testClusterPartitionReuseMultipleParallelism() throws Exception {
internalTestClusterPartitionReuse(
64, 64, jobResult ->
Assertions.assertThat(jobResult.isSuccess()).isTrue());
}
@Test
- public void
testClusterPartitionReuseWithMoreConsumerParallelismThrowException()
- throws Exception {
+ void testClusterPartitionReuseWithMoreConsumerParallelismThrowException()
throws Exception {
internalTestClusterPartitionReuse(
1,
2,
@@ -75,8 +74,7 @@ public class JobIntermediateDatasetReuseTest {
}
@Test
- public void
testClusterPartitionReuseWithLessConsumerParallelismThrowException()
- throws Exception {
+ void testClusterPartitionReuseWithLessConsumerParallelismThrowException()
throws Exception {
internalTestClusterPartitionReuse(
2,
1,
@@ -118,7 +116,7 @@ public class JobIntermediateDatasetReuseTest {
}
@Test
- public void testClusterPartitionReuseWithTMFail() throws Exception {
+ void testClusterPartitionReuseWithTMFail() throws Exception {
final TestingMiniClusterConfiguration miniClusterConfiguration =
TestingMiniClusterConfiguration.newBuilder().build();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResultTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResultTest.java
index f7a43e6070e..ca18f581329 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResultTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerRunnerResultTest.java
@@ -21,61 +21,59 @@ package org.apache.flink.runtime.jobmaster;
import
org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link JobManagerRunnerResult}. */
-public class JobManagerRunnerResultTest extends TestLogger {
+class JobManagerRunnerResultTest {
private final ExecutionGraphInfo executionGraphInfo =
new ExecutionGraphInfo(new
ArchivedExecutionGraphBuilder().build());
private final FlinkException testException = new FlinkException("test
exception");
@Test
- public void testSuccessfulJobManagerResult() {
+ void testSuccessfulJobManagerResult() {
final JobManagerRunnerResult jobManagerRunnerResult =
JobManagerRunnerResult.forSuccess(executionGraphInfo);
- assertTrue(jobManagerRunnerResult.isSuccess());
- assertFalse(jobManagerRunnerResult.isInitializationFailure());
+ assertThat(jobManagerRunnerResult.isSuccess()).isTrue();
+ assertThat(jobManagerRunnerResult.isInitializationFailure()).isFalse();
}
@Test
- public void testInitializationFailureJobManagerResult() {
+ void testInitializationFailureJobManagerResult() {
final JobManagerRunnerResult jobManagerRunnerResult =
JobManagerRunnerResult.forInitializationFailure(executionGraphInfo,
testException);
- assertTrue(jobManagerRunnerResult.isInitializationFailure());
- assertFalse(jobManagerRunnerResult.isSuccess());
+ assertThat(jobManagerRunnerResult.isInitializationFailure()).isTrue();
+ assertThat(jobManagerRunnerResult.isSuccess()).isFalse();
}
@Test
- public void testGetArchivedExecutionGraphFromSuccessfulJobManagerResult() {
+ void testGetArchivedExecutionGraphFromSuccessfulJobManagerResult() {
final JobManagerRunnerResult jobManagerRunnerResult =
JobManagerRunnerResult.forSuccess(executionGraphInfo);
- assertThat(jobManagerRunnerResult.getExecutionGraphInfo(),
is(executionGraphInfo));
+
assertThat(jobManagerRunnerResult.getExecutionGraphInfo()).isEqualTo(executionGraphInfo);
}
@Test
- public void testGetInitializationFailureFromFailedJobManagerResult() {
+ void testGetInitializationFailureFromFailedJobManagerResult() {
final JobManagerRunnerResult jobManagerRunnerResult =
JobManagerRunnerResult.forInitializationFailure(executionGraphInfo,
testException);
- assertThat(jobManagerRunnerResult.getInitializationFailure(),
is(testException));
+
assertThat(jobManagerRunnerResult.getInitializationFailure()).isEqualTo(testException);
}
- @Test(expected = IllegalStateException.class)
- public void testGetInitializationFailureFromSuccessfulJobManagerResult() {
+ @Test
+ void testGetInitializationFailureFromSuccessfulJobManagerResult() {
final JobManagerRunnerResult jobManagerRunnerResult =
JobManagerRunnerResult.forSuccess(executionGraphInfo);
- jobManagerRunnerResult.getInitializationFailure();
+ assertThatThrownBy(jobManagerRunnerResult::getInitializationFailure)
+ .isInstanceOf(IllegalStateException.class);
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
index 1a2b050031b..1777edf0cd0 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobManagerSharedServicesTest.java
@@ -25,33 +25,32 @@ import org.apache.flink.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.VoidBlobStore;
import org.apache.flink.runtime.util.Hardware;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.ThrowingRunnable;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import javax.annotation.Nonnull;
+import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests auxiliary shared services created by {@link JobManagerSharedServices}
and used by the
* {@link JobMaster}.
*/
-public class JobManagerSharedServicesTest extends TestLogger {
+class JobManagerSharedServicesTest {
private static final int CPU_CORES = Hardware.getNumberCPUCores();
- @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new
TemporaryFolder();
+ @TempDir private File TEMPORARY_FOLDER;
@Test
- public void testFutureExecutorNoConfiguration() throws Exception {
+ void testFutureExecutorNoConfiguration() throws Exception {
final Configuration config = new Configuration();
final JobManagerSharedServices jobManagerSharedServices =
@@ -67,7 +66,7 @@ public class JobManagerSharedServicesTest extends TestLogger {
}
@Test
- public void testFutureExecutorConfiguration() throws Exception {
+ void testFutureExecutorConfiguration() throws Exception {
final int futurePoolSize = 8;
final Configuration config = new Configuration();
config.set(JobManagerOptions.JOB_MANAGER_FUTURE_POOL_SIZE,
futurePoolSize);
@@ -81,7 +80,7 @@ public class JobManagerSharedServicesTest extends TestLogger {
}
@Test
- public void testIoExecutorNoConfiguration() throws Exception {
+ void testIoExecutorNoConfiguration() throws Exception {
final Configuration config = new Configuration();
final JobManagerSharedServices jobManagerSharedServices =
@@ -95,7 +94,7 @@ public class JobManagerSharedServicesTest extends TestLogger {
}
@Test
- public void testIoExecutorConfiguration() throws Exception {
+ void testIoExecutorConfiguration() throws Exception {
final int ioPoolSize = 5;
final Configuration config = new Configuration();
config.set(JobManagerOptions.JOB_MANAGER_IO_POOL_SIZE, ioPoolSize);
@@ -115,7 +114,7 @@ public class JobManagerSharedServicesTest extends
TestLogger {
throws Exception {
return JobManagerSharedServices.fromConfiguration(
configuration,
- new BlobServer(configuration, TEMPORARY_FOLDER.newFolder(),
new VoidBlobStore()),
+ new BlobServer(configuration, TEMPORARY_FOLDER, new
VoidBlobStore()),
new TestingFatalErrorHandler());
}
@@ -141,7 +140,7 @@ public class JobManagerSharedServicesTest extends
TestLogger {
// the expected pool size latch should complete since we expect to
have enough threads
expectedPoolSizeLatch.await();
- assertEquals(1, expectedPoolSizePlusOneLatch.getCount());
+ assertThat(expectedPoolSizePlusOneLatch.getCount()).isOne();
// unblock the runnables
releaseLatch.trigger();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterExecutionDeploymentReconciliationTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterExecutionDeploymentReconciliationTest.java
index 7549860c70c..edd34a136ff 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterExecutionDeploymentReconciliationTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterExecutionDeploymentReconciliationTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.jobmaster;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
@@ -35,7 +36,7 @@ import
org.apache.flink.runtime.jobmaster.utils.JobMasterBuilder;
import org.apache.flink.runtime.leaderelection.TestingLeaderElection;
import org.apache.flink.runtime.leaderretrieval.SettableLeaderRetrievalService;
import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
import org.apache.flink.runtime.taskexecutor.AccumulatorReport;
import org.apache.flink.runtime.taskexecutor.ExecutionDeploymentReport;
import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
@@ -45,14 +46,12 @@ import
org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
import org.apache.flink.runtime.taskexecutor.slot.SlotOffer;
import org.apache.flink.runtime.taskmanager.LocalUnresolvedTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.UnresolvedTaskManagerLocation;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.util.TestLogger;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Collection;
import java.util.Collections;
@@ -61,13 +60,12 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static
org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
-import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the execution deployment-reconciliation logic in the {@link
JobMaster}. */
-public class JobMasterExecutionDeploymentReconciliationTest extends TestLogger
{
+class JobMasterExecutionDeploymentReconciliationTest {
private static final Time testingTimeout = Time.seconds(10L);
@@ -79,16 +77,19 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
private final SettableLeaderRetrievalService
resourceManagerLeaderRetriever =
new SettableLeaderRetrievalService();
- @ClassRule
- public static final TestingRpcServiceResource RPC_SERVICE_RESOURCE =
- new TestingRpcServiceResource();
+ public static final TestingRpcServiceExtension
TESTING_RPC_SERVICE_EXTENSION =
+ new TestingRpcServiceExtension();
- @Rule
- public final TestingFatalErrorHandlerResource
testingFatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
+ @RegisterExtension
+ private static final AllCallbackWrapper<TestingRpcServiceExtension>
+ RPC_SERVICE_EXTENSION_WRAPPER = new
AllCallbackWrapper<>(TESTING_RPC_SERVICE_EXTENSION);
- @Before
- public void setup() {
+ @RegisterExtension
+ private final TestingFatalErrorHandlerExtension
testingFatalErrorHandlerExtension =
+ new TestingFatalErrorHandlerExtension();
+
+ @BeforeEach
+ private void setup() {
haServices.setResourceManagerLeaderRetriever(resourceManagerLeaderRetriever);
haServices.setResourceManagerLeaderElection(new
TestingLeaderElection());
haServices.setCheckpointRecoveryFactory(new
StandaloneCheckpointRecoveryFactory());
@@ -96,7 +97,7 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
/** Tests how the job master handles unknown/missing executions. */
@Test
- public void testExecutionDeploymentReconciliation() throws Exception {
+ void testExecutionDeploymentReconciliation() throws Exception {
JobMasterBuilder.TestingOnCompletionActions onCompletionActions =
new JobMasterBuilder.TestingOnCompletionActions();
@@ -106,7 +107,7 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
try (JobMaster jobMaster =
createAndStartJobMaster(onCompletionActions,
deploymentTrackerWrapper, jobGraph)) {
JobMasterGateway jobMasterGateway =
jobMaster.getSelfGateway(JobMasterGateway.class);
- RPC_SERVICE_RESOURCE
+ TESTING_RPC_SERVICE_EXTENSION
.getTestingRpcService()
.registerGateway(jobMasterGateway.getAddress(),
jobMasterGateway);
@@ -125,7 +126,7 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
ExecutionAttemptID deployedExecution =
deploymentTrackerWrapper.getTaskDeploymentFuture().get();
- assertFalse(taskCancellationFuture.isDone());
+ assertThatFuture(taskCancellationFuture).isNotDone();
ExecutionAttemptID unknownDeployment = createExecutionAttemptId();
// the deployment report is missing the just deployed task, but
contains the ID of some
@@ -138,16 +139,20 @@ public class
JobMasterExecutionDeploymentReconciliationTest extends TestLogger {
new ExecutionDeploymentReport(
Collections.singleton(unknownDeployment))));
- assertThat(taskCancellationFuture.get(), is(unknownDeployment));
- assertThat(deploymentTrackerWrapper.getStopFuture().get(),
is(deployedExecution));
+ assertThatFuture(taskCancellationFuture)
+ .eventuallySucceeds()
+ .isEqualTo(unknownDeployment);
+ assertThatFuture(deploymentTrackerWrapper.getStopFuture())
+ .eventuallySucceeds()
+ .isEqualTo(deployedExecution);
assertThat(
- onCompletionActions
- .getJobReachedGloballyTerminalStateFuture()
- .get()
- .getArchivedExecutionGraph()
- .getState(),
- is(JobStatus.FAILED));
+ onCompletionActions
+ .getJobReachedGloballyTerminalStateFuture()
+ .get()
+ .getArchivedExecutionGraph()
+ .getState())
+ .isEqualTo(JobStatus.FAILED);
}
}
@@ -156,13 +161,13 @@ public class
JobMasterExecutionDeploymentReconciliationTest extends TestLogger {
* for which the deployment was not yet acknowledged.
*/
@Test
- public void testExecutionDeploymentReconciliationForPendingExecution()
throws Exception {
+ void testExecutionDeploymentReconciliationForPendingExecution() throws
Exception {
TestingExecutionDeploymentTrackerWrapper deploymentTrackerWrapper =
new TestingExecutionDeploymentTrackerWrapper();
final JobGraph jobGraph = JobGraphTestUtils.singleNoOpJobGraph();
try (JobMaster jobMaster =
createAndStartJobMaster(deploymentTrackerWrapper, jobGraph)) {
JobMasterGateway jobMasterGateway =
jobMaster.getSelfGateway(JobMasterGateway.class);
- RPC_SERVICE_RESOURCE
+ TESTING_RPC_SERVICE_EXTENSION
.getTestingRpcService()
.registerGateway(jobMasterGateway.getAddress(),
jobMasterGateway);
@@ -200,7 +205,7 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
taskSubmissionAcknowledgeFuture.complete(Acknowledge.get());
deploymentTrackerWrapper.getTaskDeploymentFuture().get();
- assertFalse(taskCancellationFuture.isDone());
+ assertThatFuture(taskCancellationFuture).isNotDone();
}
}
@@ -220,9 +225,9 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
throws Exception {
JobMaster jobMaster =
- new JobMasterBuilder(jobGraph,
RPC_SERVICE_RESOURCE.getTestingRpcService())
+ new JobMasterBuilder(jobGraph,
TESTING_RPC_SERVICE_EXTENSION.getTestingRpcService())
.withFatalErrorHandler(
-
testingFatalErrorHandlerResource.getFatalErrorHandler())
+
testingFatalErrorHandlerExtension.getTestingFatalErrorHandler())
.withHighAvailabilityServices(haServices)
.withHeartbeatServices(heartbeatServices)
.withExecutionDeploymentTracker(executionDeploymentTracker)
@@ -261,7 +266,7 @@ public class JobMasterExecutionDeploymentReconciliationTest
extends TestLogger {
})
.createTestingTaskExecutorGateway();
- RPC_SERVICE_RESOURCE
+ TESTING_RPC_SERVICE_EXTENSION
.getTestingRpcService()
.registerGateway(taskExecutorGateway.getAddress(),
taskExecutorGateway);
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterPartitionReleaseTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterPartitionReleaseTest.java
index 1e757982933..3477f74fde4 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterPartitionReleaseTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterPartitionReleaseTest.java
@@ -51,16 +51,15 @@ import
org.apache.flink.runtime.taskmanager.LocalUnresolvedTaskManagerLocation;
import org.apache.flink.runtime.taskmanager.TaskExecutionState;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.util.TestLogger;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
@@ -71,14 +70,13 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.containsInAnyOrder;
-import static org.hamcrest.Matchers.equalTo;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the partition release logic of the {@link JobMaster}. */
-public class JobMasterPartitionReleaseTest extends TestLogger {
+class JobMasterPartitionReleaseTest {
- @ClassRule public static TemporaryFolder temporaryFolder = new
TemporaryFolder();
+ @TempDir private static File temporaryFolder;
private static final Time testingTimeout = Time.seconds(10L);
@@ -86,18 +84,18 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
private TestingFatalErrorHandler testingFatalErrorHandler;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ private static void setupClass() {
rpcService = new TestingRpcService();
}
- @Before
- public void setup() throws IOException {
+ @BeforeEach
+ private void setup() throws IOException {
testingFatalErrorHandler = new TestingFatalErrorHandler();
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ private void teardown() throws Exception {
if (testingFatalErrorHandler != null) {
testingFatalErrorHandler.rethrowError();
}
@@ -105,8 +103,8 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
rpcService.clearGateways();
}
- @AfterClass
- public static void teardownClass() {
+ @AfterAll
+ private static void teardownClass() {
if (rpcService != null) {
rpcService.closeAsync();
rpcService = null;
@@ -114,7 +112,7 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
}
@Test
- public void testPartitionTableCleanupOnDisconnect() throws Exception {
+ void testPartitionTableCleanupOnDisconnect() throws Exception {
final CompletableFuture<JobID> disconnectTaskExecutorFuture = new
CompletableFuture<>();
final TestingTaskExecutorGateway testingTaskExecutorGateway =
new TestingTaskExecutorGatewayBuilder()
@@ -131,20 +129,20 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
testSetup.getTaskExecutorResourceID(), new
Exception("test"));
disconnectTaskExecutorFuture.get();
- assertThat(
-
testSetup.getStopTrackingPartitionsTargetResourceId().get(),
- equalTo(testSetup.getTaskExecutorResourceID()));
+
assertThatFuture(testSetup.getStopTrackingPartitionsTargetResourceId())
+ .eventuallySucceeds()
+ .isEqualTo(testSetup.getTaskExecutorResourceID());
}
}
@Test
- public void testPartitionReleaseOrPromotionOnJobSuccess() throws Exception
{
+ void testPartitionReleaseOrPromotionOnJobSuccess() throws Exception {
testPartitionReleaseOrPromotionOnJobTermination(
TestSetup::getPartitionsForReleaseOrPromote,
ExecutionState.FINISHED);
}
@Test
- public void testPartitionReleaseOrPromotionOnJobFailure() throws Exception
{
+ void testPartitionReleaseOrPromotionOnJobFailure() throws Exception {
testPartitionReleaseOrPromotionOnJobTermination(
TestSetup::getPartitionsForRelease, ExecutionState.FAILED);
}
@@ -192,16 +190,13 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
jobMasterGateway.updateTaskExecutionState(
new TaskExecutionState(
taskDeploymentDescriptor.getExecutionAttemptId(),
finalExecutionState));
- assertThat(
- callSelector.apply(testSetup).get(),
- containsInAnyOrder(partitionID0, partitionID1));
+ assertThat(callSelector.apply(testSetup).get())
+ .containsExactlyInAnyOrder(partitionID0, partitionID1);
}
}
private static class TestSetup implements AutoCloseable {
- private final TemporaryFolder temporaryFolder = new TemporaryFolder();
-
private final LocalUnresolvedTaskManagerLocation
localTaskManagerUnresolvedLocation =
new LocalUnresolvedTaskManagerLocation();
@@ -224,8 +219,6 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
TaskExecutorGateway taskExecutorGateway)
throws Exception {
- temporaryFolder.create();
-
TestingHighAvailabilityServices haServices = new
TestingHighAvailabilityServices();
haServices.setCheckpointRecoveryFactory(new
StandaloneCheckpointRecoveryFactory());
@@ -243,8 +236,7 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
Configuration configuration = new Configuration();
configuration.set(
- BlobServerOptions.STORAGE_DIRECTORY,
- temporaryFolder.newFolder().getAbsolutePath());
+ BlobServerOptions.STORAGE_DIRECTORY,
temporaryFolder.getAbsolutePath());
HeartbeatServices heartbeatServices = new
HeartbeatServicesImpl(1000L, 5_000_000L);
@@ -327,12 +319,8 @@ public class JobMasterPartitionReleaseTest extends
TestLogger {
}
public void close() throws Exception {
- try {
- if (jobMaster != null) {
- RpcUtils.terminateRpcEndpoint(jobMaster);
- }
- } finally {
- temporaryFolder.delete();
+ if (jobMaster != null) {
+ RpcUtils.terminateRpcEndpoint(jobMaster);
}
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterQueryableStateTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterQueryableStateTest.java
index 06cc4f32ae4..052216d6644 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterQueryableStateTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterQueryableStateTest.java
@@ -40,31 +40,24 @@ import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.state.KeyGroupRange;
import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.TestingTaskExecutorGatewayBuilder;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.function.Consumer;
-import static org.apache.flink.core.testutils.FlinkMatchers.containsCause;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.hamcrest.CoreMatchers.either;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
/** Tests for the queryable-state logic of the {@link JobMaster}. */
-public class JobMasterQueryableStateTest extends TestLogger {
+class JobMasterQueryableStateTest {
private static final Time testingTimeout = Time.seconds(10L);
@@ -96,18 +89,18 @@ public class JobMasterQueryableStateTest extends TestLogger
{
JOB_GRAPH.setJobType(JobType.STREAMING);
}
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ private static void setupClass() {
rpcService = new TestingRpcService();
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ private void teardown() throws Exception {
rpcService.clearGateways();
}
- @AfterClass
- public static void teardownClass() {
+ @AfterAll
+ private static void teardownClass() {
if (rpcService != null) {
rpcService.closeAsync();
rpcService = null;
@@ -115,7 +108,7 @@ public class JobMasterQueryableStateTest extends TestLogger
{
}
@Test
- public void testRequestKvStateWithoutRegistration() throws Exception {
+ void testRequestKvStateWithoutRegistration() throws Exception {
try (final JobMaster jobMaster =
new JobMasterBuilder(JOB_GRAPH, rpcService).createJobMaster())
{
@@ -136,7 +129,7 @@ public class JobMasterQueryableStateTest extends TestLogger
{
}
@Test
- public void testRequestKvStateOfWrongJob() throws Exception {
+ void testRequestKvStateOfWrongJob() throws Exception {
try (final JobMaster jobMaster =
new JobMasterBuilder(JOB_GRAPH, rpcService).createJobMaster())
{
@@ -157,7 +150,7 @@ public class JobMasterQueryableStateTest extends TestLogger
{
}
@Test
- public void testRequestKvStateWithIrrelevantRegistration() throws
Exception {
+ void testRequestKvStateWithIrrelevantRegistration() throws Exception {
try (final JobMaster jobMaster =
new JobMasterBuilder(JOB_GRAPH, rpcService).createJobMaster())
{
@@ -181,7 +174,7 @@ public class JobMasterQueryableStateTest extends TestLogger
{
}
@Test
- public void testRegisterKvState() throws Exception {
+ void testRegisterKvState() throws Exception {
try (JobMaster jobMaster = new JobMasterBuilder(JOB_GRAPH,
rpcService).createJobMaster()) {
jobMaster.start();
@@ -211,19 +204,20 @@ public class JobMasterQueryableStateTest extends
TestLogger {
.requestKvStateLocation(JOB_GRAPH.getJobID(),
registrationName)
.get();
- assertEquals(JOB_GRAPH.getJobID(), location.getJobId());
- assertEquals(JOB_VERTEX_1.getID(), location.getJobVertexId());
- assertEquals(JOB_VERTEX_1.getMaxParallelism(),
location.getNumKeyGroups());
- assertEquals(1, location.getNumRegisteredKeyGroups());
- assertEquals(1, keyGroupRange.getNumberOfKeyGroups());
- assertEquals(kvStateID,
location.getKvStateID(keyGroupRange.getStartKeyGroup()));
- assertEquals(
- address,
location.getKvStateServerAddress(keyGroupRange.getStartKeyGroup()));
+ assertThat(location.getJobId()).isEqualTo(JOB_GRAPH.getJobID());
+
assertThat(location.getJobVertexId()).isEqualTo(JOB_VERTEX_1.getID());
+
assertThat(location.getNumKeyGroups()).isEqualTo(JOB_VERTEX_1.getMaxParallelism());
+ assertThat(location.getNumRegisteredKeyGroups()).isOne();
+ assertThat(keyGroupRange.getNumberOfKeyGroups()).isOne();
+ assertThat(location.getKvStateID(keyGroupRange.getStartKeyGroup()))
+ .isEqualTo(kvStateID);
+
assertThat(location.getKvStateServerAddress(keyGroupRange.getStartKeyGroup()))
+ .isEqualTo(address);
}
}
@Test
- public void testUnregisterKvState() throws Exception {
+ void testUnregisterKvState() throws Exception {
try (final JobMaster jobMaster =
new JobMasterBuilder(JOB_GRAPH, rpcService).createJobMaster())
{
@@ -258,19 +252,21 @@ public class JobMasterQueryableStateTest extends
TestLogger {
registrationName)
.get();
- try {
- jobMasterGateway
- .requestKvStateLocation(JOB_GRAPH.getJobID(),
registrationName)
- .get();
- fail("Expected to fail with an UnknownKvStateLocation.");
- } catch (Exception e) {
- assertThat(e, containsCause(UnknownKvStateLocation.class));
- }
+ assertThatThrownBy(
+ () -> {
+ jobMasterGateway
+ .requestKvStateLocation(
+ JOB_GRAPH.getJobID(),
registrationName)
+ .get();
+ })
+ .as("Expected to fail with an UnknownKvStateLocation.")
+ .isInstanceOf(Exception.class)
+ .hasCauseInstanceOf(UnknownKvStateLocation.class);
}
}
@Test
- public void testDuplicatedKvStateRegistrationsFailTask() throws Exception {
+ void testDuplicatedKvStateRegistrationsFailTask() throws Exception {
try (final JobMaster jobMaster =
new JobMasterBuilder(JOB_GRAPH, rpcService).createJobMaster())
{
@@ -287,22 +283,23 @@ public class JobMasterQueryableStateTest extends
TestLogger {
registerKvState(
jobMasterGateway, JOB_GRAPH.getJobID(),
JOB_VERTEX_1.getID(), registrationName);
- try {
- registerKvState(
- jobMasterGateway,
- JOB_GRAPH.getJobID(),
- JOB_VERTEX_2.getID(),
- registrationName);
- fail("Expected to fail because of clashing registration
message.");
- } catch (Exception e) {
- assertTrue(
- ExceptionUtils.findThrowableWithMessage(e,
"Registration name clash")
- .isPresent());
-
- assertThat(
-
jobMasterGateway.requestJobStatus(testingTimeout).get(),
-
either(is(JobStatus.FAILED)).or(is(JobStatus.FAILING)));
- }
+ assertThatThrownBy(
+ () ->
+ registerKvState(
+ jobMasterGateway,
+ JOB_GRAPH.getJobID(),
+ JOB_VERTEX_2.getID(),
+ registrationName))
+ .as("Expected to fail because of clashing registration
message.")
+ .isInstanceOf(Exception.class)
+ .hasMessageContaining("Registration name clash");
+ assertThat(jobMasterGateway.requestJobStatus(testingTimeout).get())
+ .satisfies(
+ (Consumer<JobStatus>)
+ jobStatus -> {
+ assert jobStatus == JobStatus.FAILED
+ || jobStatus ==
JobStatus.FAILING;
+ });
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterSchedulerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterSchedulerTest.java
index f5dd1e649da..1594eedac9e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterSchedulerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterSchedulerTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.core.failure.FailureEnricher;
+import org.apache.flink.core.testutils.AllCallbackWrapper;
import org.apache.flink.runtime.blob.BlobWriter;
import org.apache.flink.runtime.blocklist.BlocklistOperations;
import org.apache.flink.runtime.checkpoint.CheckpointRecoveryFactory;
@@ -35,36 +36,35 @@ import
org.apache.flink.runtime.jobmaster.slotpool.TestingSlotPoolServiceBuilder
import org.apache.flink.runtime.jobmaster.utils.JobMasterBuilder;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.runtime.rpc.TestingRpcServiceResource;
+import org.apache.flink.runtime.rpc.TestingRpcServiceExtension;
import org.apache.flink.runtime.scheduler.SchedulerNG;
import org.apache.flink.runtime.scheduler.SchedulerNGFactory;
import org.apache.flink.runtime.scheduler.TestingSchedulerNG;
import org.apache.flink.runtime.shuffle.ShuffleMaster;
import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.slf4j.Logger;
import java.util.Collection;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
-import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
-/** Tests for the JobMaster scheduler interaction. */
-public class JobMasterSchedulerTest extends TestLogger {
+class JobMasterSchedulerTest {
- @ClassRule
- public static final TestingRpcServiceResource TESTING_RPC_SERVICE_RESOURCE
=
- new TestingRpcServiceResource();
+ private static final TestingRpcServiceExtension
TESTING_RPC_SERVICE_EXTENSION =
+ new TestingRpcServiceExtension();
+
+ @RegisterExtension
+ private static final AllCallbackWrapper<TestingRpcServiceExtension>
+ RPC_SERVICE_EXTENSION_WRAPPER = new
AllCallbackWrapper<>(TESTING_RPC_SERVICE_EXTENSION);
/** Tests that the JobMaster fails if we cannot start the scheduling. See
FLINK-20382. */
@Test
- public void testIfStartSchedulingFailsJobMasterFails() throws Exception {
+ void testIfStartSchedulingFailsJobMasterFails() throws Exception {
final SchedulerNGFactory schedulerFactory = new
FailingSchedulerFactory();
final JobMasterBuilder.TestingOnCompletionActions onCompletionActions =
new JobMasterBuilder.TestingOnCompletionActions();
@@ -74,7 +74,7 @@ public class JobMasterSchedulerTest extends TestLogger {
final JobMaster jobMaster =
new JobMasterBuilder(
JobGraphTestUtils.emptyJobGraph(),
-
TESTING_RPC_SERVICE_RESOURCE.getTestingRpcService())
+
TESTING_RPC_SERVICE_EXTENSION.getTestingRpcService())
.withSlotPoolServiceSchedulerFactory(
DefaultSlotPoolServiceSchedulerFactory.create(
TestingSlotPoolServiceBuilder.newBuilder(),
@@ -85,9 +85,8 @@ public class JobMasterSchedulerTest extends TestLogger {
jobMaster.start();
- assertThat(
- onCompletionActions.getJobMasterFailedFuture().join(),
- is(instanceOf(JobMasterException.class)));
+ assertThat(onCompletionActions.getJobMasterFailedFuture().join())
+ .isInstanceOf(JobMasterException.class);
// close the jobMaster to remove it from the testing rpc service
so that it can shut
// down cleanly
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobRecoveryITCase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobRecoveryITCase.java
index 5beedf0c02e..e075427e8b1 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobRecoveryITCase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobRecoveryITCase.java
@@ -28,55 +28,53 @@ import org.apache.flink.runtime.jobgraph.JobGraphBuilder;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.minicluster.MiniCluster;
-import org.apache.flink.runtime.testutils.MiniClusterResource;
+import org.apache.flink.runtime.testutils.InternalMiniClusterExtension;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the recovery of task failures. */
-public class JobRecoveryITCase extends TestLogger {
+class JobRecoveryITCase {
private static final int NUM_TMS = 1;
private static final int SLOTS_PER_TM = 10;
private static final int PARALLELISM = NUM_TMS * SLOTS_PER_TM;
- @ClassRule
- public static final MiniClusterResource MINI_CLUSTER_RESOURCE =
- new MiniClusterResource(
+ @RegisterExtension
+ private static final InternalMiniClusterExtension MINI_CLUSTER_EXTENSION =
+ new InternalMiniClusterExtension(
new MiniClusterResourceConfiguration.Builder()
.setNumberTaskManagers(NUM_TMS)
.setNumberSlotsPerTaskManager(SLOTS_PER_TM)
.build());
@Test
- public void testTaskFailureRecovery() throws Exception {
+ void testTaskFailureRecovery() throws Exception {
runTaskFailureRecoveryTest(createjobGraph(false));
}
@Test
- public void testTaskFailureWithSlotSharingRecovery() throws Exception {
+ void testTaskFailureWithSlotSharingRecovery() throws Exception {
runTaskFailureRecoveryTest(createjobGraph(true));
}
private void runTaskFailureRecoveryTest(final JobGraph jobGraph) throws
Exception {
- final MiniCluster miniCluster = MINI_CLUSTER_RESOURCE.getMiniCluster();
+ final MiniCluster miniCluster =
MINI_CLUSTER_EXTENSION.getMiniCluster();
miniCluster.submitJob(jobGraph).get();
final CompletableFuture<JobResult> jobResultFuture =
miniCluster.requestJobResult(jobGraph.getJobID());
- assertThat(jobResultFuture.get().isSuccess(), is(true));
+ assertThat(jobResultFuture.get().isSuccess()).isTrue();
}
private JobGraph createjobGraph(boolean slotSharingEnabled) throws
IOException {
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java
index c8ad273b94d..f335cd26dfd 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java
@@ -26,31 +26,24 @@ import org.apache.flink.runtime.executiongraph.ErrorInfo;
import
org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.SerializedThrowable;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.core.IsNull.nullValue;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link JobResult}. */
-public class JobResultTest extends TestLogger {
+class JobResultTest {
@Test
- public void testNetRuntimeMandatory() {
- try {
- new JobResult.Builder().jobId(new JobID()).build();
- fail("Expected exception not thrown");
- } catch (final IllegalArgumentException e) {
- assertThat(e.getMessage(), equalTo("netRuntime must be greater
than or equals 0"));
- }
+ void testNetRuntimeMandatory() {
+ assertThatThrownBy(() -> new JobResult.Builder().jobId(new
JobID()).build())
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("netRuntime must be greater than or
equals 0");
}
@Test
- public void testIsNotSuccess() throws Exception {
+ void testIsNotSuccess() {
final JobResult jobResult =
new JobResult.Builder()
.jobId(new JobID())
@@ -58,19 +51,19 @@ public class JobResultTest extends TestLogger {
.netRuntime(Long.MAX_VALUE)
.build();
- assertThat(jobResult.isSuccess(), equalTo(false));
+ assertThat(jobResult.isSuccess()).isFalse();
}
@Test
- public void testIsSuccess() throws Exception {
+ void testIsSuccess() {
final JobResult jobResult =
new JobResult.Builder().jobId(new
JobID()).netRuntime(Long.MAX_VALUE).build();
- assertThat(jobResult.isSuccess(), equalTo(true));
+ assertThat(jobResult.isSuccess()).isTrue();
}
@Test
- public void testCancelledJobIsFailureResult() {
+ void testCancelledJobIsFailureResult() {
final JobResult jobResult =
JobResult.createFrom(
new ArchivedExecutionGraphBuilder()
@@ -78,11 +71,11 @@ public class JobResultTest extends TestLogger {
.setState(JobStatus.CANCELED)
.build());
- assertThat(jobResult.isSuccess(), is(false));
+ assertThat(jobResult.isSuccess()).isFalse();
}
@Test
- public void testFailedJobIsFailureResult() {
+ void testFailedJobIsFailureResult() {
final JobResult jobResult =
JobResult.createFrom(
new ArchivedExecutionGraphBuilder()
@@ -92,11 +85,11 @@ public class JobResultTest extends TestLogger {
new ErrorInfo(new FlinkException("Test
exception"), 42L))
.build());
- assertThat(jobResult.isSuccess(), is(false));
+ assertThat(jobResult.isSuccess()).isFalse();
}
@Test
- public void testCancelledJobThrowsJobCancellationException() throws
Exception {
+ void testCancelledJobThrowsJobCancellationException() {
final FlinkException cause = new FlinkException("Test exception");
final JobResult jobResult =
JobResult.createFrom(
@@ -106,18 +99,16 @@ public class JobResultTest extends TestLogger {
.setFailureCause(new ErrorInfo(cause, 42L))
.build());
- try {
- jobResult.toJobExecutionResult(getClass().getClassLoader());
- fail("Job should fail with an JobCancellationException.");
- } catch (JobCancellationException expected) {
- // the failure cause in the execution graph should not be the
cause of the canceled job
- // result
- assertThat(expected.getCause(), is(nullValue()));
- }
+ assertThatThrownBy(
+ () -> {
+
jobResult.toJobExecutionResult(getClass().getClassLoader());
+ })
+ .isInstanceOf(JobCancellationException.class)
+ .hasNoCause();
}
@Test
- public void testFailedJobThrowsJobExecutionException() throws Exception {
+ void testFailedJobThrowsJobExecutionException() {
final FlinkException cause = new FlinkException("Test exception");
final JobResult jobResult =
JobResult.createFrom(
@@ -127,20 +118,21 @@ public class JobResultTest extends TestLogger {
.setFailureCause(new ErrorInfo(cause, 42L))
.build());
- try {
- jobResult.toJobExecutionResult(getClass().getClassLoader());
- fail("Job should fail with JobExecutionException.");
- } catch (JobExecutionException expected) {
- assertThat(expected.getCause(), is(equalTo(cause)));
- }
+ assertThatThrownBy(() ->
jobResult.toJobExecutionResult(getClass().getClassLoader()))
+ .isInstanceOf(JobExecutionException.class)
+ .cause()
+ .isEqualTo(cause);
}
- @Test(expected = NullPointerException.class)
- public void testFailureResultRequiresFailureCause() {
- JobResult.createFrom(
- new ArchivedExecutionGraphBuilder()
- .setJobID(new JobID())
- .setState(JobStatus.FAILED)
- .build());
+ @Test
+ void testFailureResultRequiresFailureCause() {
+ assertThatThrownBy(
+ () ->
+ JobResult.createFrom(
+ new ArchivedExecutionGraphBuilder()
+ .setJobID(new JobID())
+ .setState(JobStatus.FAILED)
+ .build()))
+ .isInstanceOf(NullPointerException.class);
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractServiceConnectionManagerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractServiceConnectionManagerTest.java
index af67f3da32f..f8052275f95 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractServiceConnectionManagerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AbstractServiceConnectionManagerTest.java
@@ -18,36 +18,33 @@
package org.apache.flink.runtime.jobmaster.slotpool;
-import org.apache.flink.util.TestLogger;
+import org.junit.jupiter.api.Test;
-import org.junit.Test;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link
DefaultDeclareResourceRequirementServiceConnectionManager}. */
-public class AbstractServiceConnectionManagerTest extends TestLogger {
+class AbstractServiceConnectionManagerTest {
@Test
- public void testIsConnected() {
+ void testIsConnected() {
AbstractServiceConnectionManager<Object> connectionManager =
new TestServiceConnectionManager();
- assertThat(connectionManager.isConnected(), is(false));
+ assertThat(connectionManager.isConnected()).isFalse();
connectionManager.connect(new Object());
- assertThat(connectionManager.isConnected(), is(true));
+ assertThat(connectionManager.isConnected()).isTrue();
connectionManager.disconnect();
- assertThat(connectionManager.isConnected(), is(false));
+ assertThat(connectionManager.isConnected()).isFalse();
connectionManager.close();
- assertThat(connectionManager.isConnected(), is(false));
+ assertThat(connectionManager.isConnected()).isFalse();
}
@Test
- public void testCheckNotClosed() {
+ void testCheckNotClosed() {
AbstractServiceConnectionManager<Object> connectionManager =
new TestServiceConnectionManager();
@@ -60,11 +57,9 @@ public class AbstractServiceConnectionManagerTest extends
TestLogger {
connectionManager.checkNotClosed();
connectionManager.close();
- try {
- connectionManager.checkNotClosed();
- fail("checkNotClosed() did not fail for a closed connection
manager");
- } catch (IllegalStateException expected) {
- }
+ assertThatThrownBy(connectionManager::checkNotClosed)
+ .as("checkNotClosed() did not fail for a closed connection
manager")
+ .isInstanceOf(IllegalStateException.class);
}
private static class TestServiceConnectionManager
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AllocatedSlotOccupationTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AllocatedSlotOccupationTest.java
index 2ca5329d6ff..536de256d9c 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AllocatedSlotOccupationTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/AllocatedSlotOccupationTest.java
@@ -18,31 +18,28 @@
package org.apache.flink.runtime.jobmaster.slotpool;
-import org.apache.flink.util.TestLogger;
-
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import static
org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotTestUtils.createPhysicalSlot;
import static
org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotTestUtils.occupyPhysicalSlot;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests whether the slot occupation state of {@link AllocatedSlot} is
correctly. */
-public class AllocatedSlotOccupationTest extends TestLogger {
+class AllocatedSlotOccupationTest {
@Test
- public void testSingleTaskOccupyingSlotIndefinitely() {
+ void testSingleTaskOccupyingSlotIndefinitely() {
final PhysicalSlot physicalSlot = createPhysicalSlot();
occupyPhysicalSlot(physicalSlot, true);
- assertThat(physicalSlot.willBeOccupiedIndefinitely(), is(true));
+ assertThat(physicalSlot.willBeOccupiedIndefinitely()).isTrue();
}
@Test
- public void testSingleTaskNotOccupyingSlotIndefinitely() {
+ void testSingleTaskNotOccupyingSlotIndefinitely() {
final PhysicalSlot physicalSlot = createPhysicalSlot();
occupyPhysicalSlot(physicalSlot, false);
- assertThat(physicalSlot.willBeOccupiedIndefinitely(), is(false));
+ assertThat(physicalSlot.willBeOccupiedIndefinitely()).isFalse();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclareResourceRequirementServiceConnectionManagerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclareResourceRequirementServiceConnectionManagerTest.java
index 3c512fbf86c..74ddb093d53 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclareResourceRequirementServiceConnectionManagerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/DefaultDeclareResourceRequirementServiceConnectionManagerTest.java
@@ -26,10 +26,9 @@ import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.slots.ResourceRequirement;
import org.apache.flink.runtime.slots.ResourceRequirements;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nonnull;
@@ -40,18 +39,17 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link
DefaultDeclareResourceRequirementServiceConnectionManager}. */
-public class DefaultDeclareResourceRequirementServiceConnectionManagerTest
extends TestLogger {
+class DefaultDeclareResourceRequirementServiceConnectionManagerTest {
private final ManuallyTriggeredScheduledExecutorService scheduledExecutor =
new ManuallyTriggeredScheduledExecutorService();
private final JobID jobId = new JobID();
@Test
- public void testIgnoreDeclareResourceRequirementsIfNotConnected() {
+ void testIgnoreDeclareResourceRequirementsIfNotConnected() {
final DeclareResourceRequirementServiceConnectionManager
declareResourceRequirementServiceConnectionManager =
createResourceManagerConnectionManager();
@@ -61,7 +59,7 @@ public class
DefaultDeclareResourceRequirementServiceConnectionManagerTest exten
}
@Test
- public void testDeclareResourceRequirementsSendsRequirementsIfConnected() {
+ void testDeclareResourceRequirementsSendsRequirementsIfConnected() {
final DeclareResourceRequirementServiceConnectionManager
declareResourceRequirementServiceConnectionManager =
createResourceManagerConnectionManager();
@@ -78,12 +76,11 @@ public class
DefaultDeclareResourceRequirementServiceConnectionManagerTest exten
declareResourceRequirementServiceConnectionManager.declareResourceRequirements(
resourceRequirements);
- assertThat(declareResourceRequirementsFuture.join(),
is(resourceRequirements));
+
assertThat(declareResourceRequirementsFuture.join()).isEqualTo(resourceRequirements);
}
@Test
- public void testRetryDeclareResourceRequirementsIfTransmissionFailed()
- throws InterruptedException {
+ void testRetryDeclareResourceRequirementsIfTransmissionFailed() throws
InterruptedException {
final DeclareResourceRequirementServiceConnectionManager
declareResourceRequirementServiceConnectionManager =
createResourceManagerConnectionManager();
@@ -100,20 +97,19 @@ public class
DefaultDeclareResourceRequirementServiceConnectionManagerTest exten
scheduledExecutor.triggerNonPeriodicScheduledTasksWithRecursion();
- assertThat(
-
failingDeclareResourceRequirementsService.nextResourceRequirements(),
- is(resourceRequirements));
-
assertThat(failingDeclareResourceRequirementsService.hasResourceRequirements(),
is(false));
+
assertThat(failingDeclareResourceRequirementsService.nextResourceRequirements())
+ .isEqualTo(resourceRequirements);
+
assertThat(failingDeclareResourceRequirementsService.hasResourceRequirements()).isFalse();
}
@Test
- public void testDisconnectStopsSendingResourceRequirements() throws
InterruptedException {
+ void testDisconnectStopsSendingResourceRequirements() throws
InterruptedException {
runStopSendingResourceRequirementsTest(
DeclareResourceRequirementServiceConnectionManager::disconnect);
}
@Test
- public void testCloseStopsSendingResourceRequirements() throws
InterruptedException {
+ void testCloseStopsSendingResourceRequirements() throws
InterruptedException {
runStopSendingResourceRequirementsTest(
DeclareResourceRequirementServiceConnectionManager::close);
}
@@ -139,11 +135,11 @@ public class
DefaultDeclareResourceRequirementServiceConnectionManagerTest exten
testAction.accept(declareResourceRequirementServiceConnectionManager);
scheduledExecutor.triggerNonPeriodicScheduledTasksWithRecursion();
-
assertThat(declareResourceRequirementsService.hasResourceRequirements(),
is(false));
+
assertThat(declareResourceRequirementsService.hasResourceRequirements()).isFalse();
}
@Test
- public void testNewResourceRequirementsOverrideOldRequirements() throws
InterruptedException {
+ void testNewResourceRequirementsOverrideOldRequirements() throws
InterruptedException {
final DeclareResourceRequirementServiceConnectionManager
declareResourceRequirementServiceConnectionManager =
createResourceManagerConnectionManager();
@@ -169,10 +165,9 @@ public class
DefaultDeclareResourceRequirementServiceConnectionManagerTest exten
scheduledExecutor.triggerNonPeriodicScheduledTasksWithRecursion();
- assertThat(
-
failingDeclareResourceRequirementsService.nextResourceRequirements(),
- is(resourceRequirements2));
-
assertThat(failingDeclareResourceRequirementsService.hasResourceRequirements(),
is(false));
+
assertThat(failingDeclareResourceRequirementsService.nextResourceRequirements())
+ .isEqualTo(resourceRequirements2);
+
assertThat(failingDeclareResourceRequirementsService.hasResourceRequirements()).isFalse();
}
@Nonnull
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderResource.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderExtension.java
similarity index 86%
rename from
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderResource.java
rename to
flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderExtension.java
index 2275a6492e4..d717d61228e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderResource.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderExtension.java
@@ -24,7 +24,9 @@ import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
-import org.junit.rules.ExternalResource;
+import org.junit.jupiter.api.extension.AfterEachCallback;
+import org.junit.jupiter.api.extension.BeforeEachCallback;
+import org.junit.jupiter.api.extension.ExtensionContext;
import javax.annotation.Nonnull;
@@ -36,10 +38,10 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Function;
/**
- * {@code PhysicalSlotProviderResource} is used for testing different {@link
SlotSelectionStrategy}
+ * {@code PhysicalSlotProviderExtension} is used for testing different {@link
SlotSelectionStrategy}
* implementations on {@link PhysicalSlotProviderImpl}.
*/
-public class PhysicalSlotProviderResource extends ExternalResource {
+public class PhysicalSlotProviderExtension implements BeforeEachCallback,
AfterEachCallback {
private ScheduledExecutorService singleThreadScheduledExecutorService;
@@ -51,12 +53,12 @@ public class PhysicalSlotProviderResource extends
ExternalResource {
private PhysicalSlotProvider physicalSlotProvider;
- public PhysicalSlotProviderResource(@Nonnull SlotSelectionStrategy
slotSelectionStrategy) {
+ public PhysicalSlotProviderExtension(@Nonnull SlotSelectionStrategy
slotSelectionStrategy) {
this.slotSelectionStrategy = slotSelectionStrategy;
}
@Override
- protected void before() throws Throwable {
+ public void beforeEach(ExtensionContext context) throws Exception {
this.singleThreadScheduledExecutorService =
Executors.newSingleThreadScheduledExecutor();
this.mainThreadExecutor =
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
@@ -66,7 +68,7 @@ public class PhysicalSlotProviderResource extends
ExternalResource {
}
@Override
- protected void after() {
+ public void afterEach(ExtensionContext context) throws Exception {
CompletableFuture.runAsync(() -> slotPool.close(),
mainThreadExecutor).join();
singleThreadScheduledExecutorService.shutdown();
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest.java
index a29a698cbe0..ae2d2e48ad6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest.java
@@ -19,62 +19,60 @@
package org.apache.flink.runtime.jobmaster.slotpool;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
-import org.apache.flink.util.TestLogger;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@link PhysicalSlotProviderImpl} using {@link
* DefaultLocationPreferenceSlotSelectionStrategy}.
*/
-public class PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest
extends TestLogger {
+class PhysicalSlotProviderImplWithDefaultSlotSelectionStrategyTest {
- @Rule
- public final PhysicalSlotProviderResource physicalSlotProviderResource =
- new PhysicalSlotProviderResource(
+ @RegisterExtension
+ private final PhysicalSlotProviderExtension physicalSlotProviderExtension =
+ new PhysicalSlotProviderExtension(
LocationPreferenceSlotSelectionStrategy.createDefault());
@Test
- public void testSlotAllocationFulfilledWithAvailableSlots()
+ void testSlotAllocationFulfilledWithAvailableSlots()
throws InterruptedException, ExecutionException {
- PhysicalSlotRequest request =
physicalSlotProviderResource.createSimpleRequest();
-
physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
+ PhysicalSlotRequest request =
physicalSlotProviderExtension.createSimpleRequest();
+
physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
CompletableFuture<PhysicalSlotRequest.Result> slotFuture =
- physicalSlotProviderResource.allocateSlot(request);
+ physicalSlotProviderExtension.allocateSlot(request);
PhysicalSlotRequest.Result result = slotFuture.get();
- assertThat(result.getSlotRequestId(), is(request.getSlotRequestId()));
+
assertThat(result.getSlotRequestId()).isEqualTo(request.getSlotRequestId());
}
@Test
- public void testSlotAllocationFulfilledWithNewSlots()
- throws ExecutionException, InterruptedException {
+ void testSlotAllocationFulfilledWithNewSlots() throws ExecutionException,
InterruptedException {
final CompletableFuture<PhysicalSlotRequest.Result> slotFuture =
- physicalSlotProviderResource.allocateSlot(
- physicalSlotProviderResource.createSimpleRequest());
- assertThat(slotFuture.isDone(), is(false));
-
physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
+ physicalSlotProviderExtension.allocateSlot(
+ physicalSlotProviderExtension.createSimpleRequest());
+ assertThatFuture(slotFuture).isNotDone();
+
physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
slotFuture.get();
}
@Test
- public void
testIndividualBatchSlotRequestTimeoutCheckIsDisabledOnAllocatingNewSlots()
+ void
testIndividualBatchSlotRequestTimeoutCheckIsDisabledOnAllocatingNewSlots()
throws Exception {
DeclarativeSlotPoolBridge slotPool =
new DeclarativeSlotPoolBridgeBuilder()
-
.buildAndStart(physicalSlotProviderResource.getMainThreadExecutor());
- assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled(), is(true));
+
.buildAndStart(physicalSlotProviderExtension.getMainThreadExecutor());
+ assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled()).isTrue();
final PhysicalSlotProvider slotProvider =
new PhysicalSlotProviderImpl(
LocationPreferenceSlotSelectionStrategy.createDefault(), slotPool);
slotProvider.disableBatchSlotRequestTimeoutCheck();
- assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled(),
is(false));
+ assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled()).isFalse();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithSpreadOutStrategyTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithSpreadOutStrategyTest.java
index a1d2edfb9e3..0c9a808eaa6 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithSpreadOutStrategyTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotProviderImplWithSpreadOutStrategyTest.java
@@ -22,62 +22,59 @@ import
org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotProfileTestingUtils;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
-import org.apache.flink.util.TestLogger;
-import org.junit.Rule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.not;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Tests for {@link PhysicalSlotProviderImpl} using {@link
* EvenlySpreadOutLocationPreferenceSlotSelectionStrategy}.
*/
-public class PhysicalSlotProviderImplWithSpreadOutStrategyTest extends
TestLogger {
+class PhysicalSlotProviderImplWithSpreadOutStrategyTest {
- @Rule
- public PhysicalSlotProviderResource physicalSlotProviderResource =
- new PhysicalSlotProviderResource(
+ @RegisterExtension
+ private PhysicalSlotProviderExtension physicalSlotProviderExtension =
+ new PhysicalSlotProviderExtension(
LocationPreferenceSlotSelectionStrategy.createEvenlySpreadOut());
@Test
- public void testSlotAllocationFulfilledWithWorkloadSpreadOut()
+ void testSlotAllocationFulfilledWithWorkloadSpreadOut()
throws InterruptedException, ExecutionException {
- physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(
+ physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(
ResourceProfile.ANY, ResourceProfile.ANY, ResourceProfile.ANY,
ResourceProfile.ANY);
- physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(
+ physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(
ResourceProfile.ANY, ResourceProfile.ANY, ResourceProfile.ANY,
ResourceProfile.ANY);
- PhysicalSlotRequest request0 =
physicalSlotProviderResource.createSimpleRequest();
- PhysicalSlotRequest request1 =
physicalSlotProviderResource.createSimpleRequest();
+ PhysicalSlotRequest request0 =
physicalSlotProviderExtension.createSimpleRequest();
+ PhysicalSlotRequest request1 =
physicalSlotProviderExtension.createSimpleRequest();
PhysicalSlotRequest.Result result0 =
- physicalSlotProviderResource.allocateSlot(request0).get();
+ physicalSlotProviderExtension.allocateSlot(request0).get();
PhysicalSlotRequest.Result result1 =
- physicalSlotProviderResource.allocateSlot(request1).get();
+ physicalSlotProviderExtension.allocateSlot(request1).get();
- assertThat(
- result0.getPhysicalSlot().getTaskManagerLocation(),
- not(result1.getPhysicalSlot().getTaskManagerLocation()));
+ assertThat(result0.getPhysicalSlot().getTaskManagerLocation())
+
.isNotEqualTo(result1.getPhysicalSlot().getTaskManagerLocation());
}
@Test
- public void
testSlotAllocationFulfilledWithPreferredInputOverwrittingSpreadOut()
+ void testSlotAllocationFulfilledWithPreferredInputOverwrittingSpreadOut()
throws ExecutionException, InterruptedException {
- physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(
+ physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(
ResourceProfile.ANY, ResourceProfile.ANY);
- physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(
+ physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(
ResourceProfile.ANY, ResourceProfile.ANY);
- PhysicalSlotRequest request0 =
physicalSlotProviderResource.createSimpleRequest();
+ PhysicalSlotRequest request0 =
physicalSlotProviderExtension.createSimpleRequest();
PhysicalSlotRequest.Result result0 =
- physicalSlotProviderResource.allocateSlot(request0).get();
+ physicalSlotProviderExtension.allocateSlot(request0).get();
TaskManagerLocation preferredTaskManagerLocation =
result0.getPhysicalSlot().getTaskManagerLocation();
@@ -89,36 +86,34 @@ public class
PhysicalSlotProviderImplWithSpreadOutStrategyTest extends TestLogge
Collections.singleton(preferredTaskManagerLocation)),
false);
PhysicalSlotRequest.Result result1 =
- physicalSlotProviderResource.allocateSlot(request1).get();
+ physicalSlotProviderExtension.allocateSlot(request1).get();
- assertThat(
- result1.getPhysicalSlot().getTaskManagerLocation(),
- is(preferredTaskManagerLocation));
+ assertThat(result1.getPhysicalSlot().getTaskManagerLocation())
+ .isEqualTo(preferredTaskManagerLocation);
}
@Test
- public void testSlotAllocationFulfilledWithNewSlots()
- throws ExecutionException, InterruptedException {
+ void testSlotAllocationFulfilledWithNewSlots() throws ExecutionException,
InterruptedException {
final CompletableFuture<PhysicalSlotRequest.Result> slotFuture =
- physicalSlotProviderResource.allocateSlot(
- physicalSlotProviderResource.createSimpleRequest());
- assertThat(slotFuture.isDone(), is(false));
-
physicalSlotProviderResource.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
+ physicalSlotProviderExtension.allocateSlot(
+ physicalSlotProviderExtension.createSimpleRequest());
+ assertThatFuture(slotFuture).isNotDone();
+
physicalSlotProviderExtension.registerSlotOffersFromNewTaskExecutor(ResourceProfile.ANY);
slotFuture.get();
}
@Test
- public void
testIndividualBatchSlotRequestTimeoutCheckIsDisabledOnAllocatingNewSlots()
+ void
testIndividualBatchSlotRequestTimeoutCheckIsDisabledOnAllocatingNewSlots()
throws Exception {
DeclarativeSlotPoolBridge slotPool =
new DeclarativeSlotPoolBridgeBuilder()
-
.buildAndStart(physicalSlotProviderResource.getMainThreadExecutor());
- assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled(), is(true));
+
.buildAndStart(physicalSlotProviderExtension.getMainThreadExecutor());
+ assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled()).isTrue();
final PhysicalSlotProvider slotProvider =
new PhysicalSlotProviderImpl(
LocationPreferenceSlotSelectionStrategy.createEvenlySpreadOut(), slotPool);
slotProvider.disableBatchSlotRequestTimeoutCheck();
- assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled(),
is(false));
+ assertThat(slotPool.isBatchSlotRequestTimeoutCheckEnabled()).isFalse();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkCheckerImplTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkCheckerImplTest.java
index 663aad4c781..63abe84958d 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkCheckerImplTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkCheckerImplTest.java
@@ -25,13 +25,12 @@ import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.clock.ManualClock;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import java.util.HashSet;
import java.util.Set;
@@ -43,14 +42,14 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static
org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotTestUtils.createPhysicalSlot;
import static
org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotTestUtils.occupyPhysicalSlot;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for {@link PhysicalSlotRequestBulkCheckerImpl}. */
-public class PhysicalSlotRequestBulkCheckerImplTest extends TestLogger {
+class PhysicalSlotRequestBulkCheckerImplTest {
private static final Time TIMEOUT = Time.milliseconds(50L);
@@ -66,23 +65,23 @@ public class PhysicalSlotRequestBulkCheckerImplTest extends
TestLogger {
private Supplier<Set<SlotInfo>> slotsRetriever;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ private static void setupClass() {
singleThreadScheduledExecutorService =
Executors.newSingleThreadScheduledExecutor();
mainThreadExecutor =
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
singleThreadScheduledExecutorService);
}
- @AfterClass
- public static void teardownClass() {
+ @AfterAll
+ private static void teardownClass() {
if (singleThreadScheduledExecutorService != null) {
singleThreadScheduledExecutorService.shutdownNow();
}
}
- @Before
- public void setup() throws Exception {
+ @BeforeEach
+ private void setup() {
slots = new HashSet<>();
slotsRetriever = () -> new HashSet<>(slots);
bulkChecker = new PhysicalSlotRequestBulkCheckerImpl(slotsRetriever,
clock);
@@ -90,7 +89,7 @@ public class PhysicalSlotRequestBulkCheckerImplTest extends
TestLogger {
}
@Test
- public void testPendingBulkIsNotCancelled() throws InterruptedException,
ExecutionException {
+ void testPendingBulkIsNotCancelled() throws InterruptedException,
ExecutionException {
final CompletableFuture<SlotRequestId> cancellationFuture = new
CompletableFuture<>();
final PhysicalSlotRequestBulk bulk =
createPhysicalSlotRequestBulkWithCancellationFuture(
@@ -100,7 +99,7 @@ public class PhysicalSlotRequestBulkCheckerImplTest extends
TestLogger {
}
@Test
- public void testFulfilledBulkIsNotCancelled() throws InterruptedException,
ExecutionException {
+ void testFulfilledBulkIsNotCancelled() throws InterruptedException,
ExecutionException {
final CompletableFuture<SlotRequestId> cancellationFuture = new
CompletableFuture<>();
final PhysicalSlotRequestBulk bulk =
createPhysicalSlotRequestBulkWithCancellationFuture(
@@ -112,17 +111,18 @@ public class PhysicalSlotRequestBulkCheckerImplTest
extends TestLogger {
private static void checkNotCancelledAfter(CompletableFuture<?>
cancellationFuture, long milli)
throws ExecutionException, InterruptedException {
mainThreadExecutor.schedule(() -> {}, milli,
TimeUnit.MILLISECONDS).get();
- try {
- assertThat(cancellationFuture.isDone(), is(false));
- cancellationFuture.get(milli, TimeUnit.MILLISECONDS);
- fail("The future must not have been cancelled");
- } catch (TimeoutException e) {
- assertThat(cancellationFuture.isDone(), is(false));
- }
+ assertThatThrownBy(
+ () -> {
+ assertThatFuture(cancellationFuture).isNotDone();
+ cancellationFuture.get(milli,
TimeUnit.MILLISECONDS);
+ })
+ .withFailMessage("The future must not have been cancelled")
+ .isInstanceOf(TimeoutException.class);
+ assertThatFuture(cancellationFuture).isNotDone();
}
@Test
- public void testUnfulfillableBulkIsCancelled() {
+ void testUnfulfillableBulkIsCancelled() {
final CompletableFuture<SlotRequestId> cancellationFuture = new
CompletableFuture<>();
final SlotRequestId slotRequestId = new SlotRequestId();
final PhysicalSlotRequestBulk bulk =
@@ -130,11 +130,11 @@ public class PhysicalSlotRequestBulkCheckerImplTest
extends TestLogger {
cancellationFuture, slotRequestId);
bulkChecker.schedulePendingRequestBulkTimeoutCheck(bulk, TIMEOUT);
clock.advanceTime(TIMEOUT.toMilliseconds() + 1L,
TimeUnit.MILLISECONDS);
- assertThat(cancellationFuture.join(), is(slotRequestId));
+ assertThat(cancellationFuture.join()).isEqualTo(slotRequestId);
}
@Test
- public void testBulkFulfilledOnCheck() {
+ void testBulkFulfilledOnCheck() {
final SlotRequestId slotRequestId = new SlotRequestId();
final PhysicalSlotRequestBulkImpl bulk =
createPhysicalSlotRequestBulk(slotRequestId);
@@ -142,66 +142,62 @@ public class PhysicalSlotRequestBulkCheckerImplTest
extends TestLogger {
final PhysicalSlotRequestBulkWithTimestamp bulkWithTimestamp =
new PhysicalSlotRequestBulkWithTimestamp(bulk);
- assertThat(
- checkBulkTimeout(bulkWithTimestamp),
-
is(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.FULFILLED));
+ assertThat(checkBulkTimeout(bulkWithTimestamp))
+
.isEqualTo(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.FULFILLED);
}
@Test
- public void testBulkTimeoutOnCheck() {
+ void testBulkTimeoutOnCheck() {
final PhysicalSlotRequestBulkWithTimestamp bulk =
createPhysicalSlotRequestBulkWithTimestamp(new
SlotRequestId());
clock.advanceTime(TIMEOUT.toMilliseconds() + 1L,
TimeUnit.MILLISECONDS);
- assertThat(
- checkBulkTimeout(bulk),
-
is(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.TIMEOUT));
+ assertThat(checkBulkTimeout(bulk))
+
.isEqualTo(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.TIMEOUT);
}
@Test
- public void testBulkPendingOnCheckIfFulfillable() {
+ void testBulkPendingOnCheckIfFulfillable() {
final PhysicalSlotRequestBulkWithTimestamp bulk =
createPhysicalSlotRequestBulkWithTimestamp(new
SlotRequestId());
final PhysicalSlot slot = addOneSlot();
occupyPhysicalSlot(slot, false);
- assertThat(
- checkBulkTimeout(bulk),
-
is(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.PENDING));
+ assertThat(checkBulkTimeout(bulk))
+
.isEqualTo(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.PENDING);
}
@Test
- public void testBulkPendingOnCheckIfUnfulfillableButNotTimedOut() {
+ void testBulkPendingOnCheckIfUnfulfillableButNotTimedOut() {
final PhysicalSlotRequestBulkWithTimestamp bulk =
createPhysicalSlotRequestBulkWithTimestamp(new
SlotRequestId());
- assertThat(
- checkBulkTimeout(bulk),
-
is(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.PENDING));
+ assertThat(checkBulkTimeout(bulk))
+
.isEqualTo(PhysicalSlotRequestBulkCheckerImpl.TimeoutCheckResult.PENDING);
}
@Test
- public void testBulkFulfillable() {
+ void testBulkFulfillable() {
final PhysicalSlotRequestBulk bulk = createPhysicalSlotRequestBulk(new
SlotRequestId());
addOneSlot();
- assertThat(isFulfillable(bulk), is(true));
+ assertThat(isFulfillable(bulk)).isTrue();
}
@Test
- public void testBulkUnfulfillableWithInsufficientSlots() {
+ void testBulkUnfulfillableWithInsufficientSlots() {
final PhysicalSlotRequestBulk bulk =
createPhysicalSlotRequestBulk(new SlotRequestId(), new
SlotRequestId());
addOneSlot();
- assertThat(isFulfillable(bulk), is(false));
+ assertThat(isFulfillable(bulk)).isFalse();
}
@Test
- public void testBulkUnfulfillableWithSlotAlreadyAssignedToBulk() {
+ void testBulkUnfulfillableWithSlotAlreadyAssignedToBulk() {
final SlotRequestId slotRequestId = new SlotRequestId();
final PhysicalSlotRequestBulkImpl bulk =
createPhysicalSlotRequestBulk(slotRequestId, new
SlotRequestId());
@@ -210,11 +206,11 @@ public class PhysicalSlotRequestBulkCheckerImplTest
extends TestLogger {
bulk.markRequestFulfilled(slotRequestId, slot.getAllocationId());
- assertThat(isFulfillable(bulk), is(false));
+ assertThat(isFulfillable(bulk)).isFalse();
}
@Test
- public void testBulkUnfulfillableWithSlotOccupiedIndefinitely() {
+ void testBulkUnfulfillableWithSlotOccupiedIndefinitely() {
final PhysicalSlotRequestBulk bulk =
createPhysicalSlotRequestBulk(new SlotRequestId(), new
SlotRequestId());
@@ -223,11 +219,11 @@ public class PhysicalSlotRequestBulkCheckerImplTest
extends TestLogger {
occupyPhysicalSlot(slot1, true);
- assertThat(isFulfillable(bulk), is(false));
+ assertThat(isFulfillable(bulk)).isFalse();
}
@Test
- public void testBulkFulfillableWithSlotOccupiedTemporarily() {
+ void testBulkFulfillableWithSlotOccupiedTemporarily() {
final PhysicalSlotRequestBulk bulk =
createPhysicalSlotRequestBulk(new SlotRequestId(), new
SlotRequestId());
@@ -236,7 +232,7 @@ public class PhysicalSlotRequestBulkCheckerImplTest extends
TestLogger {
occupyPhysicalSlot(slot1, false);
- assertThat(isFulfillable(bulk), is(true));
+ assertThat(isFulfillable(bulk)).isTrue();
}
private PhysicalSlotRequestBulkWithTimestamp
createPhysicalSlotRequestBulkWithTimestamp(
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkWithTimestampTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkWithTimestampTest.java
index d68800b0e5f..8d2ec43c536 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkWithTimestampTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PhysicalSlotRequestBulkWithTimestampTest.java
@@ -18,34 +18,32 @@
package org.apache.flink.runtime.jobmaster.slotpool;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.clock.ManualClock;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.concurrent.TimeUnit;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link PhysicalSlotRequestBulkWithTimestamp}. */
-public class PhysicalSlotRequestBulkWithTimestampTest extends TestLogger {
+class PhysicalSlotRequestBulkWithTimestampTest {
private final ManualClock clock = new ManualClock();
@Test
- public void testMarkBulkUnfulfillable() {
+ void testMarkBulkUnfulfillable() {
final PhysicalSlotRequestBulkWithTimestamp bulk =
createPhysicalSlotRequestBulkWithTimestamp();
clock.advanceTime(456, TimeUnit.MILLISECONDS);
bulk.markUnfulfillable(clock.relativeTimeMillis());
- assertThat(bulk.getUnfulfillableSince(),
is(clock.relativeTimeMillis()));
+
assertThat(bulk.getUnfulfillableSince()).isEqualTo(clock.relativeTimeMillis());
}
@Test
- public void
testUnfulfillableTimestampWillNotBeOverriddenByFollowingUnfulfillableTimestamp()
{
+ void
testUnfulfillableTimestampWillNotBeOverriddenByFollowingUnfulfillableTimestamp()
{
final PhysicalSlotRequestBulkWithTimestamp bulk =
createPhysicalSlotRequestBulkWithTimestamp();
@@ -55,7 +53,7 @@ public class PhysicalSlotRequestBulkWithTimestampTest extends
TestLogger {
clock.advanceTime(456, TimeUnit.MILLISECONDS);
bulk.markUnfulfillable(clock.relativeTimeMillis());
- assertThat(bulk.getUnfulfillableSince(), is(unfulfillableSince));
+ assertThat(bulk.getUnfulfillableSince()).isEqualTo(unfulfillableSince);
}
private static PhysicalSlotRequestBulkWithTimestamp
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PreferredAllocationRequestSlotMatchingStrategyTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PreferredAllocationRequestSlotMatchingStrategyTest.java
index 711a7514259..cad72a71ced 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PreferredAllocationRequestSlotMatchingStrategyTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/PreferredAllocationRequestSlotMatchingStrategyTest.java
@@ -35,14 +35,14 @@ import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link PreferredAllocationRequestSlotMatchingStrategy}. */
@ExtendWith(TestLoggerExtension.class)
-public class PreferredAllocationRequestSlotMatchingStrategyTest {
+class PreferredAllocationRequestSlotMatchingStrategyTest {
/**
* This test ensures that new slots are matched against the preferred
allocationIds of the
* pending requests.
*/
@Test
- public void testNewSlotsAreMatchedAgainstPreferredAllocationIDs() throws
Exception {
+ void testNewSlotsAreMatchedAgainstPreferredAllocationIDs() {
final PreferredAllocationRequestSlotMatchingStrategy strategy =
PreferredAllocationRequestSlotMatchingStrategy.INSTANCE;
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SingleLogicalSlotTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SingleLogicalSlotTest.java
index 5187c9d5cf0..cdced79e907 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SingleLogicalSlotTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SingleLogicalSlotTest.java
@@ -32,10 +32,9 @@ import
org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Collection;
@@ -46,23 +45,22 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.sameInstance;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link SingleLogicalSlot} class. */
-public class SingleLogicalSlotTest extends TestLogger {
+class SingleLogicalSlotTest {
@Test
- public void testPayloadAssignment() {
+ void testPayloadAssignment() {
final SingleLogicalSlot singleLogicalSlot = createSingleLogicalSlot();
final DummyPayload dummyPayload1 = new DummyPayload();
final DummyPayload dummyPayload2 = new DummyPayload();
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload1),
is(true));
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload2),
is(false));
+ assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload1)).isTrue();
+
assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload2)).isFalse();
- assertThat(singleLogicalSlot.getPayload(),
sameInstance(dummyPayload1));
+ assertThat(singleLogicalSlot.getPayload()).isSameAs(dummyPayload1);
}
private SingleLogicalSlot createSingleLogicalSlot() {
@@ -84,33 +82,34 @@ public class SingleLogicalSlotTest extends TestLogger {
}
@Test
- public void testAlive() throws Exception {
+ void testAlive() throws Exception {
final SingleLogicalSlot singleLogicalSlot = createSingleLogicalSlot();
final DummyPayload dummyPayload = new DummyPayload();
- assertThat(singleLogicalSlot.isAlive(), is(true));
+ assertThat(singleLogicalSlot.isAlive()).isTrue();
+ ;
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload), is(true));
- assertThat(singleLogicalSlot.isAlive(), is(true));
+ assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload)).isTrue();
+ assertThat(singleLogicalSlot.isAlive()).isTrue();
final CompletableFuture<?> releaseFuture =
singleLogicalSlot.releaseSlot(new FlinkException("Test
exception"));
- assertThat(singleLogicalSlot.isAlive(), is(false));
+ assertThat(singleLogicalSlot.isAlive()).isFalse();
releaseFuture.get();
- assertThat(singleLogicalSlot.isAlive(), is(false));
+ assertThat(singleLogicalSlot.isAlive()).isFalse();
}
@Test
- public void testPayloadAssignmentAfterRelease() {
+ void testPayloadAssignmentAfterRelease() {
final SingleLogicalSlot singleLogicalSlot = createSingleLogicalSlot();
final DummyPayload dummyPayload = new DummyPayload();
singleLogicalSlot.releaseSlot(new FlinkException("Test exception"));
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload),
is(false));
+ assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload)).isFalse();
}
/**
@@ -118,7 +117,7 @@ public class SingleLogicalSlotTest extends TestLogger {
* to reach a terminal state.
*/
@Test
- public void testAllocatedSlotRelease() {
+ void testAllocatedSlotRelease() {
final CompletableFuture<LogicalSlot> returnSlotFuture = new
CompletableFuture<>();
final WaitingSlotOwner waitingSlotOwner =
new WaitingSlotOwner(returnSlotFuture, new
CompletableFuture<>());
@@ -129,19 +128,19 @@ public class SingleLogicalSlotTest extends TestLogger {
final ManualTestingPayload dummyPayload =
new ManualTestingPayload(failFuture, terminalStateFuture);
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload), is(true));
+ assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload)).isTrue();
singleLogicalSlot.release(new FlinkException("Test exception"));
- assertThat(failFuture.isDone(), is(true));
+ assertThatFuture(failFuture).isDone();
// we don't require the logical slot to return to the owner because
// the release call should only come from the owner
- assertThat(returnSlotFuture.isDone(), is(false));
+ assertThatFuture(returnSlotFuture).isNotDone();
}
/** Tests that the slot release is only signaled after the owner has taken
it back. */
@Test
- public void testSlotRelease() {
+ void testSlotRelease() {
final CompletableFuture<LogicalSlot> returnedSlotFuture = new
CompletableFuture<>();
final CompletableFuture<Boolean> returnSlotResponseFuture = new
CompletableFuture<>();
final WaitingSlotOwner waitingSlotOwner =
@@ -153,22 +152,22 @@ public class SingleLogicalSlotTest extends TestLogger {
final SingleLogicalSlot singleLogicalSlot =
createSingleLogicalSlot(waitingSlotOwner);
- assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload), is(true));
+ assertThat(singleLogicalSlot.tryAssignPayload(dummyPayload)).isTrue();
final CompletableFuture<?> releaseFuture =
singleLogicalSlot.releaseSlot(new FlinkException("Test
exception"));
- assertThat(releaseFuture.isDone(), is(false));
- assertThat(returnedSlotFuture.isDone(), is(false));
- assertThat(failFuture.isDone(), is(true));
+ assertThatFuture(releaseFuture).isNotDone();
+ assertThatFuture(returnedSlotFuture).isNotDone();
+ assertThatFuture(failFuture).isDone();
terminalStateFuture.complete(null);
- assertThat(returnedSlotFuture.isDone(), is(true));
+ assertThatFuture(returnedSlotFuture).isDone();
returnSlotResponseFuture.complete(true);
- assertThat(releaseFuture.isDone(), is(true));
+ assertThatFuture(releaseFuture).isDone();
}
/**
@@ -176,7 +175,7 @@ public class SingleLogicalSlotTest extends TestLogger {
* return of the slot once.
*/
@Test
- public void testConcurrentReleaseOperations() throws Exception {
+ void testConcurrentReleaseOperations() throws Exception {
final CountingSlotOwner countingSlotOwner = new CountingSlotOwner();
final CountingFailPayload countingFailPayload = new
CountingFailPayload();
final SingleLogicalSlot singleLogicalSlot =
createSingleLogicalSlot(countingSlotOwner);
@@ -212,8 +211,8 @@ public class SingleLogicalSlotTest extends TestLogger {
releaseOperationsFuture.get();
- assertThat(countingSlotOwner.getReleaseCount(), is(1));
- assertThat(countingFailPayload.getFailCount(), is(1));
+ assertThat(countingSlotOwner.getReleaseCount()).isOne();
+ assertThat(countingFailPayload.getFailCount()).isOne();
} finally {
executorService.shutdownNow();
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBatchSlotRequestTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBatchSlotRequestTest.java
index 718cf87c1c8..138b3809221 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBatchSlotRequestTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolBatchSlotRequestTest.java
@@ -19,7 +19,6 @@
package org.apache.flink.runtime.jobmaster.slotpool;
import org.apache.flink.api.common.time.Time;
-import org.apache.flink.core.testutils.FlinkMatchers;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
@@ -28,14 +27,13 @@ import
org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import
org.apache.flink.runtime.resourcemanager.exceptions.UnfulfillableSlotRequestException;
import
org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway;
import org.apache.flink.util.FlinkException;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.clock.Clock;
import org.apache.flink.util.clock.ManualClock;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import javax.annotation.Nullable;
@@ -49,12 +47,11 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.fail;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for batch slot requests. */
-public class SlotPoolBatchSlotRequestTest extends TestLogger {
+class SlotPoolBatchSlotRequestTest {
private static final ResourceProfile resourceProfile =
ResourceProfile.fromResources(1.0, 1024);
public static final CompletableFuture[] COMPLETABLE_FUTURES_EMPTY_ARRAY =
@@ -62,16 +59,16 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
private static ScheduledExecutorService
singleThreadScheduledExecutorService;
private static ComponentMainThreadExecutor mainThreadExecutor;
- @BeforeClass
- public static void setupClass() {
+ @BeforeAll
+ private static void setupClass() {
singleThreadScheduledExecutorService =
Executors.newSingleThreadScheduledExecutor();
mainThreadExecutor =
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
singleThreadScheduledExecutorService);
}
- @AfterClass
- public static void teardownClass() {
+ @AfterAll
+ private static void teardownClass() {
if (singleThreadScheduledExecutorService != null) {
singleThreadScheduledExecutorService.shutdownNow();
}
@@ -81,19 +78,17 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
* Tests that a batch slot request fails if there is no slot which can
fulfill the slot request.
*/
@Test
- public void testPendingBatchSlotRequestTimeout() throws Exception {
+ void testPendingBatchSlotRequestTimeout() throws Exception {
try (final SlotPool slotPool =
createAndSetUpSlotPool(mainThreadExecutor, null,
Time.milliseconds(2L))) {
final CompletableFuture<PhysicalSlot> slotFuture =
SlotPoolUtils.requestNewAllocatedBatchSlot(
slotPool, mainThreadExecutor,
ResourceProfile.UNKNOWN);
- try {
- slotFuture.get();
- fail("Expected that slot future times out.");
- } catch (ExecutionException ee) {
- assertThat(ee,
FlinkMatchers.containsCause(TimeoutException.class));
- }
+ assertThatThrownBy(slotFuture::get)
+ .withFailMessage("Expected that slot future times out.")
+ .isInstanceOf(ExecutionException.class)
+ .hasRootCauseInstanceOf(TimeoutException.class);
}
}
@@ -102,7 +97,7 @@ public class SlotPoolBatchSlotRequestTest extends TestLogger
{
* fulfills the requested {@link ResourceProfile}.
*/
@Test
- public void
testPendingBatchSlotRequestDoesNotTimeoutIfFulfillingSlotExists() throws
Exception {
+ void testPendingBatchSlotRequestDoesNotTimeoutIfFulfillingSlotExists()
throws Exception {
final Time batchSlotTimeout = Time.milliseconds(2L);
final ManualClock clock = new ManualClock();
@@ -128,7 +123,7 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
slotPool, mainThreadExecutor, clock, batchSlotTimeout);
for (CompletableFuture<PhysicalSlot> slotFuture : slotFutures) {
- assertThat(slotFuture.isDone(), is(false));
+ assertThatFuture(slotFuture).isNotDone();
}
}
}
@@ -138,7 +133,7 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
* exceptions other than {@link UnfulfillableSlotRequestException}.
*/
@Test
- public void
testPendingBatchSlotRequestDoesNotFailIfResourceDeclaringFails() throws
Exception {
+ void testPendingBatchSlotRequestDoesNotFailIfResourceDeclaringFails()
throws Exception {
final TestingResourceManagerGateway testingResourceManagerGateway =
new TestingResourceManagerGateway();
testingResourceManagerGateway.setDeclareRequiredResourcesFunction(
@@ -154,7 +149,7 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
SlotPoolUtils.requestNewAllocatedBatchSlot(
slotPool, mainThreadExecutor, resourceProfile);
- assertThat(slotFuture,
FlinkMatchers.willNotComplete(Duration.ofMillis(50L)));
+
assertThatFuture(slotFuture).willNotCompleteWithin(Duration.ofMillis(50L));
}
}
@@ -163,7 +158,7 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
* released.
*/
@Test
- public void testPendingBatchSlotRequestTimeoutAfterSlotRelease() throws
Exception {
+ void testPendingBatchSlotRequestTimeoutAfterSlotRelease() throws Exception
{
final ManualClock clock = new ManualClock();
final Time batchSlotTimeout = Time.milliseconds(10000L);
@@ -191,10 +186,10 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
advanceTimeAndTriggerCheckBatchSlotTimeout(
slotPool, mainThreadExecutor, clock, batchSlotTimeout);
- assertThat(
-
CompletableFuture.anyOf(slotFutures.toArray(COMPLETABLE_FUTURES_EMPTY_ARRAY))
- .isDone(),
- is(false));
+ assertThatFuture(
+ CompletableFuture.anyOf(
+
slotFutures.toArray(COMPLETABLE_FUTURES_EMPTY_ARRAY)))
+ .isNotDone();
SlotPoolUtils.releaseTaskManager(slotPool, mainThreadExecutor,
taskManagerResourceId);
@@ -202,14 +197,12 @@ public class SlotPoolBatchSlotRequestTest extends
TestLogger {
slotPool, mainThreadExecutor, clock, batchSlotTimeout);
for (CompletableFuture<PhysicalSlot> slotFuture : slotFutures) {
- assertThat(slotFuture.isCompletedExceptionally(), is(true));
-
- try {
- slotFuture.get();
- fail("Expected that the slot future times out.");
- } catch (ExecutionException ee) {
- assertThat(ee,
FlinkMatchers.containsCause(TimeoutException.class));
- }
+ assertThatFuture(slotFuture).isCompletedExceptionally();
+
+ assertThatThrownBy(slotFuture::get)
+ .withFailMessage("Expected that the slot future times
out.")
+ .isInstanceOf(ExecutionException.class)
+ .hasRootCauseInstanceOf(TimeoutException.class);
}
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
index 0406e9b34bd..a746a216348 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolInteractionsTest.java
@@ -23,37 +23,34 @@ import
org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import
org.apache.flink.runtime.executiongraph.TestingComponentMainThreadExecutor;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
import org.apache.flink.runtime.testutils.CommonTestUtils;
-import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLogger;
-import org.junit.ClassRule;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link DeclarativeSlotPoolBridge} interactions. */
-public class SlotPoolInteractionsTest extends TestLogger {
+class SlotPoolInteractionsTest {
private static final Time fastTimeout = Time.milliseconds(1L);
- @ClassRule
- public static final TestingComponentMainThreadExecutor.Resource
EXECUTOR_RESOURCE =
- new TestingComponentMainThreadExecutor.Resource(10L);
+ @RegisterExtension
+ private static final TestingComponentMainThreadExecutor.Extension
EXECUTOR_EXTENSION =
+ new TestingComponentMainThreadExecutor.Extension(10L);
private final TestingComponentMainThreadExecutor testMainThreadExecutor =
- EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor();
+ EXECUTOR_EXTENSION.getComponentMainThreadTestExecutor();
// ------------------------------------------------------------------------
// tests
// ------------------------------------------------------------------------
@Test
- public void testSlotAllocationNoResourceManager() throws Exception {
+ void testSlotAllocationNoResourceManager() throws Exception {
try (SlotPool pool = createAndSetUpSlotPoolWithoutResourceManager()) {
@@ -65,17 +62,15 @@ public class SlotPoolInteractionsTest extends TestLogger {
ResourceProfile.UNKNOWN,
fastTimeout));
- try {
- future.get();
- fail("We expected an ExecutionException.");
- } catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e)
instanceof TimeoutException);
- }
+ assertThatThrownBy(future::get)
+ .withFailMessage("We expected an ExecutionException.")
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
}
}
@Test
- public void testCancelSlotAllocationWithoutResourceManager() throws
Exception {
+ void testCancelSlotAllocationWithoutResourceManager() throws Exception {
try (DeclarativeSlotPoolBridge pool =
createAndSetUpSlotPoolWithoutResourceManager()) {
@@ -87,12 +82,10 @@ public class SlotPoolInteractionsTest extends TestLogger {
ResourceProfile.UNKNOWN,
fastTimeout));
- try {
- future.get();
- fail("We expected a TimeoutException.");
- } catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e)
instanceof TimeoutException);
- }
+ assertThatThrownBy(future::get)
+ .withFailMessage("We expected a TimeoutException.")
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
CommonTestUtils.waitUntilCondition(() ->
pool.getNumPendingRequests() == 0);
}
@@ -100,7 +93,7 @@ public class SlotPoolInteractionsTest extends TestLogger {
/** Tests that a slot allocation times out wrt to the specified time out.
*/
@Test
- public void testSlotAllocationTimeout() throws Exception {
+ void testSlotAllocationTimeout() throws Exception {
try (DeclarativeSlotPoolBridge pool = createAndSetUpSlotPool()) {
@@ -112,12 +105,10 @@ public class SlotPoolInteractionsTest extends TestLogger {
ResourceProfile.UNKNOWN,
fastTimeout));
- try {
- future.get();
- fail("We expected a TimeoutException.");
- } catch (ExecutionException e) {
- assertTrue(ExceptionUtils.stripExecutionException(e)
instanceof TimeoutException);
- }
+ assertThatThrownBy(future::get)
+ .withFailMessage("We expected a TimeoutException.")
+ .isInstanceOf(ExecutionException.class)
+ .hasCauseInstanceOf(TimeoutException.class);
CommonTestUtils.waitUntilCondition(() ->
pool.getNumPendingRequests() == 0);
}
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
index 7e9fe1ba6ff..f430beeb10c 100644
---
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
@@ -43,8 +43,7 @@ import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static java.util.stream.Collectors.reducing;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.is;
+import static org.assertj.core.api.Assertions.assertThat;
/** Testing utility functions for the {@link SlotPool}. */
public class SlotPoolUtils {
@@ -129,7 +128,7 @@ public class SlotPoolUtils {
taskManagerLocation,
taskManagerGateway, slotOffers);
if (assertAllSlotsAreAccepted) {
- assertThat(acceptedOffers, is(slotOffers));
+
assertThat(acceptedOffers).isEqualTo(slotOffers);
}
},
mainThreadExecutor)
