You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/08/30 14:00:33 UTC
[flink] branch master updated: [FLINK-26341][zookeeper] Upgrade all tests related to ZooKeeperTestEnvironment to JUnit5
This is an automated email from the ASF dual-hosted git repository.
mapohl 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 8b2be3f88bb [FLINK-26341][zookeeper] Upgrade all tests related to ZooKeeperTestEnvironment to JUnit5
8b2be3f88bb is described below
commit 8b2be3f88bbbc60d7540962255e07f48f705e0ee
Author: Rui Fan <19...@gmail.com>
AuthorDate: Wed Aug 30 22:00:25 2023 +0800
[FLINK-26341][zookeeper] Upgrade all tests related to ZooKeeperTestEnvironment to JUnit5
This also removes `ZooKeeperTestEnvironment` as it becomes obsolete
---
.../checkpoint/CompletedCheckpointStoreTest.java | 85 ++++----
.../StandaloneCompletedCheckpointStoreTest.java | 29 ++-
.../ZooKeeperCheckpointIDCounterITCase.java | 74 +++----
.../ZooKeeperCompletedCheckpointStoreITCase.java | 222 +++++++++------------
.../jobmanager/ZooKeeperJobGraphsStoreITCase.java | 93 +++++----
.../zookeeper/ZooKeeperTestEnvironment.java | 164 ---------------
.../JobManagerHAProcessFailureRecoveryITCase.java | 101 ++++------
7 files changed, 261 insertions(+), 507 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
index 0eeb35ff8a7..06e6ff4146e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/CompletedCheckpointStoreTest.java
@@ -24,11 +24,9 @@ import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.SharedStateRegistryImpl;
import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.Executors;
-import org.junit.Assert;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Collection;
import java.util.HashMap;
@@ -39,12 +37,11 @@ import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
/** Test for basic {@link CompletedCheckpointStore} contract. */
-public abstract class CompletedCheckpointStoreTest extends TestLogger {
+abstract class CompletedCheckpointStoreTest {
/** Creates the {@link CompletedCheckpointStore} implementation to be tested. */
protected abstract CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
@@ -59,20 +56,21 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
// ---------------------------------------------------------------------------------------------
/** Tests that at least one checkpoint needs to be retained. */
- @Test(expected = Exception.class)
- public void testExceptionOnNoRetainedCheckpoints() throws Exception {
- createRecoveredCompletedCheckpointStore(0);
+ @Test
+ void testExceptionOnNoRetainedCheckpoints() {
+ assertThatExceptionOfType(Exception.class)
+ .isThrownBy(() -> createRecoveredCompletedCheckpointStore(0));
}
/** Tests adding and getting a checkpoint. */
@Test
- public void testAddAndGetLatestCheckpoint() throws Exception {
+ void testAddAndGetLatestCheckpoint() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(4);
// Empty state
- assertEquals(0, checkpoints.getNumberOfRetainedCheckpoints());
- assertEquals(0, checkpoints.getAllCheckpoints().size());
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isZero();
+ assertThat(checkpoints.getAllCheckpoints()).isEmpty();
TestCompletedCheckpoint[] expected =
new TestCompletedCheckpoint[] {
@@ -83,12 +81,12 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
// Add and get latest
checkpoints.addCheckpointAndSubsumeOldestOne(
expected[0], new CheckpointsCleaner(), () -> {});
- assertEquals(1, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isOne();
verifyCheckpoint(expected[0], checkpoints.getLatestCheckpoint());
checkpoints.addCheckpointAndSubsumeOldestOne(
expected[1], new CheckpointsCleaner(), () -> {});
- assertEquals(2, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isEqualTo(2);
verifyCheckpoint(expected[1], checkpoints.getLatestCheckpoint());
}
@@ -97,7 +95,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
* correct class loader).
*/
@Test
- public void testAddCheckpointMoreThanMaxRetained() throws Exception {
+ void testAddCheckpointMoreThanMaxRetained() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(1);
CheckpointsCleaner checkpointsCleaner = new CheckpointsCleaner();
@@ -111,15 +109,15 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
// Add checkpoints
checkpoints.addCheckpointAndSubsumeOldestOne(expected[0], checkpointsCleaner, () -> {});
- assertEquals(1, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isOne();
for (int i = 1; i < expected.length; i++) {
checkpoints.addCheckpointAndSubsumeOldestOne(expected[i], checkpointsCleaner, () -> {});
// The ZooKeeper implementation discards asynchronously
expected[i - 1].awaitDiscard();
- assertTrue(expected[i - 1].isDiscarded());
- assertEquals(1, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(expected[i - 1].isDiscarded()).isTrue();
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isOne();
}
}
@@ -133,17 +131,17 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
* </ul>
*/
@Test
- public void testEmptyState() throws Exception {
+ void testEmptyState() throws Exception {
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(1);
- assertNull(checkpoints.getLatestCheckpoint());
- assertEquals(0, checkpoints.getAllCheckpoints().size());
- assertEquals(0, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(checkpoints.getLatestCheckpoint()).isNull();
+ assertThat(checkpoints.getAllCheckpoints()).isEmpty();
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isZero();
}
/** Tests that all added checkpoints are returned. */
@Test
- public void testGetAllCheckpoints() throws Exception {
+ void testGetAllCheckpoints() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(4);
@@ -162,16 +160,12 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
List<CompletedCheckpoint> actual = checkpoints.getAllCheckpoints();
- assertEquals(expected.length, actual.size());
-
- for (int i = 0; i < expected.length; i++) {
- assertEquals(expected[i], actual.get(i));
- }
+ assertThat(actual).hasSameSizeAs(expected).containsExactly(expected);
}
/** Tests that all checkpoints are discarded (using the correct class loader). */
@Test
- public void testDiscardAllCheckpoints() throws Exception {
+ void testDiscardAllCheckpoints() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(4);
@@ -191,31 +185,31 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
checkpoints.shutdown(JobStatus.FINISHED, new CheckpointsCleaner());
// Empty state
- assertNull(checkpoints.getLatestCheckpoint());
- assertEquals(0, checkpoints.getAllCheckpoints().size());
- assertEquals(0, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(checkpoints.getLatestCheckpoint()).isNull();
+ assertThat(checkpoints.getAllCheckpoints()).isEmpty();
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isZero();
// All have been discarded
for (TestCompletedCheckpoint checkpoint : expected) {
// The ZooKeeper implementation discards asynchronously
checkpoint.awaitDiscard();
- assertTrue(checkpoint.isDiscarded());
+ assertThat(checkpoint.isDiscarded()).isTrue();
}
}
@Test
- public void testAcquireLatestCompletedCheckpointId() throws Exception {
+ void testAcquireLatestCompletedCheckpointId() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(1);
- assertEquals(0, checkpoints.getLatestCheckpointId());
+ assertThat(checkpoints.getLatestCheckpointId()).isZero();
checkpoints.addCheckpointAndSubsumeOldestOne(
createCheckpoint(2, sharedStateRegistry), new CheckpointsCleaner(), () -> {});
- assertEquals(2, checkpoints.getLatestCheckpointId());
+ assertThat(checkpoints.getLatestCheckpointId()).isEqualTo(2);
checkpoints.addCheckpointAndSubsumeOldestOne(
createCheckpoint(4, sharedStateRegistry), new CheckpointsCleaner(), () -> {});
- assertEquals(4, checkpoints.getLatestCheckpointId());
+ assertThat(checkpoints.getLatestCheckpointId()).isEqualTo(4);
}
// ---------------------------------------------------------------------------------------------
@@ -250,30 +244,29 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
return new TestCompletedCheckpoint(new JobID(), id, 0, operatorGroupState, props);
}
- protected void verifyCheckpointRegistered(
- Collection<OperatorState> operatorStates, SharedStateRegistry registry) {
+ protected void verifyCheckpointRegistered(Collection<OperatorState> operatorStates) {
for (OperatorState operatorState : operatorStates) {
for (OperatorSubtaskState subtaskState : operatorState.getStates()) {
- Assert.assertTrue(((TestOperatorSubtaskState) subtaskState).registered);
+ assertThat(((TestOperatorSubtaskState) subtaskState).registered).isTrue();
}
}
}
public static void verifyCheckpointDiscarded(TestCompletedCheckpoint completedCheckpoint) {
- assertTrue(completedCheckpoint.isDiscarded());
+ assertThat(completedCheckpoint.isDiscarded()).isTrue();
verifyCheckpointDiscarded(completedCheckpoint.getOperatorStates().values());
}
protected static void verifyCheckpointDiscarded(Collection<OperatorState> operatorStates) {
for (OperatorState operatorState : operatorStates) {
for (OperatorSubtaskState subtaskState : operatorState.getStates()) {
- Assert.assertTrue(((TestOperatorSubtaskState) subtaskState).discarded);
+ assertThat(((TestOperatorSubtaskState) subtaskState).discarded).isTrue();
}
}
}
private void verifyCheckpoint(CompletedCheckpoint expected, CompletedCheckpoint actual) {
- assertEquals(expected, actual);
+ assertThat(actual).isEqualTo(expected);
}
/**
@@ -379,7 +372,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
@Override
public void discardState() {
super.discardState();
- Assert.assertFalse(discarded);
+ assertThat(discarded).isFalse();
discarded = true;
registered = false;
}
@@ -388,7 +381,7 @@ public abstract class CompletedCheckpointStoreTest extends TestLogger {
public void registerSharedStates(
SharedStateRegistry sharedStateRegistry, long checkpointID) {
super.registerSharedStates(sharedStateRegistry, checkpointID);
- Assert.assertFalse(discarded);
+ assertThat(discarded).isFalse();
registered = true;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
index f30df4dfd9d..52873f38c9e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/StandaloneCompletedCheckpointStoreTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.runtime.state.SharedStateRegistryImpl;
import org.apache.flink.runtime.state.testutils.TestCompletedCheckpointStorageLocation;
import org.apache.flink.util.concurrent.Executors;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.Collection;
import java.util.Collections;
@@ -33,11 +33,10 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import static org.apache.flink.runtime.checkpoint.CheckpointRetentionPolicy.NEVER_RETAIN_AFTER_TERMINATION;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for basic {@link CompletedCheckpointStore} contract. */
-public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointStoreTest {
+class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointStoreTest {
@Override
protected CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
@@ -47,19 +46,19 @@ public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointS
/** Tests that shutdown discards all checkpoints. */
@Test
- public void testShutdownDiscardsCheckpoints() throws Exception {
+ void testShutdownDiscardsCheckpoints() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore store = createRecoveredCompletedCheckpointStore(1);
TestCompletedCheckpoint checkpoint = createCheckpoint(0, sharedStateRegistry);
Collection<OperatorState> operatorStates = checkpoint.getOperatorStates().values();
store.addCheckpointAndSubsumeOldestOne(checkpoint, new CheckpointsCleaner(), () -> {});
- assertEquals(1, store.getNumberOfRetainedCheckpoints());
- verifyCheckpointRegistered(operatorStates, sharedStateRegistry);
+ assertThat(store.getNumberOfRetainedCheckpoints()).isOne();
+ verifyCheckpointRegistered(operatorStates);
store.shutdown(JobStatus.FINISHED, new CheckpointsCleaner());
- assertEquals(0, store.getNumberOfRetainedCheckpoints());
- assertTrue(checkpoint.isDiscarded());
+ assertThat(store.getNumberOfRetainedCheckpoints()).isZero();
+ assertThat(checkpoint.isDiscarded()).isTrue();
verifyCheckpointDiscarded(operatorStates);
}
@@ -68,19 +67,19 @@ public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointS
* recovery mode).
*/
@Test
- public void testSuspendDiscardsCheckpoints() throws Exception {
+ void testSuspendDiscardsCheckpoints() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore store = createRecoveredCompletedCheckpointStore(1);
TestCompletedCheckpoint checkpoint = createCheckpoint(0, sharedStateRegistry);
Collection<OperatorState> taskStates = checkpoint.getOperatorStates().values();
store.addCheckpointAndSubsumeOldestOne(checkpoint, new CheckpointsCleaner(), () -> {});
- assertEquals(1, store.getNumberOfRetainedCheckpoints());
- verifyCheckpointRegistered(taskStates, sharedStateRegistry);
+ assertThat(store.getNumberOfRetainedCheckpoints()).isOne();
+ verifyCheckpointRegistered(taskStates);
store.shutdown(JobStatus.SUSPENDED, new CheckpointsCleaner());
- assertEquals(0, store.getNumberOfRetainedCheckpoints());
- assertTrue(checkpoint.isDiscarded());
+ assertThat(store.getNumberOfRetainedCheckpoints()).isZero();
+ assertThat(checkpoint.isDiscarded()).isTrue();
verifyCheckpointDiscarded(taskStates);
}
@@ -89,7 +88,7 @@ public class StandaloneCompletedCheckpointStoreTest extends CompletedCheckpointS
* (i.e., there exists an exception thrown by the method).
*/
@Test
- public void testAddCheckpointWithFailedRemove() throws Exception {
+ void testAddCheckpointWithFailedRemove() throws Exception {
final int numCheckpointsToRetain = 1;
CompletedCheckpointStore store =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounterITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounterITCase.java
index 4e9c0d4e554..46f4448a546 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounterITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCheckpointIDCounterITCase.java
@@ -19,21 +19,21 @@
package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.core.testutils.FlinkAssertions;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.concurrent.ExecutionException;
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Unit tests for the {@link ZooKeeperCheckpointIDCounter}. The tests are inherited from the test
@@ -41,32 +41,28 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
*/
class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
- private static ZooKeeperTestEnvironment zookeeper;
+ private final ZooKeeperExtension zooKeeperExtension = new ZooKeeperExtension();
- @BeforeEach
- void setup() {
- zookeeper = new ZooKeeperTestEnvironment(1);
- }
+ @RegisterExtension
+ final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(zooKeeperExtension);
- @AfterEach
- void tearDown() throws Exception {
- cleanAndStopZooKeeperIfRunning();
- }
+ @RegisterExtension
+ final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
- private void cleanAndStopZooKeeperIfRunning() throws Exception {
- if (zookeeper.getClient().isStarted()) {
- zookeeper.deleteAll();
- zookeeper.shutdown();
- }
+ private CuratorFramework getZooKeeperClient() {
+ return zooKeeperExtension.getZooKeeperClient(
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
}
/** Tests that counter node is removed from ZooKeeper after shutdown. */
@Test
- public void testShutdownRemovesState() throws Exception {
+ void testShutdownRemovesState() throws Exception {
ZooKeeperCheckpointIDCounter counter = createCheckpointIdCounter();
counter.start();
- CuratorFramework client = zookeeper.getClient();
+ CuratorFramework client = getZooKeeperClient();
assertThat(client.checkExists().forPath(counter.getPath())).isNotNull();
counter.shutdown(JobStatus.FINISHED).join();
@@ -74,11 +70,11 @@ class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
}
@Test
- public void testIdempotentShutdown() throws Exception {
+ void testIdempotentShutdown() throws Exception {
ZooKeeperCheckpointIDCounter counter = createCheckpointIdCounter();
counter.start();
- CuratorFramework client = zookeeper.getClient();
+ CuratorFramework client = getZooKeeperClient();
counter.shutdown(JobStatus.FINISHED).join();
// shutdown shouldn't fail due to missing path
@@ -87,25 +83,25 @@ class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
}
@Test
- public void testShutdownWithFailureDueToMissingConnection() throws Exception {
+ void testShutdownWithFailureDueToMissingConnection() throws Exception {
ZooKeeperCheckpointIDCounter counter = createCheckpointIdCounter();
counter.start();
- cleanAndStopZooKeeperIfRunning();
+ zooKeeperExtension.close();
- assertThatThrownBy(() -> counter.shutdown(JobStatus.FINISHED).get())
+ assertThatFuture(counter.shutdown(JobStatus.FINISHED))
.as("The shutdown should fail because of the client connection being dropped.")
- .isInstanceOf(ExecutionException.class)
- .hasCauseExactlyInstanceOf(IllegalStateException.class);
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class);
}
@Test
- public void testShutdownWithFailureDueToExistingChildNodes() throws Exception {
+ void testShutdownWithFailureDueToExistingChildNodes() throws Exception {
final ZooKeeperCheckpointIDCounter counter = createCheckpointIdCounter();
counter.start();
final CuratorFramework client =
- ZooKeeperUtils.useNamespaceAndEnsurePath(zookeeper.getClient(), "/");
+ ZooKeeperUtils.useNamespaceAndEnsurePath(getZooKeeperClient(), "/");
final String counterNodePath = ZooKeeperUtils.generateZookeeperPath(counter.getPath());
final String childNodePath =
ZooKeeperUtils.generateZookeeperPath(
@@ -116,16 +112,12 @@ class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
ZooKeeperUtils.generateZookeeperPath(client.getNamespace(), counterNodePath);
final Throwable expectedRootCause =
KeeperException.create(KeeperException.Code.NOTEMPTY, namespacedCounterNodePath);
- assertThatThrownBy(() -> counter.shutdown(JobStatus.FINISHED).get())
+ assertThatFuture(counter.shutdown(JobStatus.FINISHED))
.as(
"The shutdown should fail because of a child node being present and the shutdown not performing an explicit recursive deletion.")
- .isInstanceOf(ExecutionException.class)
- .extracting(FlinkAssertions::chainOfCauses, FlinkAssertions.STREAM_THROWABLE)
- .anySatisfy(
- cause ->
- assertThat(cause)
- .isInstanceOf(expectedRootCause.getClass())
- .hasMessage(expectedRootCause.getMessage()));
+ .eventuallyFailsWith(ExecutionException.class)
+ .havingCause()
+ .withCause(expectedRootCause);
client.delete().forPath(childNodePath);
counter.shutdown(JobStatus.FINISHED).join();
@@ -138,11 +130,11 @@ class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
/** Tests that counter node is NOT removed from ZooKeeper after suspend. */
@Test
- public void testSuspendKeepsState() throws Exception {
+ void testSuspendKeepsState() throws Exception {
ZooKeeperCheckpointIDCounter counter = createCheckpointIdCounter();
counter.start();
- CuratorFramework client = zookeeper.getClient();
+ CuratorFramework client = getZooKeeperClient();
assertThat(client.checkExists().forPath(counter.getPath())).isNotNull();
counter.shutdown(JobStatus.SUSPENDED).join();
@@ -152,7 +144,7 @@ class ZooKeeperCheckpointIDCounterITCase extends CheckpointIDCounterTestBase {
@Override
protected ZooKeeperCheckpointIDCounter createCheckpointIdCounter() throws Exception {
return new ZooKeeperCheckpointIDCounter(
- ZooKeeperUtils.useNamespaceAndEnsurePath(zookeeper.getClient(), "/"),
+ ZooKeeperUtils.useNamespaceAndEnsurePath(getZooKeeperClient(), "/"),
new DefaultLastStateConnectionStateListener());
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
index b9588d1c2c8..206023ca40e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ZooKeeperCompletedCheckpointStoreITCase.java
@@ -20,14 +20,15 @@ package org.apache.flink.runtime.checkpoint;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.runtime.jobgraph.RestoreMode;
-import org.apache.flink.runtime.state.RetrievableStateHandle;
import org.apache.flink.runtime.state.SharedStateRegistry;
import org.apache.flink.runtime.state.SharedStateRegistryImpl;
import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
-import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
import org.apache.flink.util.clock.ManualClock;
import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.util.concurrent.ManuallyTriggeredScheduledExecutor;
@@ -36,14 +37,11 @@ import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFram
import org.apache.flink.shaded.guava31.com.google.common.collect.Iterables;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.data.Stat;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
@@ -51,38 +49,37 @@ import java.util.concurrent.atomic.AtomicInteger;
import static java.util.Collections.emptyList;
import static org.apache.flink.runtime.checkpoint.CheckpointRequestDeciderTest.regularCheckpoint;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for basic {@link CompletedCheckpointStore} contract and ZooKeeper state handling. */
-public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpointStoreTest {
+class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpointStoreTest {
- private static final ZooKeeperTestEnvironment ZOOKEEPER = new ZooKeeperTestEnvironment(1);
+ private final ZooKeeperExtension zooKeeperExtension = new ZooKeeperExtension();
- private static final String CHECKPOINT_PATH = "/checkpoints";
+ @RegisterExtension
+ final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(zooKeeperExtension);
- private static final ZooKeeperCheckpointStoreUtil checkpointStoreUtil =
- ZooKeeperCheckpointStoreUtil.INSTANCE;
+ @RegisterExtension
+ final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
- @AfterClass
- public static void tearDown() throws Exception {
- ZOOKEEPER.shutdown();
+ private CuratorFramework getZooKeeperClient() {
+ return zooKeeperExtension.getZooKeeperClient(
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
}
- @Before
- public void cleanUp() throws Exception {
- ZOOKEEPER.deleteAll();
- }
+ private static final String CHECKPOINT_PATH = "/checkpoints";
+
+ private static final ZooKeeperCheckpointStoreUtil checkpointStoreUtil =
+ ZooKeeperCheckpointStoreUtil.INSTANCE;
@Override
protected CompletedCheckpointStore createRecoveredCompletedCheckpointStore(
int maxNumberOfCheckpointsToRetain, Executor executor) throws Exception {
final ZooKeeperStateHandleStore<CompletedCheckpoint> checkpointsInZooKeeper =
ZooKeeperUtils.createZooKeeperStateHandleStore(
- ZOOKEEPER.getClient(),
+ getZooKeeperClient(),
CHECKPOINT_PATH,
new TestingRetrievableStateStorageHelper<>());
return new DefaultCompletedCheckpointStore<>(
@@ -104,7 +101,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
* to retain will be removed.
*/
@Test
- public void testRecover() throws Exception {
+ void testRecover() throws Exception {
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpoints = createRecoveredCompletedCheckpointStore(3);
@@ -124,21 +121,21 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
checkpoints.addCheckpointAndSubsumeOldestOne(
expected[2], new CheckpointsCleaner(), () -> {});
- verifyCheckpointRegistered(expected[0].getOperatorStates().values(), sharedStateRegistry);
- verifyCheckpointRegistered(expected[1].getOperatorStates().values(), sharedStateRegistry);
- verifyCheckpointRegistered(expected[2].getOperatorStates().values(), sharedStateRegistry);
+ verifyCheckpointRegistered(expected[0].getOperatorStates().values());
+ verifyCheckpointRegistered(expected[1].getOperatorStates().values());
+ verifyCheckpointRegistered(expected[2].getOperatorStates().values());
// All three should be in ZK
- assertEquals(3, ZOOKEEPER.getClient().getChildren().forPath(CHECKPOINT_PATH).size());
- assertEquals(3, checkpoints.getNumberOfRetainedCheckpoints());
+ assertThat(getZooKeeperClient().getChildren().forPath(CHECKPOINT_PATH)).hasSize(3);
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isEqualTo(3);
// Recover
sharedStateRegistry.close();
sharedStateRegistry = new SharedStateRegistryImpl();
- assertEquals(3, ZOOKEEPER.getClient().getChildren().forPath(CHECKPOINT_PATH).size());
- assertEquals(3, checkpoints.getNumberOfRetainedCheckpoints());
- assertEquals(expected[2], checkpoints.getLatestCheckpoint());
+ assertThat(getZooKeeperClient().getChildren().forPath(CHECKPOINT_PATH)).hasSize(3);
+ assertThat(checkpoints.getNumberOfRetainedCheckpoints()).isEqualTo(3);
+ assertThat(checkpoints.getLatestCheckpoint()).isEqualTo(expected[2]);
List<CompletedCheckpoint> expectedCheckpoints = new ArrayList<>(3);
expectedCheckpoints.add(expected[1]);
@@ -150,45 +147,46 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
List<CompletedCheckpoint> actualCheckpoints = checkpoints.getAllCheckpoints();
- assertEquals(expectedCheckpoints, actualCheckpoints);
+ assertThat(actualCheckpoints).isEqualTo(expectedCheckpoints);
for (CompletedCheckpoint actualCheckpoint : actualCheckpoints) {
- verifyCheckpointRegistered(
- actualCheckpoint.getOperatorStates().values(), sharedStateRegistry);
+ verifyCheckpointRegistered(actualCheckpoint.getOperatorStates().values());
}
}
/** Tests that shutdown discards all checkpoints. */
@Test
- public void testShutdownDiscardsCheckpoints() throws Exception {
- CuratorFramework client = ZOOKEEPER.getClient();
+ void testShutdownDiscardsCheckpoints() throws Exception {
+ CuratorFramework client = getZooKeeperClient();
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore store = createRecoveredCompletedCheckpointStore(1);
TestCompletedCheckpoint checkpoint = createCheckpoint(0, sharedStateRegistry);
store.addCheckpointAndSubsumeOldestOne(checkpoint, new CheckpointsCleaner(), () -> {});
- assertEquals(1, store.getNumberOfRetainedCheckpoints());
- assertNotNull(
- client.checkExists()
- .forPath(
- CHECKPOINT_PATH
- + checkpointStoreUtil.checkpointIDToName(
- checkpoint.getCheckpointID())));
+ assertThat(store.getNumberOfRetainedCheckpoints()).isOne();
+ assertThat(
+ client.checkExists()
+ .forPath(
+ CHECKPOINT_PATH
+ + checkpointStoreUtil.checkpointIDToName(
+ checkpoint.getCheckpointID())))
+ .isNotNull();
store.shutdown(JobStatus.FINISHED, new CheckpointsCleaner());
- assertEquals(0, store.getNumberOfRetainedCheckpoints());
- assertNull(
- client.checkExists()
- .forPath(
- CHECKPOINT_PATH
- + checkpointStoreUtil.checkpointIDToName(
- checkpoint.getCheckpointID())));
+ assertThat(store.getNumberOfRetainedCheckpoints()).isZero();
+ assertThat(
+ client.checkExists()
+ .forPath(
+ CHECKPOINT_PATH
+ + checkpointStoreUtil.checkpointIDToName(
+ checkpoint.getCheckpointID())))
+ .isNull();
sharedStateRegistry.close();
- assertEquals(
- 0, createRecoveredCompletedCheckpointStore(1).getNumberOfRetainedCheckpoints());
+ assertThat(createRecoveredCompletedCheckpointStore(1).getNumberOfRetainedCheckpoints())
+ .isZero();
}
/**
@@ -196,51 +194,50 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
* ZooKeeper store). Furthermore, suspending a job should release all locks.
*/
@Test
- public void testSuspendKeepsCheckpoints() throws Exception {
- CuratorFramework client = ZOOKEEPER.getClient();
+ void testSuspendKeepsCheckpoints() throws Exception {
+ CuratorFramework client = getZooKeeperClient();
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore store = createRecoveredCompletedCheckpointStore(1);
TestCompletedCheckpoint checkpoint = createCheckpoint(0, sharedStateRegistry);
store.addCheckpointAndSubsumeOldestOne(checkpoint, new CheckpointsCleaner(), () -> {});
- assertEquals(1, store.getNumberOfRetainedCheckpoints());
- assertNotNull(
- client.checkExists()
- .forPath(
- CHECKPOINT_PATH
- + checkpointStoreUtil.checkpointIDToName(
- checkpoint.getCheckpointID())));
+ assertThat(store.getNumberOfRetainedCheckpoints()).isOne();
+ assertThat(
+ client.checkExists()
+ .forPath(
+ CHECKPOINT_PATH
+ + checkpointStoreUtil.checkpointIDToName(
+ checkpoint.getCheckpointID())))
+ .isNotNull();
store.shutdown(JobStatus.SUSPENDED, new CheckpointsCleaner());
- assertEquals(0, store.getNumberOfRetainedCheckpoints());
+ assertThat(store.getNumberOfRetainedCheckpoints()).isZero();
final String checkpointPath =
CHECKPOINT_PATH
+ checkpointStoreUtil.checkpointIDToName(checkpoint.getCheckpointID());
final List<String> checkpointPathChildren = client.getChildren().forPath(checkpointPath);
- assertEquals(
- "The checkpoint node should not be marked for deletion.",
- 1,
- checkpointPathChildren.size());
+ assertThat(checkpointPathChildren)
+ .as("The checkpoint node should not be marked for deletion.")
+ .hasSize(1);
final String locksNodeName = Iterables.getOnlyElement(checkpointPathChildren);
final String locksNodePath =
ZooKeeperUtils.generateZookeeperPath(checkpointPath, locksNodeName);
final Stat locksStat = client.checkExists().forPath(locksNodePath);
- assertEquals(
- "There shouldn't be any lock node available for the checkpoint",
- 0,
- locksStat.getNumChildren());
+ assertThat(locksStat.getNumChildren())
+ .as("There shouldn't be any lock node available for the checkpoint")
+ .isZero();
// Recover again
sharedStateRegistry.close();
store = createRecoveredCompletedCheckpointStore(1);
CompletedCheckpoint recovered = store.getLatestCheckpoint();
- assertEquals(checkpoint, recovered);
+ assertThat(recovered).isEqualTo(checkpoint);
}
/**
@@ -249,7 +246,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
* <p>Tests that the latest recovered checkpoint is the one with the highest checkpoint id
*/
@Test
- public void testLatestCheckpointRecovery() throws Exception {
+ void testLatestCheckpointRecovery() throws Exception {
final int numCheckpoints = 3;
SharedStateRegistry sharedStateRegistry = new SharedStateRegistryImpl();
CompletedCheckpointStore checkpointStore =
@@ -269,7 +266,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
final CompletedCheckpoint latestCheckpoint =
createRecoveredCompletedCheckpointStore(numCheckpoints).getLatestCheckpoint();
- assertEquals(checkpoints.get(checkpoints.size() - 1), latestCheckpoint);
+ assertThat(latestCheckpoint).isEqualTo(checkpoints.get(checkpoints.size() - 1));
}
/**
@@ -279,7 +276,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
* recovered by a different completed checkpoint store.
*/
@Test
- public void testConcurrentCheckpointOperations() throws Exception {
+ void testConcurrentCheckpointOperations() throws Exception {
final int numberOfCheckpoints = 1;
final long waitingTimeout = 50L;
@@ -301,12 +298,12 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
createRecoveredCompletedCheckpointStore(numberOfCheckpoints);
CompletedCheckpoint recoveredCheckpoint = zkCheckpointStore2.getLatestCheckpoint();
- assertTrue(recoveredCheckpoint instanceof TestCompletedCheckpoint);
+ assertThat(recoveredCheckpoint).isInstanceOf(TestCompletedCheckpoint.class);
TestCompletedCheckpoint recoveredTestCheckpoint =
(TestCompletedCheckpoint) recoveredCheckpoint;
// Check that the recovered checkpoint is not yet discarded
- assertFalse(recoveredTestCheckpoint.isDiscarded());
+ assertThat(recoveredTestCheckpoint.isDiscarded()).isFalse();
// complete another checkpoint --> this should remove the first checkpoint from the store
// because the number of retained checkpoints == 1
@@ -317,15 +314,15 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
List<CompletedCheckpoint> allCheckpoints = zkCheckpointStore1.getAllCheckpoints();
// check that we have removed the first checkpoint from zkCompletedStore1
- assertEquals(Collections.singletonList(completedCheckpoint2), allCheckpoints);
+ assertThat(allCheckpoints).isEqualTo(Collections.singletonList(completedCheckpoint2));
// lets wait a little bit to see that no discard operation will be executed
- assertFalse(
- "The checkpoint should not have been discarded.",
- recoveredTestCheckpoint.awaitDiscard(waitingTimeout));
+ assertThat(recoveredTestCheckpoint.awaitDiscard(waitingTimeout))
+ .as("The checkpoint should not have been discarded.")
+ .isFalse();
// check that we have not discarded the first completed checkpoint
- assertFalse(recoveredTestCheckpoint.isDiscarded());
+ assertThat(recoveredTestCheckpoint.isDiscarded()).isFalse();
TestCompletedCheckpoint completedCheckpoint3 = createCheckpoint(3, sharedStateRegistry);
@@ -343,7 +340,7 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
* threshold.
*/
@Test
- public void testChekpointingPausesAndResumeWhenTooManyCheckpoints() throws Exception {
+ void testChekpointingPausesAndResumeWhenTooManyCheckpoints() throws Exception {
ManualClock clock = new ManualClock();
clock.advanceTime(1, TimeUnit.DAYS);
int maxCleaningCheckpoints = 1;
@@ -384,15 +381,12 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
() ->
checkpointsCleaner.getNumberOfCheckpointsToClean()
== nbCheckpointsSubmittedForCleaning);
- assertEquals(
- nbCheckpointsSubmittedForCleaning,
- checkpointsCleaner.getNumberOfCheckpointsToClean());
+ assertThat(checkpointsCleaner.getNumberOfCheckpointsToClean())
+ .isEqualTo(nbCheckpointsSubmittedForCleaning);
// checkpointing is on hold because checkpointsCleaner.getNumberOfCheckpointsToClean() >
// maxCleaningCheckpoints
- assertFalse(
- checkpointRequestDecider
- .chooseRequestToExecute(regularCheckpoint(), false, 0)
- .isPresent());
+ assertThat(checkpointRequestDecider.chooseRequestToExecute(regularCheckpoint(), false, 0))
+ .isNotPresent();
// make the executor execute checkpoint requests.
executor.triggerAll();
@@ -402,48 +396,12 @@ public class ZooKeeperCompletedCheckpointStoreITCase extends CompletedCheckpoint
checkpointsCleaner.getNumberOfCheckpointsToClean()
< nbCheckpointsSubmittedForCleaning);
// some checkpoints were cleaned
- assertTrue(
- checkpointsCleaner.getNumberOfCheckpointsToClean()
- < nbCheckpointsSubmittedForCleaning);
+ assertThat(checkpointsCleaner.getNumberOfCheckpointsToClean())
+ .isLessThan(nbCheckpointsSubmittedForCleaning);
// checkpointing is resumed because checkpointsCleaner.getNumberOfCheckpointsToClean() <=
// maxCleaningCheckpoints
- assertTrue(
- checkpointRequestDecider
- .chooseRequestToExecute(regularCheckpoint(), false, 0)
- .isPresent());
+ assertThat(checkpointRequestDecider.chooseRequestToExecute(regularCheckpoint(), false, 0))
+ .isPresent();
checkpointStore.shutdown(JobStatus.FINISHED, checkpointsCleaner);
}
-
- static class HeapRetrievableStateHandle<T extends Serializable>
- implements RetrievableStateHandle<T> {
-
- private static final long serialVersionUID = -268548467968932L;
-
- private static AtomicInteger nextKey = new AtomicInteger(0);
-
- private static HashMap<Integer, Object> stateMap = new HashMap<>();
-
- private final int key;
-
- public HeapRetrievableStateHandle(T state) {
- key = nextKey.getAndIncrement();
- stateMap.put(key, state);
- }
-
- @SuppressWarnings("unchecked")
- @Override
- public T retrieveState() {
- return (T) stateMap.get(key);
- }
-
- @Override
- public void discardState() throws Exception {
- stateMap.remove(key);
- }
-
- @Override
- public long getStateSize() {
- return 0;
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/ZooKeeperJobGraphsStoreITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/ZooKeeperJobGraphsStoreITCase.java
index cb3e33f9aa0..4560d545fe0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/ZooKeeperJobGraphsStoreITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/ZooKeeperJobGraphsStoreITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.jobmanager;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.runtime.dispatcher.NoOpJobGraphListener;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphBuilder;
@@ -28,8 +29,9 @@ import org.apache.flink.runtime.jobmanager.JobGraphStore.JobGraphListener;
import org.apache.flink.runtime.persistence.RetrievableStateStorageHelper;
import org.apache.flink.runtime.state.RetrievableStreamStateHandle;
import org.apache.flink.runtime.state.memory.ByteStreamStateHandle;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
-import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.Executors;
@@ -37,9 +39,8 @@ import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.PathChildrenCache;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -50,13 +51,8 @@ import java.util.HashMap;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.atMost;
import static org.mockito.Mockito.doAnswer;
@@ -72,7 +68,15 @@ import static org.mockito.Mockito.verify;
*/
public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
- private static final ZooKeeperTestEnvironment ZooKeeper = new ZooKeeperTestEnvironment(1);
+ private final ZooKeeperExtension zooKeeperExtension = new ZooKeeperExtension();
+
+ @RegisterExtension
+ final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(zooKeeperExtension);
+
+ @RegisterExtension
+ final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
private static final RetrievableStateStorageHelper<JobGraph> localStateStorage =
jobGraph -> {
@@ -83,16 +87,6 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
return new RetrievableStreamStateHandle<>(byteStreamStateHandle);
};
- @AfterClass
- public static void tearDown() throws Exception {
- ZooKeeper.shutdown();
- }
-
- @Before
- public void cleanUp() throws Exception {
- ZooKeeper.deleteAll();
- }
-
@Test
public void testPutAndRemoveJobGraph() throws Exception {
JobGraphStore jobGraphs = createZooKeeperJobGraphStore("/testPutAndRemoveJobGraph");
@@ -105,14 +99,14 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
JobGraph jobGraph = createJobGraph(new JobID(), "JobName");
// Empty state
- assertEquals(0, jobGraphs.getJobIds().size());
+ assertThat(jobGraphs.getJobIds()).isEmpty();
// Add initial
jobGraphs.putJobGraph(jobGraph);
// Verify initial job graph
Collection<JobID> jobIds = jobGraphs.getJobIds();
- assertEquals(1, jobIds.size());
+ assertThat(jobIds).hasSize(1);
JobID jobId = jobIds.iterator().next();
@@ -124,7 +118,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
// Verify updated
jobIds = jobGraphs.getJobIds();
- assertEquals(1, jobIds.size());
+ assertThat(jobIds).hasSize(1);
jobId = jobIds.iterator().next();
@@ -134,7 +128,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
jobGraphs.globalCleanupAsync(jobGraph.getJobID(), Executors.directExecutor()).join();
// Empty state
- assertEquals(0, jobGraphs.getJobIds().size());
+ assertThat(jobGraphs.getJobIds()).isEmpty();
// Nothing should have been notified
verify(listener, atMost(1)).onAddedJobGraph(any(JobID.class));
@@ -149,7 +143,9 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
@Nonnull
private JobGraphStore createZooKeeperJobGraphStore(String fullPath) throws Exception {
- final CuratorFramework client = ZooKeeper.getClient();
+ final CuratorFramework client =
+ zooKeeperExtension.getZooKeeperClient(
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
// Ensure that the job graphs path exists
client.newNamespaceAwareEnsurePath(fullPath).ensure(client.getZookeeperClient());
@@ -186,11 +182,11 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
Collection<JobID> actual = jobGraphs.getJobIds();
- assertEquals(expected.size(), actual.size());
+ assertThat(actual).hasSameSizeAs(expected.entrySet());
for (JobID jobId : actual) {
JobGraph jobGraph = jobGraphs.recoverJobGraph(jobId);
- assertTrue(expected.containsKey(jobGraph.getJobID()));
+ assertThat(expected).containsKey(jobGraph.getJobID());
verifyJobGraphs(expected.get(jobGraph.getJobID()), jobGraph);
@@ -200,7 +196,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
}
// Empty state
- assertEquals(0, jobGraphs.getJobIds().size());
+ assertThat(jobGraphs.getJobIds()).isEmpty();
// Nothing should have been notified
verify(listener, atMost(expected.size())).onAddedJobGraph(any(JobID.class));
@@ -260,7 +256,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
verify(listener, times(1)).onAddedJobGraph(any(JobID.class));
verify(listener, never()).onRemovedJobGraph(any(JobID.class));
- assertEquals(otherJobGraph.getJobID(), actualOtherJobId[0]);
+ assertThat(actualOtherJobId[0]).isEqualTo(otherJobGraph.getJobID());
} finally {
if (jobGraphs != null) {
jobGraphs.stop();
@@ -272,7 +268,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
}
}
- @Test(expected = IllegalStateException.class)
+ @Test
public void testUpdateJobGraphYouDidNotGetOrAdd() throws Exception {
JobGraphStore jobGraphs =
createZooKeeperJobGraphStore("/testUpdateJobGraphYouDidNotGetOrAdd");
@@ -287,7 +283,8 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
jobGraphs.putJobGraph(jobGraph);
- otherJobGraphs.putJobGraph(jobGraph);
+ assertThatExceptionOfType(IllegalStateException.class)
+ .isThrownBy(() -> otherJobGraphs.putJobGraph(jobGraph));
}
/**
@@ -313,17 +310,18 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
final JobGraph recoveredJobGraph =
otherSubmittedJobGraphStore.recoverJobGraph(jobGraph.getJobID());
- assertThat(recoveredJobGraph, is(notNullValue()));
+ assertThat(recoveredJobGraph).isNotNull();
- try {
- otherSubmittedJobGraphStore
- .globalCleanupAsync(recoveredJobGraph.getJobID(), Executors.directExecutor())
- .join();
- fail(
- "It should not be possible to remove the JobGraph since the first store still has a lock on it.");
- } catch (Exception ignored) {
- // expected
- }
+ assertThatExceptionOfType(Exception.class)
+ .as(
+ "It should not be possible to remove the JobGraph since the first store still has a lock on it.")
+ .isThrownBy(
+ () ->
+ otherSubmittedJobGraphStore
+ .globalCleanupAsync(
+ recoveredJobGraph.getJobID(),
+ Executors.directExecutor())
+ .join());
submittedJobGraphStore.stop();
@@ -332,9 +330,8 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
.globalCleanupAsync(recoveredJobGraph.getJobID(), Executors.directExecutor())
.join();
- assertThat(
- otherSubmittedJobGraphStore.recoverJobGraph(recoveredJobGraph.getJobID()),
- is(nullValue()));
+ assertThat(otherSubmittedJobGraphStore.recoverJobGraph(recoveredJobGraph.getJobID()))
+ .isNull();
otherSubmittedJobGraphStore.stop();
}
@@ -357,7 +354,7 @@ public class ZooKeeperJobGraphsStoreITCase extends TestLogger {
}
private void verifyJobGraphs(JobGraph expected, JobGraph actual) {
- assertEquals(expected.getName(), actual.getName());
- assertEquals(expected.getJobID(), actual.getJobID());
+ assertThat(actual.getName()).isEqualTo(expected.getName());
+ assertThat(actual.getJobID()).isEqualTo(expected.getJobID());
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/zookeeper/ZooKeeperTestEnvironment.java b/flink-runtime/src/test/java/org/apache/flink/runtime/zookeeper/ZooKeeperTestEnvironment.java
deleted file mode 100644
index f89ce707acd..00000000000
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/zookeeper/ZooKeeperTestEnvironment.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.zookeeper;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.runtime.highavailability.zookeeper.CuratorFrameworkWithUnhandledErrorListener;
-import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
-import org.apache.flink.runtime.util.ExitJVMFatalErrorHandler;
-import org.apache.flink.runtime.util.ZooKeeperUtils;
-
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
-import org.apache.flink.shaded.curator5.org.apache.curator.utils.ZKPaths;
-import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
-
-import org.apache.curator.test.TestingCluster;
-import org.apache.curator.test.TestingServer;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
-
-/** Simple ZooKeeper and CuratorFramework setup for tests. */
-public class ZooKeeperTestEnvironment {
-
- private final TestingServer zooKeeperServer;
-
- private final TestingCluster zooKeeperCluster;
-
- private final CuratorFramework client;
-
- private final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
-
- /**
- * Starts a ZooKeeper cluster with the number of quorum peers and a client.
- *
- * @param numberOfZooKeeperQuorumPeers Starts a {@link TestingServer}, if <code>1</code>. Starts
- * a {@link TestingCluster}, if <code>=>1</code>.
- */
- public ZooKeeperTestEnvironment(int numberOfZooKeeperQuorumPeers) {
- if (numberOfZooKeeperQuorumPeers <= 0) {
- throw new IllegalArgumentException("Number of peers needs to be >= 1.");
- }
-
- final Configuration conf = new Configuration();
-
- try {
- if (numberOfZooKeeperQuorumPeers == 1) {
- zooKeeperServer = ZooKeeperTestUtils.createAndStartZookeeperTestingServer();
- zooKeeperCluster = null;
-
- conf.setString(
- HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
- zooKeeperServer.getConnectString());
- } else {
- zooKeeperServer = null;
- zooKeeperCluster = new TestingCluster(numberOfZooKeeperQuorumPeers);
-
- zooKeeperCluster.start();
-
- conf.setString(
- HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
- zooKeeperCluster.getConnectString());
- }
-
- curatorFrameworkWrapper =
- ZooKeeperUtils.startCuratorFramework(conf, ExitJVMFatalErrorHandler.INSTANCE);
-
- client = curatorFrameworkWrapper.asCuratorFramework();
-
- client.newNamespaceAwareEnsurePath("/").ensure(client.getZookeeperClient());
- } catch (Exception e) {
- throw new RuntimeException("Error setting up ZooKeeperTestEnvironment", e);
- }
- }
-
- /** Shutdown the client and ZooKeeper server/cluster. */
- public void shutdown() throws Exception {
- if (curatorFrameworkWrapper != null) {
- curatorFrameworkWrapper.close();
- }
-
- if (zooKeeperServer != null) {
- zooKeeperServer.close();
- }
-
- if (zooKeeperCluster != null) {
- zooKeeperCluster.close();
- }
- }
-
- public String getConnectString() {
- if (zooKeeperServer != null) {
- return zooKeeperServer.getConnectString();
- } else {
- return zooKeeperCluster.getConnectString();
- }
- }
-
- /** Returns a client for the started ZooKeeper server/cluster. */
- public CuratorFramework getClient() {
- return client;
- }
-
- public String getClientNamespace() {
- return client.getNamespace();
- }
-
- @Nullable
- public TestingCluster getZooKeeperCluster() {
- return zooKeeperCluster;
- }
-
- public List<String> getChildren(String path) throws Exception {
- return client.getChildren().forPath(path);
- }
-
- /**
- * Deletes all ZNodes under the root node.
- *
- * @throws Exception If the ZooKeeper operation fails
- */
- public void deleteAll() throws Exception {
- final String path = "/" + client.getNamespace();
-
- int maxAttempts = 10;
-
- for (int i = 0; i < maxAttempts; i++) {
- try {
- ZKPaths.deleteChildren(client.getZookeeperClient().getZooKeeper(), path, false);
- return;
- } catch (org.apache.zookeeper.KeeperException.NoNodeException e) {
- // that seems all right. if one of the children we want to delete is
- // actually already deleted, that's fine.
- return;
- } catch (KeeperException.ConnectionLossException e) {
- // Keep retrying
- Thread.sleep(100);
- }
- }
-
- throw new Exception(
- "Could not clear the ZNodes under "
- + path
- + ". ZooKeeper is not in "
- + "a clean state.");
- }
-}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
index d0b4cfc7862..6f6cce4cc08 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
@@ -35,6 +35,7 @@ import org.apache.flink.configuration.NettyShuffleEnvironmentOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.core.plugin.PluginUtils;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.dispatcher.DispatcherId;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
@@ -49,24 +50,22 @@ import org.apache.flink.runtime.taskexecutor.TaskExecutorResourceUtils;
import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
import org.apache.flink.runtime.testutils.DispatcherProcess;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
-import org.apache.flink.runtime.zookeeper.ZooKeeperTestEnvironment;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
+import org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
+import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
+import org.apache.flink.testutils.junit.utils.TempDirUtils;
import org.apache.flink.util.Collector;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.FutureUtils;
import org.apache.flink.util.concurrent.ScheduledExecutorServiceAdapter;
import org.apache.commons.io.FileUtils;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
import java.time.Duration;
@@ -78,9 +77,8 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.fail;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
/**
* Verify behaviour in case of JobManager process failure during job execution.
@@ -97,35 +95,22 @@ import static org.junit.Assert.fail;
* <p>This follows the same structure as {@link AbstractTaskManagerProcessFailureRecoveryTest}.
*/
@SuppressWarnings("serial")
-@RunWith(Parameterized.class)
-public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
+@ExtendWith(ParameterizedTestExtension.class)
+class JobManagerHAProcessFailureRecoveryITCase {
- private static ZooKeeperTestEnvironment zooKeeper;
+ private final ZooKeeperExtension zooKeeperExtension = new ZooKeeperExtension();
- private static final Duration TEST_TIMEOUT = Duration.ofMinutes(5);
-
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
-
- @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @RegisterExtension
+ final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(zooKeeperExtension);
- @BeforeClass
- public static void setup() {
- zooKeeper = new ZooKeeperTestEnvironment(1);
- }
+ private static final Duration TEST_TIMEOUT = Duration.ofMinutes(5);
- @Before
- public void cleanUp() throws Exception {
- zooKeeper.deleteAll();
- }
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
+ TestingUtils.defaultExecutorExtension();
- @AfterClass
- public static void tearDown() throws Exception {
- if (zooKeeper != null) {
- zooKeeper.shutdown();
- }
- }
+ @TempDir private java.nio.file.Path tempDir;
protected static final String READY_MARKER_FILE_PREFIX = "ready_";
protected static final String FINISH_MARKER_FILE_PREFIX = "finish_";
@@ -137,15 +122,11 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
// Parametrization (run pipelined and batch)
// --------------------------------------------------------------------------------------------
- private final ExecutionMode executionMode;
-
- public JobManagerHAProcessFailureRecoveryITCase(ExecutionMode executionMode) {
- this.executionMode = executionMode;
- }
+ @Parameter private ExecutionMode executionMode;
- @Parameterized.Parameters(name = "ExecutionMode {0}")
- public static Collection<Object[]> executionMode() {
- return Arrays.asList(new Object[][] {{ExecutionMode.PIPELINED}, {ExecutionMode.BATCH}});
+ @Parameters(name = "executionMode={0}")
+ private static Collection<ExecutionMode> executionMode() {
+ return Arrays.asList(ExecutionMode.PIPELINED, ExecutionMode.BATCH);
}
/**
@@ -225,9 +206,8 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
@Override
public void flatMap(Long value, Collector<Long> out)
throws Exception {
- assertEquals(
- numElements * (numElements + 1L) / 2L,
- (long) value);
+ assertThat((long) value)
+ .isEqualTo(numElements * (numElements + 1L) / 2L);
int taskIndex = getRuntimeContext().getIndexOfThisSubtask();
AbstractTaskManagerProcessFailureRecoveryTest.touchFile(
@@ -237,22 +217,21 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
}
});
- result.output(new DiscardingOutputFormat<Long>());
+ result.output(new DiscardingOutputFormat<>());
env.execute();
}
- @Test
- public void testDispatcherProcessFailure() throws Exception {
- final Time timeout = Time.seconds(30L);
- final File zookeeperStoragePath = temporaryFolder.newFolder();
+ @TestTemplate
+ void testDispatcherProcessFailure() throws Exception {
+ final File zookeeperStoragePath = TempDirUtils.newFolder(tempDir);
// Config
final int numberOfJobManagers = 2;
final int numberOfTaskManagers = 2;
final int numberOfSlotsPerTaskManager = 2;
- assertEquals(PARALLELISM, numberOfTaskManagers * numberOfSlotsPerTaskManager);
+ assertThat(numberOfTaskManagers * numberOfSlotsPerTaskManager).isEqualTo(PARALLELISM);
// Job managers
final DispatcherProcess[] dispatcherProcesses = new DispatcherProcess[numberOfJobManagers];
@@ -270,7 +249,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
// Cluster config
Configuration config =
ZooKeeperTestUtils.createZooKeeperHAConfig(
- zooKeeper.getConnectString(), zookeeperStoragePath.getPath());
+ zooKeeperExtension.getConnectString(), zookeeperStoragePath.getPath());
// Task manager configuration
config.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("4m"));
config.set(TaskManagerOptions.NETWORK_MEMORY_MIN, MemorySize.parse("3200k"));
@@ -288,7 +267,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
final Deadline deadline = Deadline.fromNow(TEST_TIMEOUT);
// Coordination directory
- coordinateTempDir = temporaryFolder.newFolder();
+ coordinateTempDir = TempDirUtils.newFolder(tempDir);
// Start first process
dispatcherProcesses[0] = new DispatcherProcess(0, config);
@@ -297,7 +276,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
highAvailabilityServices =
HighAvailabilityServicesUtils.createAvailableOrEmbeddedServices(
config,
- EXECUTOR_RESOURCE.getExecutor(),
+ EXECUTOR_EXTENSION.getExecutor(),
NoOpFatalErrorHandler.INSTANCE);
final PluginManager pluginManager =
@@ -343,7 +322,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
public void run() {
try {
testJobManagerFailure(
- zooKeeper.getConnectString(),
+ zooKeeperExtension.getConnectString(),
coordinateDirClosure,
zookeeperStoragePath);
} catch (Throwable t) {
@@ -385,7 +364,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
deadline.timeLeft().toMillis());
// check that the program really finished
- assertFalse("The program did not finish in time", programTrigger.isAlive());
+ assertThat(programTrigger.isAlive()).as("The program did not finish in time").isFalse();
// check whether the program encountered an error
if (errorRef[0] != null) {