ZOOKEEPER-4308: Fix flaky test EagerACLFilterTest

There are several problems in this test:
* It uses `ParameterizedTest` which run tests in single jvm. But
  `ZooKeeperServer.enableEagerACLCheck` is `static` and loaded from env
  variable.
* It uses `assertNotSame` which assert on object reference equiality.
* It asserts on `zkLeader.getLastLoggedZxid()` while client connect to
  `connectedServer`. There is no happen-before between
  `zkLeader.getLastLoggedZxid()` and successful response from other
  server. The commit and response are routed to different servers and
  performed asynchronous in each server.

Author: Kezhu Wang <kezhuw@gmail.com>

Reviewers: maoling <maoling199210191@sina.com>, Mate Szalay-Beko <symat@apache.org>

Closes #1851 from kezhuw/ZOOKEEPER-4308-EagerACLFilterTest

zookeeper-server/src/main/java/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -110,7 +110,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
 
     // When enabled, will check ACL constraints appertained to the requests first,
     // before sending the requests to the quorum.
-    static final boolean enableEagerACLCheck;
+    static boolean enableEagerACLCheck;
 
     static final boolean skipACL;
 
@@ -157,6 +157,17 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         LOG.info("{} = {}", CLOSE_SESSION_TXN_ENABLED, closeSessionTxnEnabled);
     }
 
+    // @VisibleForTesting
+    public static boolean isEnableEagerACLCheck() {
+        return enableEagerACLCheck;
+    }
+
+    // @VisibleForTesting
+    public static void setEnableEagerACLCheck(boolean enabled) {
+        ZooKeeperServer.enableEagerACLCheck = enabled;
+        LOG.info("Update {} to {}", ENABLE_EAGER_ACL_CHECK, enabled);
+    }
+
     public static boolean isCloseSessionTxnEnabled() {
         return closeSessionTxnEnabled;
     }

zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/EagerACLFilterTest.java

@@ -19,14 +19,16 @@
 package org.apache.zookeeper.server.quorum;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
-import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CompletableFuture;
 import java.util.stream.Stream;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.TestableZooKeeper;
 import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
 import org.apache.zookeeper.test.QuorumBase;
