tillrohrmann commented on a change in pull request #17485:
URL: https://github.com/apache/flink/pull/17485#discussion_r791741579
##########
File path:
flink-kubernetes/src/test/java/org/apache/flink/kubernetes/highavailability/KubernetesMultipleComponentLeaderElectionDriverTest.java
##########
@@ -56,59 +62,179 @@
new
TestExecutorExtension<>(Executors::newSingleThreadScheduledExecutor);
@Test
- public void testElectionDriverGainsLeadership() throws
InterruptedException {
- final Configuration configuration = new Configuration();
- configuration.setString(KubernetesConfigOptions.CLUSTER_ID,
CLUSTER_ID);
- final KubernetesLeaderElectionConfiguration
leaderElectionConfiguration =
- new KubernetesLeaderElectionConfiguration("foobar", "barfoo",
configuration);
-
- CompletableFuture<KubernetesLeaderElector.LeaderCallbackHandler>
- leaderCallbackHandlerFuture = new CompletableFuture<>();
- final FlinkKubeClient flinkKubeClient =
- TestingFlinkKubeClient.builder()
- .setCreateLeaderElectorFunction(
- (leaderConfig, callbackHandler) -> {
-
leaderCallbackHandlerFuture.complete(callbackHandler);
- return new TestingFlinkKubeClient
- .TestingKubernetesLeaderElector(
- leaderConfig, callbackHandler);
- })
- .build();
-
- final KubernetesConfigMapSharedWatcher configMapSharedWatcher =
- flinkKubeClient.createConfigMapSharedWatcher(
- KubernetesUtils.getConfigMapLabels(
- CLUSTER_ID,
LABEL_CONFIGMAP_TYPE_HIGH_AVAILABILITY));
-
- final TestingLeaderElectionListener leaderElectionListener =
- new TestingLeaderElectionListener();
-
- final KubernetesMultipleComponentLeaderElectionDriver
leaderElectionDriver =
- new KubernetesMultipleComponentLeaderElectionDriver(
- leaderElectionConfiguration,
- flinkKubeClient,
- leaderElectionListener,
- configMapSharedWatcher,
- testExecutorExtension.getExecutor(),
-
testingFatalErrorHandlerExtension.getTestingFatalErrorHandler());
-
- final KubernetesLeaderElector.LeaderCallbackHandler
leaderCallbackHandler =
- leaderCallbackHandlerFuture.join();
-
- leaderCallbackHandler.isLeader();
-
- leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
+ public void testElectionDriverGainsLeadership() throws Exception {
+ new TestFixture() {
+ {
+ runTest(
+ () -> {
+ leaderCallbackGrantLeadership();
+
leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
+ });
+ }
+ };
}
@Test
- public void testElectionDriverLosesLeadership() throws Exception {}
+ public void testElectionDriverLosesLeadership() throws Exception {
+ new TestFixture() {
+ {
+ runTest(
+ () -> {
+ leaderCallbackGrantLeadership();
+
leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
+ getLeaderCallback().notLeader();
+
leaderElectionListener.await(LeaderElectionEvent.NotLeaderEvent.class);
+ });
+ }
+ };
+ }
@Test
- public void testPublishLeaderInformation() throws Exception {}
+ public void testPublishLeaderInformation() throws Exception {
+ new TestFixture() {
+ {
+ runTest(
+ () -> {
+ leaderCallbackGrantLeadership();
+
leaderElectionListener.await(LeaderElectionEvent.IsLeaderEvent.class);
+ final LeaderInformation leaderInformation =
+ LeaderInformation.known(UUID.randomUUID(),
"localhost");
+ final String componentId = "foobar";
+
+ final DefaultLeaderRetrievalService
leaderRetrievalService =
+ new DefaultLeaderRetrievalService(
+ new
KubernetesMultipleComponentLeaderRetrievalDriverFactory(
+ getFlinkKubeClient(),
+
getConfigMapSharedWatcher(),
+
testExecutorExtension.getExecutor(),
+ LEADER_CONFIGMAP_NAME,
+ componentId));
+
+ final TestingListener leaderRetrievalListener =
new TestingListener();
+
leaderRetrievalService.start(leaderRetrievalListener);
+
+ leaderElectionDriver.publishLeaderInformation(
+ componentId, leaderInformation);
+
+ notifyLeaderRetrievalWatchOnModifiedConfigMap();
+
+ leaderRetrievalListener.waitForNewLeader(10_000L);
+ assertThat(leaderRetrievalListener.getLeader())
+ .isEqualTo(leaderInformation);
+ });
+ }
+ };
+ }
@Test
- public void testLeaderInformationChange() throws Exception {}
+ public void testLeaderInformationChangeNotifiesListener() throws Exception
{
+ new TestFixture() {
+ {
+ runTest(
+ () -> {
+ leaderCallbackGrantLeadership();
+ final String componentA = "componentA";
+ final LeaderInformation leaderInformationA =
+ LeaderInformation.known(UUID.randomUUID(),
"localhost");
+ final String componentB = "componentB";
+ final LeaderInformation leaderInformationB =
+ LeaderInformation.known(UUID.randomUUID(),
"localhost");
+ leaderElectionDriver.publishLeaderInformation(
+ componentA, leaderInformationA);
+ leaderElectionDriver.publishLeaderInformation(
+ componentB, leaderInformationB);
+
+ notifyLeaderElectionWatchOnModifiedConfigMap();
+
+ final
LeaderElectionEvent.AllKnownLeaderInformationEvent
+ allKnownLeaderInformationEvent =
+ leaderElectionListener.await(
+ LeaderElectionEvent
+
.AllKnownLeaderInformationEvent.class);
+
+ assertThat(
+ allKnownLeaderInformationEvent
+
.getLeaderInformationWithComponentIds())
+ .contains(
Review comment:
True. Will change it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
