From 579d9cc2e731742360d05d8ab27bfe6d1195fba2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=89=B6=E6=91=87=20=28=20Fu=20Yao=20=29?= Date: Thu, 24 Jun 2021 11:22:01 +0800 Subject: [PATCH 1/5] [ Issue 11032 ] Fixed flaky test ZKSessionTest.. #11032 In some cases, testing errors occurred due to Zookeeper's notification delay. - Waiting async notice until timeout. --- .../apache/pulsar/metadata/ZKSessionTest.java | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java index 24c4bd709007f..73adbd8b66896 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java @@ -22,13 +22,10 @@ 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; @@ -171,10 +168,12 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { e = sessionEvents.poll(10, TimeUnit.SECONDS); assertEquals(e, SessionEvent.SessionLost); - Awaitility.await().untilAsserted(() -> { - assertEquals(le1.getState(), LeaderElectionState.Leading); - }); - + // due to zookeeper async notice, we need to wait. + Awaitility.await() + .until(() -> le1.getState() == LeaderElectionState.Leading); + assertEquals(le1.getState(), LeaderElectionState.Leading); + Awaitility.await() + .until(()-> leaderElectionEvents.poll() == null); les = leaderElectionEvents.poll(); assertNull(les); @@ -183,9 +182,10 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { e = sessionEvents.poll(10, TimeUnit.SECONDS); assertEquals(e, SessionEvent.SessionReestablished); - Awaitility.await().untilAsserted(() -> { - assertEquals(le1.getState(), LeaderElectionState.Leading); - }); + // due to zookeeper async notice, we need to wait. + Awaitility.await() + .until(() -> le1.getState() == LeaderElectionState.Leading); + assertEquals(le1.getState(), LeaderElectionState.Leading); les = leaderElectionEvents.poll(); assertNull(les); From fdf6a3bc281decd6faa41614108c7fbf7ea88013 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=89=B6=E6=91=87=20=28=20Fu=20Yao=20=29?= Date: Thu, 24 Jun 2021 15:49:13 +0800 Subject: [PATCH 2/5] [ Issue 11032 ] Fixed flaky test ZKSessionTest Firstly,This test to validate leader election after session is recovered. I think the leaderElectionEvenets check is unstable in here. eg: 1.When the node state is late to update => node state = Leading, leaderElectionEvents is empty. 2.When a node has been elected as the leader => node sate = Leading, leaderElectionEvents = [noLeader,Leading/Following]. 3.Leader election may be triggered at any time.(It may be just after the election due to network delays). - Delete leaderElectionEvents null check. - Waiting for node can be leader after session expire. --- .../apache/pulsar/metadata/ZKSessionTest.java | 28 ++++++------------- 1 file changed, 9 insertions(+), 19 deletions(-) diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java index 73adbd8b66896..b6e7d104b8f4c 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java @@ -136,6 +136,7 @@ public void testReacquireLocksAfterSessionLost() throws Exception { @Test public void testReacquireLeadershipAfterSessionLost() throws Exception { + // --- init @Cleanup MetadataStoreExtended store = MetadataStoreExtended.create(zks.getConnectionString(), MetadataStoreConfig.builder() @@ -144,7 +145,6 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { BlockingQueue sessionEvents = new LinkedBlockingQueue<>(); store.registerSessionListener(sessionEvents::add); - BlockingQueue leaderElectionEvents = new LinkedBlockingQueue<>(); String path = newKey(); @@ -153,13 +153,13 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { @Cleanup LeaderElection 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); @@ -167,28 +167,18 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { e = sessionEvents.poll(10, TimeUnit.SECONDS); assertEquals(e, SessionEvent.SessionLost); - - // due to zookeeper async notice, we need to wait. - Awaitility.await() - .until(() -> le1.getState() == LeaderElectionState.Leading); - assertEquals(le1.getState(), LeaderElectionState.Leading); + // --- test le1 can be leader Awaitility.await() - .until(()-> leaderElectionEvents.poll() == null); - les = leaderElectionEvents.poll(); - assertNull(les); - + .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); - - // due to zookeeper async notice, we need to wait. + // ---- after 2000 millis, the le1 is also can be leader. + Thread.sleep(2_000); Awaitility.await() - .until(() -> le1.getState() == LeaderElectionState.Leading); - assertEquals(le1.getState(), LeaderElectionState.Leading); - les = leaderElectionEvents.poll(); - assertNull(les); - + .untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading)); + assertEquals(le1.getState(),LeaderElectionState.Leading); assertTrue(store.get(path).join().isPresent()); } } From 57de54e5d7b6e4c5a5b68d68bc3f058b5ef23834 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=89=B6=E6=91=87=20=28=20Fu=20Yao=20=29?= Date: Thu, 24 Jun 2021 16:06:49 +0800 Subject: [PATCH 3/5] [ Issue 11032 ] Fixed flaky test ZKSessionTest - remove redundant test --- .../src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java index b6e7d104b8f4c..6e126fdb2bb52 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java @@ -178,7 +178,6 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { Thread.sleep(2_000); Awaitility.await() .untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading)); - assertEquals(le1.getState(),LeaderElectionState.Leading); assertTrue(store.get(path).join().isPresent()); } } From fefacefae722407de369ef6e3d7cf006007f2cf7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=89=B6=E6=91=87=20=28=20Fu=20Yao=20=29?= Date: Thu, 24 Jun 2021 18:57:58 +0800 Subject: [PATCH 4/5] [ Issues 10786 ] Fixed flaky test testTransactionMetaStoreAssignAndFailover (#10786) Replaced check after sleep to `untilAsserted` and extended `atMost` time to 10 seconds(default of `Awaitility`). --- .../TransactionMetaStoreAssignmentTest.java | 25 ++++++++----------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/TransactionMetaStoreAssignmentTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/TransactionMetaStoreAssignmentTest.java index cc0e185c582ba..ce173ba63d5d5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/TransactionMetaStoreAssignmentTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/transaction/coordinator/TransactionMetaStoreAssignmentTest.java @@ -20,6 +20,7 @@ import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import org.apache.pulsar.broker.PulsarService; import org.awaitility.Awaitility; @@ -32,12 +33,13 @@ public class TransactionMetaStoreAssignmentTest extends TransactionMetaStoreTest @Test(groups = "broker") public void testTransactionMetaStoreAssignAndFailover() throws IOException { - int transactionMetaStoreCount = 0; - for (PulsarService pulsarService : pulsarServices) { - transactionMetaStoreCount += pulsarService.getTransactionMetadataStoreService().getStores().size(); - } - - Assert.assertEquals(transactionMetaStoreCount, 16); + Awaitility.await() + .untilAsserted(() -> { + int transactionMetaStoreCount = Arrays.stream(pulsarServices) + .mapToInt(pulsarService -> pulsarService.getTransactionMetadataStoreService().getStores().size()) + .sum(); + Assert.assertEquals(transactionMetaStoreCount, 16); + }); PulsarService crashedMetaStore = null; for (int i = pulsarServices.length - 1; i >= 0; i--) { @@ -62,16 +64,11 @@ public void testTransactionMetaStoreAssignAndFailover() throws IOException { Awaitility.await() .untilAsserted(() -> { - - int transactionMetaStoreCount2 = 0; - for (PulsarService pulsarService : pulsarServices) { - transactionMetaStoreCount2 += pulsarService.getTransactionMetadataStoreService().getStores() - .size(); - } - + int transactionMetaStoreCount2 = Arrays.stream(pulsarServices) + .mapToInt(pulsarService -> pulsarService.getTransactionMetadataStoreService().getStores().size()) + .sum(); Assert.assertEquals(transactionMetaStoreCount2, 16); }); - transactionCoordinatorClient.close(); } } From a481f98930fbe04f54e0605463f5fb7052ae3ac8 Mon Sep 17 00:00:00 2001 From: Qiang Zhao Date: Fri, 25 Jun 2021 16:10:04 +0800 Subject: [PATCH 5/5] [ Issue 11032 ] Fixed flaky test ZKSessionTest - Delete useless code. --- .../src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java index 6e126fdb2bb52..5f8c690752b4f 100644 --- a/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java +++ b/pulsar-metadata/src/test/java/org/apache/pulsar/metadata/ZKSessionTest.java @@ -174,8 +174,6 @@ public void testReacquireLeadershipAfterSessionLost() throws Exception { assertEquals(e, SessionEvent.Reconnected); e = sessionEvents.poll(10, TimeUnit.SECONDS); assertEquals(e, SessionEvent.SessionReestablished); - // ---- after 2000 millis, the le1 is also can be leader. - Thread.sleep(2_000); Awaitility.await() .untilAsserted(()-> assertEquals(le1.getState(),LeaderElectionState.Leading)); assertTrue(store.get(path).join().isPresent());