[flink] branch release-1.14 updated: [FLINK-26596][runtime][test] Adds leadership loss handling

[mapohl]

This is an automated email from the ASF dual-hosted git repository.

mapohl pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.14 by this push:
     new 143f8e0  [FLINK-26596][runtime][test] Adds leadership loss handling
143f8e0 is described below

commit 143f8e0b16f3127f5f70c7a827a7642b463e9493
Author: Matthias Pohl <matth...@ververica.com>
AuthorDate: Fri Mar 11 18:14:34 2022 +0100

    [FLINK-26596][runtime][test] Adds leadership loss handling
    
    If the ZK connection is flaky, we might collect another
    ZK connection loss. This is now handled properly.
---
 .../ZooKeeperLeaderRetrievalConnectionHandlingTest.java            | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

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 a9cc05a..5a8e02a 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
@@ -338,9 +338,10 @@ public class 
ZooKeeperLeaderRetrievalConnectionHandlingTest extends TestLogger {
             // check that we find the new leader information eventually
             CommonTestUtils.waitUntilCondition(
                     () -> {
-                        final CompletableFuture<String> 
afterConnectionReconnect =
-                                queueLeaderElectionListener.next();
-                        return 
afterConnectionReconnect.get().equals(newLeaderAddress);
+                        final String afterConnectionReconnect =
+                                queueLeaderElectionListener.next().get();
+                        return afterConnectionReconnect != null
+                                && 
afterConnectionReconnect.equals(newLeaderAddress);
                     },
                     Deadline.fromNow(Duration.ofSeconds(30L)));
         } finally {