[ Issue 11032 ] Fixed flaky test ZKSessionTest..
Fixes
Master Issue:
Motivation
In some cases, testing errors occurred due to Zookeeper's notification delay.
Modifications
Waiting async notice until timeout.
@@ -22,13 +22,10 @@ import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
-
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
-
import lombok.Cleanup;
-
import org.apache.pulsar.metadata.api.MetadataStoreConfig;
import org.apache.pulsar.metadata.api.coordination.CoordinationService;
import org.apache.pulsar.metadata.api.coordination.LeaderElection;
@@ -139,6 +136,7 @@ public class ZKSessionTest extends BaseMetadataStoreTest {
@Test
public void testReacquireLeadershipAfterSessionLost() throws Exception {
+ // --- init
@Cleanup
MetadataStoreExtended store = MetadataStoreExtended.create(zks.getConnectionString(),
MetadataStoreConfig.builder()
@@ -147,7 +145,6 @@ public class ZKSessionTest extends BaseMetadataStoreTest {
BlockingQueue<SessionEvent> sessionEvents = new LinkedBlockingQueue<>();
store.registerSessionListener(sessionEvents::add);
-
BlockingQueue<LeaderElectionState> leaderElectionEvents = new LinkedBlockingQueue<>();
String path = newKey();
@@ -156,13 +153,13 @@ public class ZKSessionTest extends BaseMetadataStoreTest {
@Cleanup
LeaderElection<String> le1 = coordinationService.getLeaderElection(String.class, path,
leaderElectionEvents::add);
-
+ // --- test manual elect
le1.elect("value-1").join();
assertEquals(le1.getState(), LeaderElectionState.Leading);
LeaderElectionState les = leaderElectionEvents.poll(5, TimeUnit.SECONDS);
assertEquals(les, LeaderElectionState.Leading);
-
+ // --- expire session
zks.expireSession(((ZKMetadataStore) store).getZkSessionId());
SessionEvent e = sessionEvents.poll(5, TimeUnit.SECONDS);
@@ -170,25 +167,15 @@ public class ZKSessionTest extends BaseMetadataStoreTest {
e = sessionEvents.poll(10, TimeUnit.SECONDS);
assertEquals(e, SessionEvent.SessionLost);
-
- Awaitility.await().untilAsserted(() -> {
- assertEquals(le1.getState(), LeaderElectionState.Leading);
- });
-
- les = leaderElectionEvents.poll();
- assertNull(les);
-
+ // --- test le1 can be leader
+ Awaitility.await()
+ .untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading)); // reacquire leadership
e = sessionEvents.poll(10, TimeUnit.SECONDS);
assertEquals(e, SessionEvent.Reconnected);
e = sessionEvents.poll(10, TimeUnit.SECONDS);
assertEquals(e, SessionEvent.SessionReestablished);
-
- Awaitility.await().untilAsserted(() -> {
- assertEquals(le1.getState(), LeaderElectionState.Leading);
- });
- les = leaderElectionEvents.poll();
- assertNull(les);
-
+ Awaitility.await()
+ .untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading));
assertTrue(store.get(path).join().isPresent());
}
}
