You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "zentol (via GitHub)" <gi...@apache.org> on 2023/06/20 09:32:14 UTC
[GitHub] [flink] zentol commented on a diff in pull request #22830: [FLINK-30343][test] Migrates KubernetesLeaderElectionAndRetrievalITCase to use the MultipleComponentLeaderElectionDriver interface
zentol commented on code in PR #22830:
URL: https://github.com/apache/flink/pull/22830#discussion_r1234999783
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java:
##########
@@ -89,68 +92,67 @@ void testLeaderElectionAndRetrieval() throws Exception {
clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
closeables.add(configMapSharedWatcher);
- final TestingLeaderElectionEventHandler electionEventHandler =
- new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
- closeables.add(electionEventHandler);
+ final TestingLeaderElectionListener electionEventHandler =
+ new TestingLeaderElectionListener();
+ final TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
try {
- final KubernetesLeaderElectionDriver leaderElectionDriver =
- new KubernetesLeaderElectionDriver(
- flinkKubeClient,
- configMapSharedWatcher,
- EXECUTOR_EXTENSION.getExecutor(),
+ final KubernetesMultipleComponentLeaderElectionDriver leaderElectionDriver =
+ new KubernetesMultipleComponentLeaderElectionDriver(
new KubernetesLeaderElectionConfiguration(
configMapName, UUID.randomUUID().toString(), configuration),
+ flinkKubeClient,
electionEventHandler,
- electionEventHandler::handleError);
+ configMapSharedWatcher,
+ EXECUTOR_EXTENSION.getExecutor(),
+ fatalErrorHandler);
closeables.add(leaderElectionDriver);
- electionEventHandler.init(leaderElectionDriver);
-
- final Function<TestingLeaderRetrievalEventHandler, AutoCloseable>
- leaderRetrievalDriverFactory =
- leaderRetrievalEventHandler ->
- new KubernetesLeaderRetrievalDriver(
- configMapSharedWatcher,
- EXECUTOR_EXTENSION.getExecutor(),
- configMapName,
- leaderRetrievalEventHandler,
- KubernetesUtils::getLeaderInformationFromConfigMap,
- leaderRetrievalEventHandler::handleError);
+ final KubernetesMultipleComponentLeaderRetrievalDriverFactory driverFactory =
+ new KubernetesMultipleComponentLeaderRetrievalDriverFactory(
+ configMapSharedWatcher,
+ EXECUTOR_EXTENSION.getExecutor(),
+ configMapName,
+ contenderID);
final TestingLeaderRetrievalEventHandler firstLeaderRetrievalEventHandler =
new TestingLeaderRetrievalEventHandler();
- closeables.add(leaderRetrievalDriverFactory.apply(firstLeaderRetrievalEventHandler));
+ closeables.add(
+ driverFactory.createLeaderRetrievalDriver(
+ firstLeaderRetrievalEventHandler, fatalErrorHandler));
// Wait for the driver to obtain leadership.
- electionEventHandler.waitForLeader();
- final LeaderInformation confirmedLeaderInformation =
- electionEventHandler.getConfirmedLeaderInformation();
- assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
+ electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
+ final UUID leaderSessionID = UUID.randomUUID();
+ leaderElectionDriver.publishLeaderInformation(
+ contenderID, LeaderInformation.known(leaderSessionID, leaderAddress));
// Check if the leader retrieval driver is notified about the leader address
- awaitLeadership(firstLeaderRetrievalEventHandler, confirmedLeaderInformation);
+ awaitLeadership(firstLeaderRetrievalEventHandler, leaderSessionID, leaderAddress);
// Start a second leader retrieval that should be notified immediately because we
// already know who the leader is.
final TestingLeaderRetrievalEventHandler secondRetrievalEventHandler =
new TestingLeaderRetrievalEventHandler();
- closeables.add(leaderRetrievalDriverFactory.apply(secondRetrievalEventHandler));
- awaitLeadership(secondRetrievalEventHandler, confirmedLeaderInformation);
+ closeables.add(
+ driverFactory.createLeaderRetrievalDriver(
+ secondRetrievalEventHandler, fatalErrorHandler));
+ awaitLeadership(secondRetrievalEventHandler, leaderSessionID, leaderAddress);
} finally {
for (AutoCloseable closeable : closeables) {
closeable.close();
}
+
+ Thread.sleep(1000);
Review Comment:
add a comment?
##########
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesLeaderElectionAndRetrievalITCase.java:
##########
@@ -89,68 +92,67 @@ void testLeaderElectionAndRetrieval() throws Exception {
clusterId, LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
closeables.add(configMapSharedWatcher);
- final TestingLeaderElectionEventHandler electionEventHandler =
- new TestingLeaderElectionEventHandler(LEADER_ADDRESS);
- closeables.add(electionEventHandler);
+ final TestingLeaderElectionListener electionEventHandler =
+ new TestingLeaderElectionListener();
+ final TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
try {
- final KubernetesLeaderElectionDriver leaderElectionDriver =
- new KubernetesLeaderElectionDriver(
- flinkKubeClient,
- configMapSharedWatcher,
- EXECUTOR_EXTENSION.getExecutor(),
+ final KubernetesMultipleComponentLeaderElectionDriver leaderElectionDriver =
+ new KubernetesMultipleComponentLeaderElectionDriver(
new KubernetesLeaderElectionConfiguration(
configMapName, UUID.randomUUID().toString(), configuration),
+ flinkKubeClient,
electionEventHandler,
- electionEventHandler::handleError);
+ configMapSharedWatcher,
+ EXECUTOR_EXTENSION.getExecutor(),
+ fatalErrorHandler);
closeables.add(leaderElectionDriver);
- electionEventHandler.init(leaderElectionDriver);
-
- final Function<TestingLeaderRetrievalEventHandler, AutoCloseable>
- leaderRetrievalDriverFactory =
- leaderRetrievalEventHandler ->
- new KubernetesLeaderRetrievalDriver(
- configMapSharedWatcher,
- EXECUTOR_EXTENSION.getExecutor(),
- configMapName,
- leaderRetrievalEventHandler,
- KubernetesUtils::getLeaderInformationFromConfigMap,
- leaderRetrievalEventHandler::handleError);
+ final KubernetesMultipleComponentLeaderRetrievalDriverFactory driverFactory =
+ new KubernetesMultipleComponentLeaderRetrievalDriverFactory(
+ configMapSharedWatcher,
+ EXECUTOR_EXTENSION.getExecutor(),
+ configMapName,
+ contenderID);
final TestingLeaderRetrievalEventHandler firstLeaderRetrievalEventHandler =
new TestingLeaderRetrievalEventHandler();
- closeables.add(leaderRetrievalDriverFactory.apply(firstLeaderRetrievalEventHandler));
+ closeables.add(
+ driverFactory.createLeaderRetrievalDriver(
+ firstLeaderRetrievalEventHandler, fatalErrorHandler));
// Wait for the driver to obtain leadership.
- electionEventHandler.waitForLeader();
- final LeaderInformation confirmedLeaderInformation =
- electionEventHandler.getConfirmedLeaderInformation();
- assertThat(confirmedLeaderInformation.getLeaderAddress()).isEqualTo(LEADER_ADDRESS);
+ electionEventHandler.await(LeaderElectionEvent.IsLeaderEvent.class);
+ final UUID leaderSessionID = UUID.randomUUID();
+ leaderElectionDriver.publishLeaderInformation(
+ contenderID, LeaderInformation.known(leaderSessionID, leaderAddress));
// Check if the leader retrieval driver is notified about the leader address
- awaitLeadership(firstLeaderRetrievalEventHandler, confirmedLeaderInformation);
+ awaitLeadership(firstLeaderRetrievalEventHandler, leaderSessionID, leaderAddress);
// Start a second leader retrieval that should be notified immediately because we
// already know who the leader is.
final TestingLeaderRetrievalEventHandler secondRetrievalEventHandler =
new TestingLeaderRetrievalEventHandler();
- closeables.add(leaderRetrievalDriverFactory.apply(secondRetrievalEventHandler));
- awaitLeadership(secondRetrievalEventHandler, confirmedLeaderInformation);
+ closeables.add(
+ driverFactory.createLeaderRetrievalDriver(
+ secondRetrievalEventHandler, fatalErrorHandler));
+ awaitLeadership(secondRetrievalEventHandler, leaderSessionID, leaderAddress);
} finally {
for (AutoCloseable closeable : closeables) {
closeable.close();
}
+
+ Thread.sleep(1000);
flinkKubeClient.deleteConfigMap(configMapName).get();
}
}
private static void awaitLeadership(
- TestingLeaderRetrievalEventHandler handler, LeaderInformation leaderInformation)
+ TestingLeaderRetrievalEventHandler handler, UUID leaderSessionID, String leaderAddress)
Review Comment:
not sure about this signature change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org