@@ -38,7 +40,6 @@ import org.junit.jupiter.params.provider.MethodSource;
 
 public class EagerACLFilterTest extends QuorumBase {
 
-    protected final CountDownLatch callComplete = new CountDownLatch(1);
     protected boolean complete = false;
     protected static final String PARENT_PATH = "/foo";
     protected static final String CHILD_PATH = "/foo/bar";
@@ -48,7 +49,9 @@ public class EagerACLFilterTest extends QuorumBase {
     protected static final byte[] DATA = "Hint Water".getBytes();
     protected TestableZooKeeper zkClient;
     protected TestableZooKeeper zkClientB;
+    protected TestableZooKeeper zkLeaderClient;
     protected QuorumPeer zkLeader;
+    protected QuorumPeer zkConnected;
     protected ZooKeeperServer connectedServer;
 
     public static Stream<Arguments> data() {
@@ -70,6 +73,7 @@ public class EagerACLFilterTest extends QuorumBase {
 
     public void setUp(ServerState serverState, boolean checkEnabled) throws Exception {
         ensureCheck(checkEnabled);
+        CountdownWatcher leaderWatch = new CountdownWatcher();
         CountdownWatcher clientWatch = new CountdownWatcher();
         CountdownWatcher clientWatchB = new CountdownWatcher();
         super.setUp(true, true);
@@ -78,16 +82,31 @@ public class EagerACLFilterTest extends QuorumBase {
         int clientPort = Integer.parseInt(hostPort.split(":")[1]);
 
         zkLeader = getPeerList().get(getLeaderIndex());
-        connectedServer = getPeerByClientPort(clientPort).getActiveServer();
+        zkConnected = getPeerByClientPort(clientPort);
+        connectedServer = zkConnected.getActiveServer();
 
+        zkLeaderClient = createClient(leaderWatch, getPeersMatching(ServerState.LEADING));
         zkClient = createClient(clientWatch, hostPort);
         zkClientB = createClient(clientWatchB, hostPort);
         zkClient.addAuthInfo(AUTH_PROVIDER, AUTH);
         zkClientB.addAuthInfo(AUTH_PROVIDER, AUTHB);
+        leaderWatch.waitForConnected(CONNECTION_TIMEOUT);
         clientWatch.waitForConnected(CONNECTION_TIMEOUT);
         clientWatchB.waitForConnected(CONNECTION_TIMEOUT);
     }
 
+    void syncClient(ZooKeeper zk) {
+        CompletableFuture<Void> synced = new CompletableFuture<>();
+        zk.sync("/", (rc, path, ctx) -> {
+            if (rc == 0) {
+                synced.complete(null);
+            } else {
+                synced.completeExceptionally(KeeperException.create(KeeperException.Code.get(rc)));
+            }
+        }, null);
+        synced.join();
+    }
+
     @AfterEach
     public void tearDown() throws Exception {
         if (zkClient != null) {
@@ -102,19 +121,29 @@ public class EagerACLFilterTest extends QuorumBase {
     }
 
     private void ensureCheck(boolean enabled) {
-        if (enabled) {
-            System.setProperty(ZooKeeperServer.ENABLE_EAGER_ACL_CHECK, "true");
-        } else {
-            System.clearProperty(ZooKeeperServer.ENABLE_EAGER_ACL_CHECK);
-        }
+        ZooKeeperServer.setEnableEagerACLCheck(enabled);
     }
 
-    private void assertTransactionState(String condition, long lastxid, ServerState serverState, boolean checkEnabled) {
-        String assertion = String.format("Server State: %s Check Enabled: %s %s", serverState, checkEnabled, condition);
-        if (checkEnabled) {
-            assertEquals(lastxid, zkLeader.getLastLoggedZxid(), assertion);
+    private void assertTransactionState(String operation, QuorumPeer peer, long lastxid) {
+        if (peer == zkLeader && peer != zkConnected) {
+            // The operation is performed on no leader, but we are asserting on leader.
+            // There is no happen-before between `zkLeader.getLastLoggedZxid()` and
+            // successful response from other server. The commit and response are routed
+            // to different servers and performed asynchronous in each server. So we have
+            // to sync leader client to go through commit and response path in leader to
+            // build happen-before between `zkLeader.getLastLoggedZxid()` and side effect
+            // of previous operation.
+            syncClient(zkLeaderClient);
+        }
+        assertTrue(peer == zkLeader || peer == zkConnected);
+        boolean eagerACL = ZooKeeperServer.isEnableEagerACLCheck();
+        String assertion = String.format(
+                "Connecting: %s Checking: %s EagerACL: %s Operation: %s",
+                zkConnected.getPeerState(), peer.getPeerState(), eagerACL, operation);
+        if (eagerACL) {
+            assertEquals(lastxid, peer.getLastLoggedZxid(), assertion);
         } else {
-            assertNotSame(lastxid, zkLeader.getLastLoggedZxid(), assertion);
+            assertNotEquals(lastxid, peer.getLastLoggedZxid(), assertion);
         }
     }
 
@@ -144,15 +173,17 @@ public class EagerACLFilterTest extends QuorumBase {
     public void testCreateFail(ServerState serverState, boolean checkEnabled) throws Exception {
         setUp(serverState, checkEnabled);
         zkClient.create(PARENT_PATH, DATA, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
-        long lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
         try {
             zkClientB.create(CHILD_PATH, DATA, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+            fail("expect no auth");
         } catch (KeeperException.NoAuthException e) {
         }
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("Transaction state on Leader after failed create", lastxid, serverState, checkEnabled);
+        assertTransactionState("failed create", zkConnected, lastxid);
+        assertTransactionState("failed create", zkLeader, lastxid);
     }
 
     @ParameterizedTest
@@ -160,15 +191,17 @@ public class EagerACLFilterTest extends QuorumBase {
     public void testCreate2Fail(ServerState serverState, boolean checkEnabled) throws Exception {
         setUp(serverState, checkEnabled);
         zkClient.create(PARENT_PATH, DATA, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT, null);
-        long lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
         try {
             zkClientB.create(CHILD_PATH, DATA, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, null);
+            fail("expect no auth");
         } catch (KeeperException.NoAuthException e) {
         }
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("Transaction state on Leader after failed create2", lastxid, serverState, checkEnabled);
+        assertTransactionState("failed create2", zkConnected, lastxid);
+        assertTransactionState("failed create2", zkLeader, lastxid);
     }
 
     @ParameterizedTest
@@ -187,15 +220,17 @@ public class EagerACLFilterTest extends QuorumBase {
         setUp(serverState, checkEnabled);
         zkClient.create(PARENT_PATH, DATA, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT, null);
         zkClient.create(CHILD_PATH, DATA, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT, null);
-        long lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
         try {
             zkClientB.delete(CHILD_PATH, -1);
+            fail("expect no auth");
         } catch (KeeperException.NoAuthException e) {
         }
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("Transaction state on Leader after failed delete", lastxid, serverState, checkEnabled);
+        assertTransactionState("failed delete", zkConnected, lastxid);
+        assertTransactionState("failed delete", zkLeader, lastxid);
     }
 
     @ParameterizedTest
@@ -211,15 +246,17 @@ public class EagerACLFilterTest extends QuorumBase {
     public void testSetDataFail(ServerState serverState, boolean checkEnabled) throws Exception {
         setUp(serverState, checkEnabled);
         zkClient.create(PARENT_PATH, null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT, null);
-        long lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
         try {
             zkClientB.setData(PARENT_PATH, DATA, -1);
+            fail("expect no auth");
         } catch (KeeperException.NoAuthException e) {
         }
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("Transaction state on Leader after failed setData", lastxid, serverState, checkEnabled);
+        assertTransactionState("failed setData", zkConnected, lastxid);
+        assertTransactionState("failed setData", zkLeader, lastxid);
     }
 
     @ParameterizedTest
@@ -237,15 +274,17 @@ public class EagerACLFilterTest extends QuorumBase {
     public void testSetACLFail(ServerState serverState, boolean checkEnabled) throws Exception {
         setUp(serverState, checkEnabled);
         zkClient.create(PARENT_PATH, null, Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT, null);
-        long lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
         try {
             zkClientB.setACL(PARENT_PATH, Ids.READ_ACL_UNSAFE, -1);
-        } catch (KeeperException.NoAuthException e) {
+            fail("expect no auth");
+        } catch (KeeperException.NoAuthException ignored) {
         }
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("Transaction state on Leader after failed setACL", lastxid, serverState, checkEnabled);
+        assertTransactionState("failed setACL", zkConnected, lastxid);
+        assertTransactionState("failed setACL", zkLeader, lastxid);
     }
 
     @ParameterizedTest
@@ -253,12 +292,12 @@ public class EagerACLFilterTest extends QuorumBase {
     public void testBadACL(ServerState serverState, boolean checkEnabled) throws Exception {
         setUp(serverState, checkEnabled);
         CountdownWatcher cw = new CountdownWatcher();
-        TestableZooKeeper zk = createClient(cw, getPeersMatching(serverState));
-        long lastxid;
+        String addr = String.format("%s:%d", LOCALADDR, zkConnected.getClientPort());
+        TestableZooKeeper zk = createClient(cw, addr);
 
         cw.waitForConnected(CONNECTION_TIMEOUT);
 
-        lastxid = zkLeader.getLastLoggedZxid();
+        long lastxid = zkConnected.getLastLoggedZxid();
 
         try {
             zk.create("/acltest", new byte[0], Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
@@ -268,7 +307,8 @@ public class EagerACLFilterTest extends QuorumBase {
 
         assertEquals(0, connectedServer.getInProcess(), "OutstandingRequests not decremented");
 
-        assertTransactionState("zxid after invalid ACL", lastxid, serverState, checkEnabled);
+        assertTransactionState("invalid ACL", zkConnected, lastxid);
+        assertTransactionState("invalid ACL", zkLeader, lastxid);
     }
 
 }

zookeeper-server/src/test/java/org/apache/zookeeper/test/QuorumBase.java

@@ -47,7 +47,7 @@ public class QuorumBase extends ClientBase {
 
     private static final Logger LOG = LoggerFactory.getLogger(QuorumBase.class);
 
-    private static final String LOCALADDR = "127.0.0.1";
+    protected static final String LOCALADDR = "127.0.0.1";
 
     private File oracleDir;
     private static final String oraclePath_0 = "/oraclePath/0/mastership/";