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/02/07 14:07:28 UTC
[flink] 01/02: [FLINK-30801][runtime] Migrates LeaderElection- and LeaderRetrieval-related tests in flink-runtime 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
commit cba2e1e433e7fa0674834daed7823f6b4fa231b8
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Fri Jan 27 16:00:58 2023 +0100
[FLINK-30801][runtime] Migrates LeaderElection- and LeaderRetrieval-related tests in flink-runtime to JUnit5
Additionally, the deprecated curator API NodeCache was replaced by CuratorCache
---
.../DefaultLeaderElectionServiceTest.java | 121 ++++----
...MultipleComponentLeaderElectionServiceTest.java | 17 +-
.../LeaderChangeClusterComponentsTest.java | 62 ++--
.../runtime/leaderelection/LeaderElectionTest.java | 119 +++-----
.../StandaloneLeaderElectionTest.java | 27 +-
...KeeperLeaderElectionConnectionHandlingTest.java | 55 ++--
.../ZooKeeperLeaderElectionTest.java | 336 ++++++++++-----------
...rMultipleComponentLeaderElectionDriverTest.java | 19 +-
.../DefaultLeaderRetrievalServiceTest.java | 50 ++-
.../SettableLeaderRetrievalServiceTest.java | 30 +-
...eeperLeaderRetrievalConnectionHandlingTest.java | 15 +-
.../ZooKeeperLeaderRetrievalTest.java | 61 ++--
12 files changed, 419 insertions(+), 493 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java
index 17eb19f1b43..b506c54883d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultLeaderElectionServiceTest.java
@@ -19,27 +19,22 @@
package org.apache.flink.runtime.leaderelection;
import org.apache.flink.core.testutils.CheckedThread;
-import org.apache.flink.core.testutils.FlinkMatchers;
import org.apache.flink.util.Preconditions;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.RunnableWithException;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.UUID;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link DefaultLeaderElectionService}. */
-public class DefaultLeaderElectionServiceTest extends TestLogger {
+class DefaultLeaderElectionServiceTest {
private static final String TEST_URL = "akka//user/jobmanager";
@Test
- public void testOnGrantAndRevokeLeadership() throws Exception {
+ void testOnGrantAndRevokeLeadership() throws Exception {
new Context() {
{
runTest(
@@ -48,34 +43,31 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
testingLeaderElectionDriver.isLeader();
testingContender.waitForLeader();
- assertThat(testingContender.getDescription(), is(TEST_URL));
- assertThat(
- testingContender.getLeaderSessionID(),
- is(leaderElectionService.getLeaderSessionID()));
+ assertThat(testingContender.getDescription()).isEqualTo(TEST_URL);
+ assertThat(testingContender.getLeaderSessionID())
+ .isEqualTo(leaderElectionService.getLeaderSessionID());
// Check the external storage
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(
LeaderInformation.known(
leaderElectionService.getLeaderSessionID(),
- TEST_URL)));
+ TEST_URL));
// revoke leadership
testingLeaderElectionDriver.notLeader();
testingContender.waitForRevokeLeader();
- assertThat(testingContender.getLeaderSessionID(), is(nullValue()));
- assertThat(leaderElectionService.getLeaderSessionID(), is(nullValue()));
+ assertThat(testingContender.getLeaderSessionID()).isNull();
+ assertThat(leaderElectionService.getLeaderSessionID()).isNull();
// External storage should be cleared
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(LeaderInformation.empty()));
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(LeaderInformation.empty());
});
}
};
}
@Test
- public void testLeaderInformationChangedAndShouldBeCorrected() throws Exception {
+ void testLeaderInformationChangedAndShouldBeCorrected() throws Exception {
new Context() {
{
runTest(
@@ -90,22 +82,20 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
// corrected.
testingLeaderElectionDriver.leaderInformationChanged(
LeaderInformation.empty());
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(expectedLeader));
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(expectedLeader);
testingLeaderElectionDriver.leaderInformationChanged(
LeaderInformation.known(UUID.randomUUID(), "faulty-address"));
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(expectedLeader));
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(expectedLeader);
});
}
};
}
@Test
- public void testHasLeadership() throws Exception {
+ void testHasLeadership() throws Exception {
new Context() {
{
runTest(
@@ -113,25 +103,22 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
testingLeaderElectionDriver.isLeader();
final UUID currentLeaderSessionId =
leaderElectionService.getLeaderSessionID();
- assertThat(currentLeaderSessionId, is(notNullValue()));
- assertThat(
- leaderElectionService.hasLeadership(currentLeaderSessionId),
- is(true));
- assertThat(
- leaderElectionService.hasLeadership(UUID.randomUUID()),
- is(false));
+ assertThat(currentLeaderSessionId).isNotNull();
+ assertThat(leaderElectionService.hasLeadership(currentLeaderSessionId))
+ .isTrue();
+ assertThat(leaderElectionService.hasLeadership(UUID.randomUUID()))
+ .isFalse();
leaderElectionService.stop();
- assertThat(
- leaderElectionService.hasLeadership(currentLeaderSessionId),
- is(false));
+ assertThat(leaderElectionService.hasLeadership(currentLeaderSessionId))
+ .isFalse();
});
}
};
}
@Test
- public void testLeaderInformationChangedIfNotBeingLeader() throws Exception {
+ void testLeaderInformationChangedIfNotBeingLeader() throws Exception {
new Context() {
{
runTest(
@@ -140,16 +127,15 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
LeaderInformation.known(UUID.randomUUID(), "faulty-address");
testingLeaderElectionDriver.leaderInformationChanged(faultyLeader);
// External storage should keep the wrong value.
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(faultyLeader));
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(faultyLeader);
});
}
};
}
@Test
- public void testOnGrantLeadershipIsIgnoredAfterBeingStop() throws Exception {
+ void testOnGrantLeadershipIsIgnoredAfterBeingStop() throws Exception {
new Context() {
{
runTest(
@@ -157,14 +143,14 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
leaderElectionService.stop();
testingLeaderElectionDriver.isLeader();
// leader contender is not granted leadership
- assertThat(testingContender.getLeaderSessionID(), is(nullValue()));
+ assertThat(testingContender.getLeaderSessionID()).isNull();
});
}
};
}
@Test
- public void testOnLeaderInformationChangeIsIgnoredAfterBeingStop() throws Exception {
+ void testOnLeaderInformationChangeIsIgnoredAfterBeingStop() throws Exception {
new Context() {
{
runTest(
@@ -176,36 +162,37 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
LeaderInformation.empty());
// External storage should not be corrected
- assertThat(
- testingLeaderElectionDriver.getLeaderInformation(),
- is(LeaderInformation.empty()));
+ assertThat(testingLeaderElectionDriver.getLeaderInformation())
+ .isEqualTo(LeaderInformation.empty());
});
}
};
}
@Test
- public void testOnRevokeLeadershipIsIgnoredAfterBeingStop() throws Exception {
+ void testOnRevokeLeadershipIsIgnoredAfterBeingStop() throws Exception {
new Context() {
{
runTest(
() -> {
testingLeaderElectionDriver.isLeader();
final UUID oldSessionId = leaderElectionService.getLeaderSessionID();
- assertThat(testingContender.getLeaderSessionID(), is(oldSessionId));
+ assertThat(testingContender.getLeaderSessionID())
+ .isEqualTo(oldSessionId);
leaderElectionService.stop();
testingLeaderElectionDriver.notLeader();
// leader contender is not revoked leadership
- assertThat(testingContender.getLeaderSessionID(), is(oldSessionId));
+ assertThat(testingContender.getLeaderSessionID())
+ .isEqualTo(oldSessionId);
});
}
};
}
@Test
- public void testOldConfirmLeaderInformation() throws Exception {
+ void testOldConfirmLeaderInformation() throws Exception {
new Context() {
{
runTest(
@@ -213,20 +200,19 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
testingLeaderElectionDriver.isLeader();
final UUID currentLeaderSessionId =
leaderElectionService.getLeaderSessionID();
- assertThat(currentLeaderSessionId, is(notNullValue()));
+ assertThat(currentLeaderSessionId).isNotNull();
// Old confirm call should be ignored.
leaderElectionService.confirmLeadership(UUID.randomUUID(), TEST_URL);
- assertThat(
- leaderElectionService.getLeaderSessionID(),
- is(currentLeaderSessionId));
+ assertThat(leaderElectionService.getLeaderSessionID())
+ .isEqualTo(currentLeaderSessionId);
});
}
};
}
@Test
- public void testErrorForwarding() throws Exception {
+ void testErrorForwarding() throws Exception {
new Context() {
{
runTest(
@@ -236,17 +222,16 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
testingLeaderElectionDriver.onFatalError(testException);
testingContender.waitForError();
- assertThat(testingContender.getError(), is(notNullValue()));
- assertThat(
- testingContender.getError(),
- FlinkMatchers.containsCause(testException));
+ assertThat(testingContender.getError())
+ .isNotNull()
+ .hasCause(testException);
});
}
};
}
@Test
- public void testErrorIsIgnoredAfterBeingStop() throws Exception {
+ void testErrorIsIgnoredAfterBeingStop() throws Exception {
new Context() {
{
runTest(
@@ -255,7 +240,7 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
leaderElectionService.stop();
testingLeaderElectionDriver.onFatalError(testException);
- assertThat(testingContender.getError(), is(nullValue()));
+ assertThat(testingContender.getError()).isNull();
});
}
};
@@ -266,7 +251,7 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
* holds an internal lock. See FLINK-20008 for more details.
*/
@Test
- public void testServiceShutDownWithSynchronizedDriver() throws Exception {
+ void testServiceShutDownWithSynchronizedDriver() throws Exception {
final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
testingLeaderElectionDriverFactory =
new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
@@ -294,7 +279,7 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
isLeaderThread.sync();
}
- private class Context {
+ private static class Context {
final TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory
testingLeaderElectionDriverFactory =
new TestingLeaderElectionDriver.TestingLeaderElectionDriverFactory();
@@ -310,7 +295,7 @@ public class DefaultLeaderElectionServiceTest extends TestLogger {
testingLeaderElectionDriver =
testingLeaderElectionDriverFactory.getCurrentLeaderDriver();
- assertThat(testingLeaderElectionDriver, is(notNullValue()));
+ assertThat(testingLeaderElectionDriver).isNotNull();
testMethod.run();
leaderElectionService.stop();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java
index cb2da45cf6c..f30eb54b689 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/DefaultMultipleComponentLeaderElectionServiceTest.java
@@ -21,11 +21,9 @@ package org.apache.flink.runtime.leaderelection;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.concurrent.Executors;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import javax.annotation.Nullable;
@@ -41,15 +39,14 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link DefaultMultipleComponentLeaderElectionService}. */
-@ExtendWith(TestLoggerExtension.class)
class DefaultMultipleComponentLeaderElectionServiceTest {
@RegisterExtension
- public final TestingFatalErrorHandlerExtension fatalErrorHandlerExtension =
+ final TestingFatalErrorHandlerExtension fatalErrorHandlerExtension =
new TestingFatalErrorHandlerExtension();
@Test
- public void isLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
+ void isLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
@@ -90,7 +87,7 @@ class DefaultMultipleComponentLeaderElectionServiceTest {
}
@Test
- public void notLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
+ void notLeaderInformsAllRegisteredLeaderElectionEventHandlers() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
@@ -122,7 +119,7 @@ class DefaultMultipleComponentLeaderElectionServiceTest {
}
@Test
- public void unregisteredEventHandlersAreNotNotified() throws Exception {
+ void unregisteredEventHandlersAreNotNotified() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
@@ -146,7 +143,7 @@ class DefaultMultipleComponentLeaderElectionServiceTest {
}
@Test
- public void newlyRegisteredEventHandlersAreInformedAboutLeadership() throws Exception {
+ void newlyRegisteredEventHandlersAreInformedAboutLeadership() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
final DefaultMultipleComponentLeaderElectionService leaderElectionService =
@@ -212,7 +209,7 @@ class DefaultMultipleComponentLeaderElectionServiceTest {
}
@Test
- public void allKnownLeaderInformationCallsEventHandlers() throws Exception {
+ void allKnownLeaderInformationCallsEventHandlers() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
final DefaultMultipleComponentLeaderElectionService leaderElectionService =
@@ -252,7 +249,7 @@ class DefaultMultipleComponentLeaderElectionServiceTest {
}
@Test
- public void allKnownLeaderInformationDoesNotBlock() throws Exception {
+ void allKnownLeaderInformationDoesNotBlock() throws Exception {
final TestingMultipleComponentLeaderElectionDriver leaderElectionDriver =
TestingMultipleComponentLeaderElectionDriver.newBuilder().build();
final DefaultMultipleComponentLeaderElectionService leaderElectionService =
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderChangeClusterComponentsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderChangeClusterComponentsTest.java
index 90c1c1ad280..0285222c388 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderChangeClusterComponentsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderChangeClusterComponentsTest.java
@@ -34,26 +34,22 @@ import org.apache.flink.runtime.minicluster.TestingMiniClusterConfiguration;
import org.apache.flink.runtime.testutils.CommonTestUtils;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.ClassRule;
-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 org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests which verify the cluster behaviour in case of leader changes. */
-public class LeaderChangeClusterComponentsTest extends TestLogger {
+class LeaderChangeClusterComponentsTest {
private static final Duration TESTING_TIMEOUT = Duration.ofMinutes(2L);
@@ -61,9 +57,9 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
private static final int NUM_TMS = 2;
public static final int PARALLELISM = SLOTS_PER_TM * NUM_TMS;
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ private static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
private static TestingMiniCluster miniCluster;
@@ -73,8 +69,8 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
private JobID jobId;
- @BeforeClass
- public static void setupClass() throws Exception {
+ @BeforeAll
+ static void setupClass() throws Exception {
highAvailabilityServices =
new EmbeddedHaServicesWithLeadershipControl(EXECUTOR_RESOURCE.getExecutor());
@@ -91,21 +87,21 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
miniCluster.start();
}
- @Before
- public void setup() throws Exception {
+ @BeforeEach
+ void setup() {
jobGraph = createJobGraph(PARALLELISM);
jobId = jobGraph.getJobID();
}
- @AfterClass
- public static void teardownClass() throws Exception {
+ @AfterAll
+ static void teardownClass() throws Exception {
if (miniCluster != null) {
miniCluster.close();
}
}
@Test
- public void testReelectionOfDispatcher() throws Exception {
+ void testReelectionOfDispatcher() throws Exception {
final CompletableFuture<JobSubmissionResult> submissionFuture =
miniCluster.submitJob(jobGraph);
@@ -116,7 +112,7 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
highAvailabilityServices.revokeDispatcherLeadership().get();
JobResult jobResult = jobResultFuture.get();
- assertEquals(jobResult.getApplicationStatus(), ApplicationStatus.UNKNOWN);
+ assertThat(jobResult.getApplicationStatus()).isEqualTo(ApplicationStatus.UNKNOWN);
highAvailabilityServices.grantDispatcherLeadership();
@@ -135,7 +131,7 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
}
@Test
- public void testReelectionOfJobMaster() throws Exception {
+ void testReelectionOfJobMaster() throws Exception {
final CompletableFuture<JobSubmissionResult> submissionFuture =
miniCluster.submitJob(jobGraph);
@@ -160,18 +156,19 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
}
@Test
- public void testTaskExecutorsReconnectToClusterWithLeadershipChange() throws Exception {
+ void testTaskExecutorsReconnectToClusterWithLeadershipChange() throws Exception {
waitUntilTaskExecutorsHaveConnected(NUM_TMS);
highAvailabilityServices.revokeResourceManagerLeadership().get();
highAvailabilityServices.grantResourceManagerLeadership();
// wait for the ResourceManager to confirm the leadership
assertThat(
- LeaderRetrievalUtils.retrieveLeaderConnectionInfo(
- highAvailabilityServices.getResourceManagerLeaderRetriever(),
- TESTING_TIMEOUT)
- .getLeaderSessionId(),
- is(notNullValue()));
+ LeaderRetrievalUtils.retrieveLeaderConnectionInfo(
+ highAvailabilityServices
+ .getResourceManagerLeaderRetriever(),
+ TESTING_TIMEOUT)
+ .getLeaderSessionId())
+ .isNotNull();
waitUntilTaskExecutorsHaveConnected(NUM_TMS);
}
@@ -193,7 +190,10 @@ public class LeaderChangeClusterComponentsTest extends TestLogger {
return JobGraphTestUtils.streamingJobGraph(vertex);
}
- /** Blocking invokable which is controlled by a static field. */
+ /**
+ * Blocking invokable which is controlled by a static field. This class needs to be {@code
+ * public} because it is going to be instantiated from outside this testing class.
+ */
public static class BlockingOperator extends AbstractInvokable {
static boolean isBlocking = true;
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
index 9f9792c9f8c..c5a09be8bc1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/LeaderElectionTest.java
@@ -24,20 +24,20 @@ import org.apache.flink.runtime.highavailability.nonha.embedded.EmbeddedLeaderSe
import org.apache.flink.runtime.highavailability.zookeeper.CuratorFrameworkWithUnhandledErrorListener;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+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.curator.test.TestingServer;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Arrays;
import java.util.Collection;
@@ -45,84 +45,63 @@ import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for leader election. */
-@RunWith(Parameterized.class)
-public class LeaderElectionTest extends TestLogger {
-
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
-
- @Rule
- public final TestingFatalErrorHandlerResource testingFatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
-
- enum LeaderElectionType {
- ZooKeeper,
- Embedded,
- Standalone
+@ExtendWith(ParameterizedTestExtension.class)
+public class LeaderElectionTest {
+
+ @RegisterExtension
+ private static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
+
+ @RegisterExtension
+ private final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
+
+ @Parameters(name = "Leader election: {0}")
+ public static Collection<ServiceClass> parameters() {
+ return Arrays.asList(
+ new ZooKeeperServiceClass(),
+ new EmbeddedServiceClass(),
+ new StandaloneServiceClass());
}
- @Parameterized.Parameters(name = "Leader election: {0}")
- public static Collection<LeaderElectionType> parameters() {
- return Arrays.asList(LeaderElectionType.values());
- }
-
- private final ServiceClass serviceClass;
-
- public LeaderElectionTest(LeaderElectionType leaderElectionType) {
- switch (leaderElectionType) {
- case ZooKeeper:
- serviceClass = new ZooKeeperServiceClass();
- break;
- case Embedded:
- serviceClass = new EmbeddedServiceClass();
- break;
- case Standalone:
- serviceClass = new StandaloneServiceClass();
- break;
- default:
- throw new IllegalArgumentException(
- String.format("Unknown leader election type: %s.", leaderElectionType));
- }
- }
+ @Parameter public ServiceClass serviceClass;
- @Before
- public void setup() throws Exception {
- serviceClass.setup(testingFatalErrorHandlerResource.getFatalErrorHandler());
+ @BeforeEach
+ void setup() throws Exception {
+ serviceClass.setup(testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
}
- @After
- public void teardown() throws Exception {
+ @AfterEach
+ void teardown() throws Exception {
serviceClass.teardown();
}
- @Test
- public void testHasLeadership() throws Exception {
+ @TestTemplate
+ void testHasLeadership() throws Exception {
final LeaderElectionService leaderElectionService =
serviceClass.createLeaderElectionService();
final ManualLeaderContender manualLeaderContender = new ManualLeaderContender();
try {
- assertThat(leaderElectionService.hasLeadership(UUID.randomUUID()), is(false));
+ assertThat(leaderElectionService.hasLeadership(UUID.randomUUID())).isFalse();
leaderElectionService.start(manualLeaderContender);
final UUID leaderSessionId = manualLeaderContender.waitForLeaderSessionId();
- assertThat(leaderElectionService.hasLeadership(leaderSessionId), is(true));
- assertThat(leaderElectionService.hasLeadership(UUID.randomUUID()), is(false));
+ assertThat(leaderElectionService.hasLeadership(leaderSessionId)).isTrue();
+ assertThat(leaderElectionService.hasLeadership(UUID.randomUUID())).isFalse();
leaderElectionService.confirmLeadership(leaderSessionId, "foobar");
- assertThat(leaderElectionService.hasLeadership(leaderSessionId), is(true));
+ assertThat(leaderElectionService.hasLeadership(leaderSessionId)).isTrue();
leaderElectionService.stop();
- assertThat(leaderElectionService.hasLeadership(leaderSessionId), is(false));
+ assertThat(leaderElectionService.hasLeadership(leaderSessionId)).isFalse();
} finally {
manualLeaderContender.rethrowError();
}
@@ -181,17 +160,15 @@ public class LeaderElectionTest extends TestLogger {
private CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
- private Configuration configuration;
-
@Override
- public void setup(FatalErrorHandler fatalErrorHandler) throws Exception {
+ public void setup(FatalErrorHandler fatalErrorHandler) {
try {
testingServer = ZooKeeperTestUtils.createAndStartZookeeperTestingServer();
} catch (Exception e) {
throw new RuntimeException("Could not start ZooKeeper testing cluster.", e);
}
- configuration = new Configuration();
+ final Configuration configuration = new Configuration();
configuration.setString(
HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, testingServer.getConnectString());
@@ -215,7 +192,7 @@ public class LeaderElectionTest extends TestLogger {
}
@Override
- public LeaderElectionService createLeaderElectionService() throws Exception {
+ public LeaderElectionService createLeaderElectionService() {
return ZooKeeperUtils.createLeaderElectionService(
curatorFrameworkWrapper.asCuratorFramework());
}
@@ -238,7 +215,7 @@ public class LeaderElectionTest extends TestLogger {
}
@Override
- public LeaderElectionService createLeaderElectionService() throws Exception {
+ public LeaderElectionService createLeaderElectionService() {
return embeddedLeaderService.createLeaderElectionService();
}
}
@@ -246,17 +223,17 @@ public class LeaderElectionTest extends TestLogger {
private static final class StandaloneServiceClass implements ServiceClass {
@Override
- public void setup(FatalErrorHandler fatalErrorHandler) throws Exception {
+ public void setup(FatalErrorHandler fatalErrorHandler) {
// noop
}
@Override
- public void teardown() throws Exception {
+ public void teardown() {
// noop
}
@Override
- public LeaderElectionService createLeaderElectionService() throws Exception {
+ public LeaderElectionService createLeaderElectionService() {
return new StandaloneLeaderElectionService();
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/StandaloneLeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/StandaloneLeaderElectionTest.java
index ae715a3ee83..9dd42033038 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/StandaloneLeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/StandaloneLeaderElectionTest.java
@@ -20,25 +20,25 @@ package org.apache.flink.runtime.leaderelection;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderretrieval.StandaloneLeaderRetrievalService;
-import org.apache.flink.util.TestLogger;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class StandaloneLeaderElectionTest {
-public class StandaloneLeaderElectionTest extends TestLogger {
private static final String TEST_URL = "akka://users/jobmanager";
/**
* Tests that the standalone leader election and retrieval service return the same leader URL.
*/
@Test
- public void testStandaloneLeaderElectionRetrieval() throws Exception {
+ void testStandaloneLeaderElectionRetrieval() throws Exception {
StandaloneLeaderElectionService leaderElectionService =
new StandaloneLeaderElectionService();
StandaloneLeaderRetrievalService leaderRetrievalService =
- new StandaloneLeaderRetrievalService(TEST_URL);
+ new StandaloneLeaderRetrievalService(
+ TEST_URL, HighAvailabilityServices.DEFAULT_LEADER_ID);
TestingContender contender = new TestingContender(TEST_URL, leaderElectionService);
TestingListener testingListener = new TestingListener();
@@ -48,16 +48,15 @@ public class StandaloneLeaderElectionTest extends TestLogger {
contender.waitForLeader();
- assertTrue(contender.isLeader());
- assertEquals(
- HighAvailabilityServices.DEFAULT_LEADER_ID, contender.getLeaderSessionID());
+ assertThat(contender.isLeader()).isTrue();
+ assertThat(contender.getLeaderSessionID())
+ .isEqualTo(HighAvailabilityServices.DEFAULT_LEADER_ID);
testingListener.waitForNewLeader();
- assertEquals(TEST_URL, testingListener.getAddress());
- assertEquals(
- HighAvailabilityServices.DEFAULT_LEADER_ID,
- testingListener.getLeaderSessionID());
+ assertThat(testingListener.getAddress()).isEqualTo(TEST_URL);
+ assertThat(testingListener.getLeaderSessionID())
+ .isEqualTo(HighAvailabilityServices.DEFAULT_LEADER_ID);
} finally {
leaderElectionService.stop();
leaderRetrievalService.stop();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
index 25160c7a275..67a6387c25b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionConnectionHandlingTest.java
@@ -20,49 +20,52 @@ package org.apache.flink.runtime.leaderelection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.highavailability.zookeeper.CuratorFrameworkWithUnhandledErrorListener;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.runtime.zookeeper.ZooKeeperResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.util.function.BiConsumerWithException;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.state.ConnectionState;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.state.ConnectionStateListener;
-import org.junit.Before;
-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.UUID;
-import static org.junit.Assert.assertFalse;
+import static org.assertj.core.api.Assertions.assertThat;
/**
* Test behaviors of {@link ZooKeeperLeaderElectionDriver} when losing the connection to ZooKeeper.
*/
-public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
+class ZooKeeperLeaderElectionConnectionHandlingTest {
private static final String PATH = "/path";
- @Rule public final ZooKeeperResource zooKeeperResource = new ZooKeeperResource();
+ @RegisterExtension
+ private static final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(new ZooKeeperExtension());
- @Rule
- public final TestingFatalErrorHandlerResource fatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
+ @RegisterExtension
+ private final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
private final Configuration configuration = new Configuration();
- @Before
- public void setup() {
+ @BeforeEach
+ void setup() {
configuration.set(
- HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, zooKeeperResource.getConnectString());
+ HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
+ zooKeeperResource.getCustomExtension().getConnectString());
}
@Test
- public void testLoseLeadershipOnConnectionSuspended() throws Exception {
+ void testLoseLeadershipOnConnectionSuspended() throws Exception {
runTestWithBrieflySuspendedZooKeeperConnection(
configuration,
(connectionStateListener, contender) -> {
@@ -72,7 +75,7 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
}
@Test
- public void testKeepLeadershipOnSuspendedConnectionIfTolerateSuspendedConnectionsIsEnabled()
+ void testKeepLeadershipOnSuspendedConnectionIfTolerateSuspendedConnectionsIsEnabled()
throws Exception {
configuration.set(HighAvailabilityOptions.ZOOKEEPER_TOLERATE_SUSPENDED_CONNECTIONS, true);
runTestWithBrieflySuspendedZooKeeperConnection(
@@ -80,12 +83,12 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
(connectionStateListener, contender) -> {
connectionStateListener.awaitSuspendedConnection();
connectionStateListener.awaitReconnectedConnection();
- assertFalse(contender.hasRevokeLeadershipBeenTriggered());
+ assertThat(contender.hasRevokeLeadershipBeenTriggered()).isFalse();
});
}
@Test
- public void testLoseLeadershipOnLostConnectionIfTolerateSuspendedConnectionsIsEnabled()
+ void testLoseLeadershipOnLostConnectionIfTolerateSuspendedConnectionsIsEnabled()
throws Exception {
configuration.set(HighAvailabilityOptions.ZOOKEEPER_SESSION_TIMEOUT, 1000);
configuration.set(HighAvailabilityOptions.ZOOKEEPER_CONNECTION_TIMEOUT, 1000);
@@ -129,7 +132,8 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
throws Exception {
CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper =
ZooKeeperUtils.startCuratorFramework(
- configuration, fatalErrorHandlerResource.getFatalErrorHandler());
+ configuration,
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
CuratorFramework client = curatorFrameworkWrapper.asCuratorFramework();
LeaderElectionDriverFactory leaderElectionDriverFactory =
new ZooKeeperLeaderElectionDriverFactory(client, PATH);
@@ -148,10 +152,10 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
switch (problem) {
case SUSPENDED_CONNECTION:
- zooKeeperResource.restart();
+ zooKeeperResource.getCustomExtension().restart();
break;
case LOST_CONNECTION:
- zooKeeperResource.stop();
+ zooKeeperResource.getCustomExtension().stop();
break;
default:
throw new IllegalArgumentException(
@@ -165,7 +169,7 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
if (problem == Problem.LOST_CONNECTION) {
// in case of lost connections we accept that some unhandled error can occur
- fatalErrorHandlerResource.getFatalErrorHandler().clearError();
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler().clearError();
}
}
}
@@ -192,7 +196,10 @@ public class ZooKeeperLeaderElectionConnectionHandlingTest extends TestLogger {
@Override
public void handleError(Exception exception) {
- fatalErrorHandlerResource.getFatalErrorHandler().onFatalError(exception);
+ ZooKeeperLeaderElectionConnectionHandlingTest.this
+ .testingFatalErrorHandlerResource
+ .getTestingFatalErrorHandler()
+ .onFatalError(exception);
}
public void awaitGrantLeadership() throws InterruptedException {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
index 352bad03bee..a78585c8d08 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperLeaderElectionTest.java
@@ -21,7 +21,8 @@ package org.apache.flink.runtime.leaderelection;
import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
-import org.apache.flink.core.testutils.FlinkMatchers;
+import org.apache.flink.core.testutils.EachCallbackWrapper;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.runtime.highavailability.zookeeper.CuratorFrameworkWithUnhandledErrorListener;
import org.apache.flink.runtime.leaderretrieval.DefaultLeaderRetrievalService;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalDriver;
@@ -29,41 +30,34 @@ import org.apache.flink.runtime.leaderretrieval.TestingLeaderRetrievalEventHandl
import org.apache.flink.runtime.leaderretrieval.ZooKeeperLeaderRetrievalDriver;
import org.apache.flink.runtime.rest.util.NoOpFatalErrorHandler;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
-import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
-import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.util.FlinkRuntimeException;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.api.ACLProvider;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.api.CreateBuilder;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.ChildData;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.NodeCache;
-import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.NodeCacheListener;
+import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.CuratorCache;
+import org.apache.flink.shaded.curator5.org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.flink.shaded.curator5.org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.CreateMode;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.KeeperException;
import org.apache.flink.shaded.zookeeper3.org.apache.zookeeper.data.ACL;
-import org.apache.curator.test.TestingServer;
-import org.assertj.core.api.Assertions;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.mockito.Matchers;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.ArgumentMatchers;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nonnull;
-
import java.io.ByteArrayOutputStream;
-import java.io.IOException;
import java.io.ObjectOutputStream;
import java.time.Duration;
import java.util.List;
@@ -76,12 +70,8 @@ import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import static org.hamcrest.Matchers.is;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-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.fail;
import static org.mockito.Mockito.any;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.doAnswer;
@@ -98,57 +88,36 @@ import static org.mockito.Mockito.when;
* ZooKeeper. For the complicated tests(e.g. multiple leaders), we will use {@link
* DefaultLeaderElectionService} with {@link TestingContender}.
*/
-public class ZooKeeperLeaderElectionTest extends TestLogger {
- private TestingServer testingServer;
+class ZooKeeperLeaderElectionTest {
- private Configuration configuration;
+ @RegisterExtension
+ private static final EachCallbackWrapper<ZooKeeperExtension> zooKeeperResource =
+ new EachCallbackWrapper<>(new ZooKeeperExtension());
- private CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
+ @RegisterExtension
+ private final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
+
+ private Configuration configuration;
private static final String LEADER_ADDRESS = "akka//user/jobmanager";
private static final long timeout = 200L * 1000L;
private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperLeaderElectionTest.class);
- @Rule
- public final TestingFatalErrorHandlerResource testingFatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
-
- @Before
- public void before() {
- try {
- testingServer = ZooKeeperTestUtils.createAndStartZookeeperTestingServer();
- } catch (Exception e) {
- throw new RuntimeException("Could not start ZooKeeper testing cluster.", e);
- }
-
+ @BeforeEach
+ void before() {
configuration = new Configuration();
configuration.setString(
- HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM, testingServer.getConnectString());
+ HighAvailabilityOptions.HA_ZOOKEEPER_QUORUM,
+ zooKeeperResource.getCustomExtension().getConnectString());
configuration.setString(HighAvailabilityOptions.HA_MODE, "zookeeper");
-
- curatorFrameworkWrapper =
- ZooKeeperUtils.startCuratorFramework(
- configuration, testingFatalErrorHandlerResource.getFatalErrorHandler());
- }
-
- @After
- public void after() throws IOException {
- if (curatorFrameworkWrapper != null) {
- curatorFrameworkWrapper.close();
- curatorFrameworkWrapper = null;
- }
-
- if (testingServer != null) {
- testingServer.close();
- testingServer = null;
- }
}
/** Tests that the ZooKeeperLeaderElection/RetrievalService return both the correct URL. */
@Test
- public void testZooKeeperLeaderElectionRetrieval() throws Exception {
+ void testZooKeeperLeaderElectionRetrieval() throws Exception {
final TestingLeaderElectionEventHandler electionEventHandler =
new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
@@ -160,26 +129,23 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
leaderElectionDriver =
createAndInitLeaderElectionDriver(
- curatorFrameworkWrapper.asCuratorFramework(), electionEventHandler);
+ createZooKeeperClient(), electionEventHandler);
leaderRetrievalDriver =
- ZooKeeperUtils.createLeaderRetrievalDriverFactory(
- curatorFrameworkWrapper.asCuratorFramework())
+ ZooKeeperUtils.createLeaderRetrievalDriverFactory(createZooKeeperClient())
.createLeaderRetrievalDriver(
retrievalEventHandler, retrievalEventHandler::handleError);
electionEventHandler.waitForLeader();
final LeaderInformation confirmedLeaderInformation =
electionEventHandler.getConfirmedLeaderInformation();
- assertThat(confirmedLeaderInformation.getLeaderAddress(), is(LEADER_ADDRESS));
+ assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
retrievalEventHandler.waitForNewLeader();
- assertThat(
- retrievalEventHandler.getLeaderSessionID(),
- is(confirmedLeaderInformation.getLeaderSessionID()));
- assertThat(
- retrievalEventHandler.getAddress(),
- is(confirmedLeaderInformation.getLeaderAddress()));
+ assertThat(retrievalEventHandler.getLeaderSessionID())
+ .isEqualTo(confirmedLeaderInformation.getLeaderSessionID());
+ assertThat(retrievalEventHandler.getAddress())
+ .isEqualTo(confirmedLeaderInformation.getLeaderAddress());
} finally {
electionEventHandler.close();
if (leaderElectionDriver != null) {
@@ -197,7 +163,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
* elect a new leader.
*/
@Test
- public void testZooKeeperReelection() throws Exception {
+ void testZooKeeperReelection() throws Exception {
Deadline deadline = Deadline.fromNow(Duration.ofMinutes(5L));
int num = 10;
@@ -211,8 +177,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
try {
leaderRetrievalService =
- ZooKeeperUtils.createLeaderRetrievalService(
- curatorFrameworkWrapper.asCuratorFramework());
+ ZooKeeperUtils.createLeaderRetrievalService(createZooKeeperClient());
LOG.debug("Start leader retrieval service for the TestingListener.");
@@ -220,8 +185,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
for (int i = 0; i < num; i++) {
leaderElectionService[i] =
- ZooKeeperUtils.createLeaderElectionService(
- curatorFrameworkWrapper.asCuratorFramework());
+ ZooKeeperUtils.createLeaderElectionService(createZooKeeperClient());
contenders[i] = new TestingContender(createAddress(i), leaderElectionService[i]);
LOG.debug("Start leader election service for contender #{}.", i);
@@ -263,9 +227,10 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
}
- assertFalse("Did not complete the leader reelection in time.", deadline.isOverdue());
- assertEquals(num, numberSeenLeaders);
-
+ assertThat(deadline.isOverdue())
+ .as("Did not complete the leader reelection in time.")
+ .isFalse();
+ assertThat(num).isEqualTo(numberSeenLeaders);
} finally {
if (leaderRetrievalService != null) {
leaderRetrievalService.stop();
@@ -279,7 +244,6 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
}
- @Nonnull
private String createAddress(int i) {
return LEADER_ADDRESS + "_" + i;
}
@@ -290,7 +254,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
* successfully register at ZooKeeper and take part in the leader election.
*/
@Test
- public void testZooKeeperReelectionWithReplacement() throws Exception {
+ void testZooKeeperReelectionWithReplacement() throws Exception {
int num = 3;
int numTries = 30;
@@ -303,15 +267,13 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
try {
leaderRetrievalService =
- ZooKeeperUtils.createLeaderRetrievalService(
- curatorFrameworkWrapper.asCuratorFramework());
+ ZooKeeperUtils.createLeaderRetrievalService(createZooKeeperClient());
leaderRetrievalService.start(listener);
for (int i = 0; i < num; i++) {
leaderElectionService[i] =
- ZooKeeperUtils.createLeaderElectionService(
- curatorFrameworkWrapper.asCuratorFramework());
+ ZooKeeperUtils.createLeaderElectionService(createZooKeeperClient());
contenders[i] =
new TestingContender(
LEADER_ADDRESS + "_" + i + "_0", leaderElectionService[i]);
@@ -333,15 +295,14 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
int index = Integer.parseInt(m.group(1));
int lastTry = Integer.parseInt(m.group(2));
- assertEquals(
- listener.getLeaderSessionID(), contenders[index].getLeaderSessionID());
+ assertThat(listener.getLeaderSessionID())
+ .isEqualTo(contenders[index].getLeaderSessionID());
// stop leader election service = revoke leadership
leaderElectionService[index].stop();
// create new leader election service which takes part in the leader election
leaderElectionService[index] =
- ZooKeeperUtils.createLeaderElectionService(
- curatorFrameworkWrapper.asCuratorFramework());
+ ZooKeeperUtils.createLeaderElectionService(createZooKeeperClient());
contenders[index] =
new TestingContender(
LEADER_ADDRESS + "_" + index + "_" + (lastTry + 1),
@@ -378,19 +339,18 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
try {
leaderElectionDriver =
createAndInitLeaderElectionDriver(
- curatorFrameworkWrapper.asCuratorFramework(), electionEventHandler);
+ createZooKeeperClient(), electionEventHandler);
electionEventHandler.waitForLeader();
final LeaderInformation confirmedLeaderInformation =
electionEventHandler.getConfirmedLeaderInformation();
- Assertions.assertThat(confirmedLeaderInformation.getLeaderAddress())
- .isEqualTo(LEADER_ADDRESS);
+ assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
// First update will successfully complete.
- Assertions.assertThat(leaderInformationConsumer.getFirstUpdateFuture())
+ assertThat(leaderInformationConsumer.getFirstUpdateFuture())
.succeedsWithin(5, TimeUnit.SECONDS);
// Wait for a while to make sure other updates don't appear.
- Assertions.assertThat(leaderInformationConsumer.getAnotherUpdateFuture())
+ assertThat(leaderInformationConsumer.getAnotherUpdateFuture())
.withFailMessage("Another leader information update is not expected.")
.failsWithin(5, TimeUnit.MILLISECONDS)
.withThrowableOfType(TimeoutException.class);
@@ -408,7 +368,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
* ZooKeeper.
*/
@Test
- public void testLeaderShouldBeCorrectedWhenOverwritten() throws Exception {
+ void testLeaderShouldBeCorrectedWhenOverwritten() throws Exception {
final String faultyContenderUrl = "faultyContender";
final TestingLeaderElectionEventHandler electionEventHandler =
@@ -425,12 +385,12 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
leaderElectionDriver =
createAndInitLeaderElectionDriver(
- curatorFrameworkWrapper.asCuratorFramework(), electionEventHandler);
+ createZooKeeperClient(), electionEventHandler);
electionEventHandler.waitForLeader();
final LeaderInformation confirmedLeaderInformation =
electionEventHandler.getConfirmedLeaderInformation();
- assertThat(confirmedLeaderInformation.getLeaderAddress(), is(LEADER_ADDRESS));
+ assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
anotherCuratorFrameworkWrapper =
ZooKeeperUtils.startCuratorFramework(
@@ -470,8 +430,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
// The faulty leader should be corrected on ZooKeeper
leaderRetrievalDriver =
- ZooKeeperUtils.createLeaderRetrievalDriverFactory(
- curatorFrameworkWrapper.asCuratorFramework())
+ ZooKeeperUtils.createLeaderRetrievalDriverFactory(createZooKeeperClient())
.createLeaderRetrievalDriver(
retrievalEventHandler, retrievalEventHandler::handleError);
@@ -479,12 +438,10 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
retrievalEventHandler.waitForNewLeader();
}
- assertThat(
- retrievalEventHandler.getLeaderSessionID(),
- is(confirmedLeaderInformation.getLeaderSessionID()));
- assertThat(
- retrievalEventHandler.getAddress(),
- is(confirmedLeaderInformation.getLeaderAddress()));
+ assertThat(retrievalEventHandler.getLeaderSessionID())
+ .isEqualTo(confirmedLeaderInformation.getLeaderSessionID());
+ assertThat(retrievalEventHandler.getAddress())
+ .isEqualTo(confirmedLeaderInformation.getLeaderAddress());
} finally {
electionEventHandler.close();
if (leaderElectionDriver != null) {
@@ -504,47 +461,44 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
* LeaderContender}.
*/
@Test
- public void testExceptionForwarding() throws Exception {
- LeaderElectionDriver leaderElectionDriver = null;
- final TestingLeaderElectionEventHandler electionEventHandler =
- new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
-
- CuratorFramework client = null;
+ void testExceptionForwarding() throws Exception {
final CreateBuilder mockCreateBuilder =
mock(CreateBuilder.class, Mockito.RETURNS_DEEP_STUBS);
final String exMsg = "Test exception";
final Exception testException = new Exception(exMsg);
- final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper =
- ZooKeeperUtils.startCuratorFramework(configuration, NoOpFatalErrorHandler.INSTANCE);
-
- try {
- client = spy(curatorFrameworkWrapper.asCuratorFramework());
+ try (final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper =
+ ZooKeeperUtils.startCuratorFramework(
+ configuration, NoOpFatalErrorHandler.INSTANCE)) {
+ CuratorFramework client = spy(curatorFrameworkWrapper.asCuratorFramework());
doAnswer(invocation -> mockCreateBuilder).when(client).create();
when(mockCreateBuilder
.creatingParentsIfNeeded()
- .withMode(Matchers.any(CreateMode.class))
+ .withMode(ArgumentMatchers.any(CreateMode.class))
.forPath(anyString(), any(byte[].class)))
.thenThrow(testException);
- leaderElectionDriver = createAndInitLeaderElectionDriver(client, electionEventHandler);
-
- electionEventHandler.waitForError();
-
- assertNotNull(electionEventHandler.getError());
- assertThat(
- ExceptionUtils.findThrowableWithMessage(electionEventHandler.getError(), exMsg)
- .isPresent(),
- is(true));
- } finally {
- electionEventHandler.close();
- if (leaderElectionDriver != null) {
- leaderElectionDriver.close();
- }
+ TestingLeaderElectionEventHandler electionEventHandler = null;
+ LeaderElectionDriver leaderElectionDriver = null;
+ try {
+ electionEventHandler = new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
+ leaderElectionDriver =
+ createAndInitLeaderElectionDriver(client, electionEventHandler);
+
+ electionEventHandler.waitForError();
+
+ assertThat(electionEventHandler.getError())
+ .isNotNull()
+ .satisfies(FlinkAssertions.anyCauseMatches(exMsg));
+ } finally {
+ if (leaderElectionDriver != null) {
+ leaderElectionDriver.close();
+ }
- if (curatorFrameworkWrapper != null) {
- curatorFrameworkWrapper.close();
+ if (electionEventHandler != null) {
+ electionEventHandler.close();
+ }
}
}
}
@@ -555,25 +509,27 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
* ephemeral nodes.
*/
@Test
- public void testEphemeralZooKeeperNodes() throws Exception {
- ZooKeeperLeaderElectionDriver leaderElectionDriver = null;
+ void testEphemeralZooKeeperNodes() throws Exception {
+ ZooKeeperLeaderElectionDriver leaderElectionDriver;
LeaderRetrievalDriver leaderRetrievalDriver = null;
final TestingLeaderElectionEventHandler electionEventHandler =
new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
final TestingLeaderRetrievalEventHandler retrievalEventHandler =
new TestingLeaderRetrievalEventHandler();
- CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper = null;
+ CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper;
CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper2 = null;
- NodeCache cache = null;
+ CuratorCache cache = null;
try {
curatorFrameworkWrapper =
ZooKeeperUtils.startCuratorFramework(
- configuration, testingFatalErrorHandlerResource.getFatalErrorHandler());
+ configuration,
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
curatorFrameworkWrapper2 =
ZooKeeperUtils.startCuratorFramework(
- configuration, testingFatalErrorHandlerResource.getFatalErrorHandler());
+ configuration,
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
leaderElectionDriver =
createAndInitLeaderElectionDriver(
@@ -585,14 +541,14 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
retrievalEventHandler, retrievalEventHandler::handleError);
cache =
- new NodeCache(
+ CuratorCache.build(
curatorFrameworkWrapper2.asCuratorFramework(),
leaderElectionDriver.getConnectionInformationPath());
- ExistsCacheListener existsListener = new ExistsCacheListener(cache);
- DeletedCacheListener deletedCacheListener = new DeletedCacheListener(cache);
-
- cache.getListenable().addListener(existsListener);
+ final ExistsCacheListener existsListener =
+ ExistsCacheListener.createWithNodeIsMissingValidation(
+ cache, leaderElectionDriver.getConnectionInformationPath());
+ cache.listenable().addListener(existsListener);
cache.start();
electionEventHandler.waitForLeader();
@@ -603,7 +559,10 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
existsFuture.get(timeout, TimeUnit.MILLISECONDS);
- cache.getListenable().addListener(deletedCacheListener);
+ final DeletedCacheListener deletedCacheListener =
+ DeletedCacheListener.createWithNodeExistValidation(
+ cache, leaderElectionDriver.getConnectionInformationPath());
+ cache.listenable().addListener(deletedCacheListener);
leaderElectionDriver.close();
@@ -633,7 +592,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
@Test
- public void testNotLeaderShouldNotCleanUpTheLeaderInformation() throws Exception {
+ void testNotLeaderShouldNotCleanUpTheLeaderInformation() throws Exception {
final TestingLeaderElectionEventHandler electionEventHandler =
new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
@@ -645,34 +604,30 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
try {
leaderElectionDriver =
createAndInitLeaderElectionDriver(
- curatorFrameworkWrapper.asCuratorFramework(), electionEventHandler);
+ createZooKeeperClient(), electionEventHandler);
electionEventHandler.waitForLeader();
final LeaderInformation confirmedLeaderInformation =
electionEventHandler.getConfirmedLeaderInformation();
- assertThat(confirmedLeaderInformation.getLeaderAddress(), is(LEADER_ADDRESS));
+ assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
// Leader is revoked
leaderElectionDriver.notLeader();
electionEventHandler.waitForRevokeLeader();
- assertThat(
- electionEventHandler.getConfirmedLeaderInformation(),
- is(LeaderInformation.empty()));
- // The data on ZooKeeper it not be cleared
+ assertThat(electionEventHandler.getConfirmedLeaderInformation())
+ .isEqualTo(LeaderInformation.empty());
+ // The data on ZooKeeper has not been cleared
leaderRetrievalDriver =
- ZooKeeperUtils.createLeaderRetrievalDriverFactory(
- curatorFrameworkWrapper.asCuratorFramework())
+ ZooKeeperUtils.createLeaderRetrievalDriverFactory(createZooKeeperClient())
.createLeaderRetrievalDriver(
retrievalEventHandler, retrievalEventHandler::handleError);
retrievalEventHandler.waitForNewLeader();
- assertThat(
- retrievalEventHandler.getLeaderSessionID(),
- is(confirmedLeaderInformation.getLeaderSessionID()));
- assertThat(
- retrievalEventHandler.getAddress(),
- is(confirmedLeaderInformation.getLeaderAddress()));
+ assertThat(retrievalEventHandler.getLeaderSessionID())
+ .isEqualTo(confirmedLeaderInformation.getLeaderSessionID());
+ assertThat(retrievalEventHandler.getAddress())
+ .isEqualTo(confirmedLeaderInformation.getLeaderAddress());
} finally {
electionEventHandler.close();
if (leaderElectionDriver != null) {
@@ -699,7 +654,7 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
new FlinkRuntimeException("testUnExpectedErrorForwarding");
final CuratorFrameworkFactory.Builder curatorFrameworkBuilder =
CuratorFrameworkFactory.builder()
- .connectString(testingServer.getConnectString())
+ .connectString(zooKeeperResource.getCustomExtension().getConnectString())
.retryPolicy(new ExponentialBackoffRetry(1, 0))
.aclProvider(
new ACLProvider() {
@@ -719,12 +674,10 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
try (CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper =
ZooKeeperUtils.startCuratorFramework(curatorFrameworkBuilder, fatalErrorHandler)) {
CuratorFramework clientWithErrorHandler = curatorFrameworkWrapper.asCuratorFramework();
- assertFalse(fatalErrorHandler.getErrorFuture().isDone());
+ assertThat(fatalErrorHandler.getErrorFuture()).isNotDone();
leaderElectionDriver =
createAndInitLeaderElectionDriver(clientWithErrorHandler, electionEventHandler);
- assertThat(
- fatalErrorHandler.getErrorFuture().join(),
- FlinkMatchers.containsCause(testException));
+ assertThat(fatalErrorHandler.getErrorFuture().get()).isEqualTo(testException);
} finally {
electionEventHandler.close();
@@ -734,13 +687,35 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
}
- private static class ExistsCacheListener implements NodeCacheListener {
+ private CuratorFramework createZooKeeperClient() {
+ return zooKeeperResource
+ .getCustomExtension()
+ .getZooKeeperClient(testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
+ }
+
+ private static class ExistsCacheListener implements CuratorCacheListener {
final CompletableFuture<Boolean> existsPromise = new CompletableFuture<>();
- final NodeCache cache;
+ final CuratorCache cache;
+
+ /**
+ * Factory method that's used to ensure consistency in the implementation. The method
+ * validates that the given node doesn't exist, yet.
+ *
+ * @throws IllegalStateException If the passed path is already present in the passed cache.
+ */
+ public static ExistsCacheListener createWithNodeIsMissingValidation(
+ CuratorCache cache, String path) {
+ Preconditions.checkState(
+ !cache.get(path).isPresent(),
+ "The given path %s should not lead to an already existing node. This listener will then check that the node was created.",
+ path);
+
+ return new ExistsCacheListener(cache);
+ }
- public ExistsCacheListener(final NodeCache cache) {
+ private ExistsCacheListener(final CuratorCache cache) {
this.cache = cache;
}
@@ -749,23 +724,30 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
@Override
- public void nodeChanged() throws Exception {
- ChildData data = cache.getCurrentData();
-
- if (data != null && !existsPromise.isDone()) {
+ public void event(Type type, ChildData oldData, ChildData data) {
+ if (type == Type.NODE_CREATED && data != null && !existsPromise.isDone()) {
existsPromise.complete(true);
- cache.getListenable().removeListener(this);
+ cache.listenable().removeListener(this);
}
}
}
- private static class DeletedCacheListener implements NodeCacheListener {
+ private static class DeletedCacheListener implements CuratorCacheListener {
final CompletableFuture<Boolean> deletedPromise = new CompletableFuture<>();
- final NodeCache cache;
+ final CuratorCache cache;
- public DeletedCacheListener(final NodeCache cache) {
+ public static DeletedCacheListener createWithNodeExistValidation(
+ CuratorCache cache, String path) {
+ Preconditions.checkState(
+ cache.get(path).isPresent(),
+ "The given path %s should lead to an already existing node. This listener will then check that the node was properly deleted.",
+ path);
+ return new DeletedCacheListener(cache);
+ }
+
+ private DeletedCacheListener(final CuratorCache cache) {
this.cache = cache;
}
@@ -774,12 +756,10 @@ public class ZooKeeperLeaderElectionTest extends TestLogger {
}
@Override
- public void nodeChanged() throws Exception {
- ChildData data = cache.getCurrentData();
-
- if (data == null && !deletedPromise.isDone()) {
+ public void event(Type type, ChildData oldData, ChildData data) {
+ if ((type == Type.NODE_DELETED || data == null) && !deletedPromise.isDone()) {
deletedPromise.complete(true);
- cache.getListenable().removeListener(this);
+ cache.listenable().removeListener(this);
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperMultipleComponentLeaderElectionDriverTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperMultipleComponentLeaderElectionDriverTest.java
index 822044c4dbf..8422b285c9f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperMultipleComponentLeaderElectionDriverTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/ZooKeeperMultipleComponentLeaderElectionDriverTest.java
@@ -29,14 +29,12 @@ import org.apache.flink.runtime.rest.util.NoOpFatalErrorHandler;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.runtime.zookeeper.ZooKeeperExtension;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.TestLoggerExtension;
import org.apache.flink.util.function.RunnableWithException;
import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFramework;
import org.apache.flink.shaded.guava30.com.google.common.collect.Iterables;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.time.Duration;
@@ -51,7 +49,6 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the {@link ZooKeeperMultipleComponentLeaderElectionDriver}. */
-@ExtendWith(TestLoggerExtension.class)
class ZooKeeperMultipleComponentLeaderElectionDriverTest {
private final ZooKeeperExtension zooKeeperExtension = new ZooKeeperExtension();
@@ -61,7 +58,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
new EachCallbackWrapper<>(zooKeeperExtension);
@Test
- public void testElectionDriverGainsLeadershipAtStartup() throws Exception {
+ void testElectionDriverGainsLeadershipAtStartup() throws Exception {
new Context() {
{
runTest(
@@ -73,7 +70,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testElectionDriverLosesLeadership() throws Exception {
+ void testElectionDriverLosesLeadership() throws Exception {
new Context() {
{
runTest(
@@ -87,7 +84,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testPublishLeaderInformation() throws Exception {
+ void testPublishLeaderInformation() throws Exception {
new Context() {
{
runTest(
@@ -122,7 +119,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testPublishEmptyLeaderInformation() throws Exception {
+ void testPublishEmptyLeaderInformation() throws Exception {
new Context() {
{
runTest(
@@ -161,7 +158,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testNonLeaderCannotPublishLeaderInformation() throws Exception {
+ void testNonLeaderCannotPublishLeaderInformation() throws Exception {
new Context() {
{
runTest(
@@ -198,7 +195,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testLeaderInformationChange() throws Exception {
+ void testLeaderInformationChange() throws Exception {
new Context() {
{
runTest(
@@ -233,7 +230,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testLeaderElectionWithMultipleDrivers() throws Exception {
+ void testLeaderElectionWithMultipleDrivers() throws Exception {
final CuratorFrameworkWithUnhandledErrorListener curatorFramework = startCuratorFramework();
try {
@@ -290,7 +287,7 @@ class ZooKeeperMultipleComponentLeaderElectionDriverTest {
}
@Test
- public void testLeaderConnectionInfoNodeRemovalLeadsToLeaderChangeWithEmptyLeaderInformation()
+ void testLeaderConnectionInfoNodeRemovalLeadsToLeaderChangeWithEmptyLeaderInformation()
throws Exception {
new Context() {
{
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalServiceTest.java
index 0e751e4db29..e4577951e91 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/DefaultLeaderRetrievalServiceTest.java
@@ -18,29 +18,24 @@
package org.apache.flink.runtime.leaderretrieval;
-import org.apache.flink.core.testutils.FlinkMatchers;
import org.apache.flink.runtime.leaderelection.DefaultLeaderElectionService;
import org.apache.flink.runtime.leaderelection.LeaderInformation;
import org.apache.flink.runtime.leaderelection.TestingListener;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.function.RunnableWithException;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.UUID;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.notNullValue;
-import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link DefaultLeaderElectionService}. */
-public class DefaultLeaderRetrievalServiceTest extends TestLogger {
+class DefaultLeaderRetrievalServiceTest {
private static final String TEST_URL = "akka//user/jobmanager";
@Test
- public void testNotifyLeaderAddress() throws Exception {
+ void testNotifyLeaderAddress() throws Exception {
new Context() {
{
runTest(
@@ -49,18 +44,17 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
LeaderInformation.known(UUID.randomUUID(), TEST_URL);
testingLeaderRetrievalDriver.onUpdate(newLeader);
testingListener.waitForNewLeader();
- assertThat(
- testingListener.getLeaderSessionID(),
- is(newLeader.getLeaderSessionID()));
- assertThat(
- testingListener.getAddress(), is(newLeader.getLeaderAddress()));
+ assertThat(testingListener.getLeaderSessionID())
+ .isEqualTo(newLeader.getLeaderSessionID());
+ assertThat(testingListener.getAddress())
+ .isEqualTo(newLeader.getLeaderAddress());
});
}
};
}
@Test
- public void testNotifyLeaderAddressEmpty() throws Exception {
+ void testNotifyLeaderAddressEmpty() throws Exception {
new Context() {
{
runTest(
@@ -72,15 +66,15 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
testingLeaderRetrievalDriver.onUpdate(LeaderInformation.empty());
testingListener.waitForEmptyLeaderInformation();
- assertThat(testingListener.getLeaderSessionID(), is(nullValue()));
- assertThat(testingListener.getAddress(), is(nullValue()));
+ assertThat(testingListener.getLeaderSessionID()).isNull();
+ assertThat(testingListener.getAddress()).isNull();
});
}
};
}
@Test
- public void testErrorForwarding() throws Exception {
+ void testErrorForwarding() throws Exception {
new Context() {
{
runTest(
@@ -90,16 +84,14 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
testingLeaderRetrievalDriver.onFatalError(testException);
testingListener.waitForError();
- assertThat(
- testingListener.getError(),
- FlinkMatchers.containsCause(testException));
+ assertThat(testingListener.getError()).hasCause(testException);
});
}
};
}
@Test
- public void testErrorIsIgnoredAfterBeingStop() throws Exception {
+ void testErrorIsIgnoredAfterBeingStop() throws Exception {
new Context() {
{
runTest(
@@ -109,14 +101,14 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
leaderRetrievalService.stop();
testingLeaderRetrievalDriver.onFatalError(testException);
- assertThat(testingListener.getError(), is(nullValue()));
+ assertThat(testingListener.getError()).isNull();
});
}
};
}
@Test
- public void testNotifyLeaderAddressOnlyWhenLeaderTrulyChanged() throws Exception {
+ void testNotifyLeaderAddressOnlyWhenLeaderTrulyChanged() throws Exception {
new Context() {
{
runTest(
@@ -124,22 +116,22 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
final LeaderInformation newLeader =
LeaderInformation.known(UUID.randomUUID(), TEST_URL);
testingLeaderRetrievalDriver.onUpdate(newLeader);
- assertThat(testingListener.getLeaderEventQueueSize(), is(1));
+ assertThat(testingListener.getLeaderEventQueueSize()).isOne();
// Same leader information should not be notified twice.
testingLeaderRetrievalDriver.onUpdate(newLeader);
- assertThat(testingListener.getLeaderEventQueueSize(), is(1));
+ assertThat(testingListener.getLeaderEventQueueSize()).isOne();
// Leader truly changed.
testingLeaderRetrievalDriver.onUpdate(
LeaderInformation.known(UUID.randomUUID(), TEST_URL + 1));
- assertThat(testingListener.getLeaderEventQueueSize(), is(2));
+ assertThat(testingListener.getLeaderEventQueueSize()).isEqualTo(2);
});
}
};
}
- private class Context {
+ private static class Context {
private final TestingLeaderRetrievalDriver.TestingLeaderRetrievalDriverFactory
leaderRetrievalDriverFactory =
new TestingLeaderRetrievalDriver.TestingLeaderRetrievalDriverFactory();
@@ -153,7 +145,7 @@ public class DefaultLeaderRetrievalServiceTest extends TestLogger {
leaderRetrievalService.start(testingListener);
testingLeaderRetrievalDriver = leaderRetrievalDriverFactory.getCurrentRetrievalDriver();
- assertThat(testingLeaderRetrievalDriver, is(notNullValue()));
+ assertThat(testingLeaderRetrievalDriver).isNotNull();
testMethod.run();
leaderRetrievalService.stop();
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/SettableLeaderRetrievalServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/SettableLeaderRetrievalServiceTest.java
index c2e329ef12b..e06a1388e01 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/SettableLeaderRetrievalServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/SettableLeaderRetrievalServiceTest.java
@@ -20,26 +20,24 @@ package org.apache.flink.runtime.leaderretrieval;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.TestingListener;
-import org.apache.flink.util.TestLogger;
-import org.junit.Before;
-import org.junit.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
-import static org.hamcrest.Matchers.equalTo;
-import static org.junit.Assert.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for {@link SettableLeaderRetrievalService}. */
-public class SettableLeaderRetrievalServiceTest extends TestLogger {
+class SettableLeaderRetrievalServiceTest {
private SettableLeaderRetrievalService settableLeaderRetrievalService;
- @Before
- public void setUp() {
+ @BeforeEach
+ void setUp() {
settableLeaderRetrievalService = new SettableLeaderRetrievalService();
}
@Test
- public void testNotifyListenerLater() throws Exception {
+ void testNotifyListenerLater() throws Exception {
final String localhost = "localhost";
settableLeaderRetrievalService.notifyListener(
localhost, HighAvailabilityServices.DEFAULT_LEADER_ID);
@@ -48,13 +46,13 @@ public class SettableLeaderRetrievalServiceTest extends TestLogger {
settableLeaderRetrievalService.start(listener);
listener.waitForNewLeader();
- assertThat(listener.getAddress(), equalTo(localhost));
- assertThat(
- listener.getLeaderSessionID(), equalTo(HighAvailabilityServices.DEFAULT_LEADER_ID));
+ assertThat(listener.getAddress()).isEqualTo(localhost);
+ assertThat(listener.getLeaderSessionID())
+ .isEqualTo(HighAvailabilityServices.DEFAULT_LEADER_ID);
}
@Test
- public void testNotifyListenerImmediately() throws Exception {
+ void testNotifyListenerImmediately() throws Exception {
final TestingListener listener = new TestingListener();
settableLeaderRetrievalService.start(listener);
@@ -63,8 +61,8 @@ public class SettableLeaderRetrievalServiceTest extends TestLogger {
localhost, HighAvailabilityServices.DEFAULT_LEADER_ID);
listener.waitForNewLeader();
- assertThat(listener.getAddress(), equalTo(localhost));
- assertThat(
- listener.getLeaderSessionID(), equalTo(HighAvailabilityServices.DEFAULT_LEADER_ID));
+ assertThat(listener.getAddress()).isEqualTo(localhost);
+ assertThat(listener.getLeaderSessionID())
+ .isEqualTo(HighAvailabilityServices.DEFAULT_LEADER_ID);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalConnectionHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalConnectionHandlingTest.java
index e8cda90a063..893b34a083a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalConnectionHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalConnectionHandlingTest.java
@@ -24,7 +24,6 @@ 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.util.TestLoggerExtension;
import org.apache.flink.util.function.BiConsumerWithException;
import org.apache.flink.util.function.FunctionWithException;
@@ -32,7 +31,6 @@ import org.apache.flink.shaded.curator5.org.apache.curator.framework.CuratorFram
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.api.extension.RegisterExtension;
import javax.annotation.Nullable;
@@ -53,7 +51,6 @@ import static org.assertj.core.api.Assertions.assertThat;
* Tests for the error handling in case of a suspended connection to the ZooKeeper instance when
* retrieving the leader information.
*/
-@ExtendWith(TestLoggerExtension.class)
class ZooKeeperLeaderRetrievalConnectionHandlingTest {
@RegisterExtension
@@ -67,7 +64,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
@Nullable private CuratorFramework zooKeeperClient;
@BeforeEach
- public void before() throws Exception {
+ void before() throws Exception {
zooKeeperClient =
zooKeeperExtension
.getCustomExtension()
@@ -81,7 +78,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
}
@Test
- public void testConnectionSuspendedHandlingDuringInitialization() throws Exception {
+ void testConnectionSuspendedHandlingDuringInitialization() throws Exception {
testWithQueueLeaderElectionListener(
queueLeaderElectionListener ->
ZooKeeperUtils.createLeaderRetrievalDriverFactory(zooKeeperClient)
@@ -107,7 +104,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
}
@Test
- public void testConnectionSuspendedHandling() throws Exception {
+ void testConnectionSuspendedHandling() throws Exception {
testWithQueueLeaderElectionListener(
queueLeaderElectionListener ->
new ZooKeeperLeaderRetrievalDriver(
@@ -141,7 +138,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
}
@Test
- public void testSuspendedConnectionDoesNotClearLeaderInformationIfClearanceOnLostConnection()
+ void testSuspendedConnectionDoesNotClearLeaderInformationIfClearanceOnLostConnection()
throws Exception {
testWithQueueLeaderElectionListener(
queueLeaderElectionListener ->
@@ -176,7 +173,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
}
@Test
- public void testSameLeaderAfterReconnectTriggersListenerNotification() throws Exception {
+ void testSameLeaderAfterReconnectTriggersListenerNotification() throws Exception {
testWithQueueLeaderElectionListener(
queueLeaderElectionListener ->
new ZooKeeperLeaderRetrievalDriver(
@@ -235,7 +232,7 @@ class ZooKeeperLeaderRetrievalConnectionHandlingTest {
}
@Test
- public void testNewLeaderAfterReconnectTriggersListenerNotification() throws Exception {
+ void testNewLeaderAfterReconnectTriggersListenerNotification() throws Exception {
testWithQueueLeaderElectionListener(
queueLeaderElectionListener ->
new ZooKeeperLeaderRetrievalDriver(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
index b9a884e42a8..d1b1b8a6b14 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderretrieval/ZooKeeperLeaderRetrievalTest.java
@@ -30,18 +30,16 @@ import org.apache.flink.runtime.rpc.AddressResolution;
import org.apache.flink.runtime.rpc.RpcSystem;
import org.apache.flink.runtime.testutils.ZooKeeperTestUtils;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
-import org.apache.flink.runtime.util.TestingFatalErrorHandlerResource;
+import org.apache.flink.runtime.util.TestingFatalErrorHandlerExtension;
import org.apache.flink.runtime.util.ZooKeeperUtils;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.TestLogger;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.curator.test.TestingServer;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.ClassRule;
-import org.junit.Rule;
-import org.junit.Test;
+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.io.IOException;
import java.net.InetAddress;
@@ -53,16 +51,20 @@ import java.net.UnknownHostException;
import java.time.Duration;
import java.util.concurrent.ScheduledExecutorService;
-import static org.junit.Assert.assertEquals;
+import static org.assertj.core.api.Assertions.assertThat;
/** Tests for the ZooKeeper based leader election and retrieval. */
-public class ZooKeeperLeaderRetrievalTest extends TestLogger {
+class ZooKeeperLeaderRetrievalTest {
private static final RpcSystem RPC_SYSTEM = RpcSystem.load();
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+ @RegisterExtension
+ private static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_RESOURCE =
+ TestingUtils.defaultExecutorExtension();
+
+ @RegisterExtension
+ private final TestingFatalErrorHandlerExtension testingFatalErrorHandlerResource =
+ new TestingFatalErrorHandlerExtension();
private TestingServer testingServer;
@@ -70,12 +72,8 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
private HighAvailabilityServices highAvailabilityServices;
- @Rule
- public final TestingFatalErrorHandlerResource testingFatalErrorHandlerResource =
- new TestingFatalErrorHandlerResource();
-
- @Before
- public void before() throws Exception {
+ @BeforeEach
+ void before() throws Exception {
testingServer = ZooKeeperTestUtils.createAndStartZookeeperTestingServer();
config = new Configuration();
@@ -86,15 +84,16 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
highAvailabilityServices =
new ZooKeeperMultipleComponentLeaderElectionHaServices(
ZooKeeperUtils.startCuratorFramework(
- config, testingFatalErrorHandlerResource.getFatalErrorHandler()),
+ config,
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler()),
config,
EXECUTOR_RESOURCE.getExecutor(),
new VoidBlobStore(),
- testingFatalErrorHandlerResource.getFatalErrorHandler());
+ testingFatalErrorHandlerResource.getTestingFatalErrorHandler());
}
- @After
- public void after() throws Exception {
+ @AfterEach
+ void after() throws Exception {
if (highAvailabilityServices != null) {
highAvailabilityServices.closeAndCleanupAllData();
@@ -115,7 +114,7 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
* been written to ZooKeeper.
*/
@Test
- public void testConnectingAddressRetrievalWithDelayedLeaderElection() throws Exception {
+ void testConnectingAddressRetrievalWithDelayedLeaderElection() throws Exception {
Duration timeout = Duration.ofMinutes(1L);
long sleepingTime = 1000;
@@ -173,7 +172,8 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
new FindConnectingAddress(
timeout,
highAvailabilityServices.getJobManagerLeaderRetriever(
- HighAvailabilityServices.DEFAULT_JOB_ID));
+ HighAvailabilityServices.DEFAULT_JOB_ID,
+ "unused-default-address"));
thread = new Thread(findConnectingAddress);
@@ -196,15 +196,12 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
InetAddress result = findConnectingAddress.getInetAddress();
// check that we can connect to the localHost
- Socket socket = new Socket();
- try {
+ try (Socket socket = new Socket()) {
// port 0 = let the OS choose the port
SocketAddress bindP = new InetSocketAddress(result, 0);
// machine
socket.bind(bindP);
socket.connect(correctInetSocketAddress, 1000);
- } finally {
- socket.close();
}
} finally {
if (leaderElectionService != null) {
@@ -219,17 +216,17 @@ public class ZooKeeperLeaderRetrievalTest extends TestLogger {
* InetAddress.getLocalHost().
*/
@Test
- public void testTimeoutOfFindConnectingAddress() throws Exception {
+ void testTimeoutOfFindConnectingAddress() throws Exception {
Duration timeout = Duration.ofSeconds(1L);
LeaderRetrievalService leaderRetrievalService =
highAvailabilityServices.getJobManagerLeaderRetriever(
- HighAvailabilityServices.DEFAULT_JOB_ID);
+ HighAvailabilityServices.DEFAULT_JOB_ID, "unused-default-address");
InetAddress result =
LeaderRetrievalUtils.findConnectingAddress(
leaderRetrievalService, timeout, RPC_SYSTEM);
- assertEquals(InetAddress.getLocalHost(), result);
+ assertThat(InetAddress.getLocalHost()).isEqualTo(result);
}
static class FindConnectingAddress implements Runnable {