ZOOKEEPER-2623: Forbid OpCode.check outside OpCode.multi

Individual `OpCode.check` will get `UnimplementedException`.

This is a squashed backport of several commits from master:
1. b31f776 (#1988)
2. dc99bd7 (#2067)
3. 66f9cc3 (#2104)

Author: kezhuw

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

@@ -295,8 +295,8 @@ static boolean isValid(int type) {
         // make sure this is always synchronized with Zoodefs!!
         switch (type) {
         case OpCode.notification:
-            return false;
         case OpCode.check:
+            return false;
         case OpCode.closeSession:
         case OpCode.create:
         case OpCode.create2:

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

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.test;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import java.io.File;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.TestableZooKeeper;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.client.ZKClientConfig;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.CheckVersionRequest;
+import org.apache.zookeeper.proto.ReplyHeader;
+import org.apache.zookeeper.proto.RequestHeader;
+import org.apache.zookeeper.server.ZKDatabase;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.apache.zookeeper.server.quorum.QuorumPeer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+public class CheckTest extends ClientBase {
+
+    @BeforeEach
+    public void setUp(TestInfo testInfo) throws Exception {
+        System.setProperty(ZKClientConfig.ZOOKEEPER_REQUEST_TIMEOUT, "2000");
+        if (testInfo.getDisplayName().contains("Cluster")) {
+            return;
+        }
+        super.setUp();
+    }
+
+    @AfterEach
+    public void tearDown(TestInfo testInfo) throws Exception {
+        System.clearProperty(ZKClientConfig.ZOOKEEPER_REQUEST_TIMEOUT);
+        if (testInfo.getDisplayName().contains("Cluster")) {
+            return;
+        }
+        super.tearDown();
+    }
+
+    @Override
+    public void setUp() throws Exception {
+    }
+
+    @Override
+    public void tearDown() throws Exception {
+    }
+
+    private static void checkVersion(TestableZooKeeper zk, String path, int version) throws Exception {
+        RequestHeader header = new RequestHeader();
+        header.setType(ZooDefs.OpCode.check);
+        CheckVersionRequest request = new CheckVersionRequest(path, version);
+        ReplyHeader replyHeader = zk.submitRequest(header, request, null, null);
+        if (replyHeader.getErr() != 0) {
+            throw KeeperException.create(KeeperException.Code.get(replyHeader.getErr()), path);
+        }
+    }
+
+    private void testOperations(TestableZooKeeper zk) throws Exception {
+        Stat stat = new Stat();
+        zk.getData("/", false, stat);
+        assertThrows(KeeperException.UnimplementedException.class, () -> checkVersion(zk, "/", -1));
+    }
+
+    @Test
+    public void testStandalone() throws Exception {
+        testOperations(createClient());
+    }
+
+    @Test
+    public void testStandaloneDatabaseReloadAfterCheck() throws Exception {
+        try {
+            testOperations(createClient());
+        } catch (Throwable ignored) {
+            // Ignore to test database reload after check
+        }
+        stopServer();
+        startServer();
+    }
+
+    @Test
+    public void testCluster() throws Exception {
+        QuorumBase qb = new QuorumBase();
+        try {
+            qb.setUp(true, true);
+            testOperations(qb.createClient(new CountdownWatcher(), QuorumPeer.ServerState.OBSERVING));
+            testOperations(qb.createClient(new CountdownWatcher(), QuorumPeer.ServerState.FOLLOWING));
+            testOperations(qb.createClient(new CountdownWatcher(), QuorumPeer.ServerState.LEADING));
+        } finally {
+            try {
+                qb.tearDown();
+            } catch (Exception ignored) {}
+        }
+    }
+
+    @Test
+    public void testClusterDatabaseReloadAfterCheck() throws Exception {
+        QuorumBase qb = new QuorumBase();
+        try {
+            qb.setUp(true, true);
+
+            // Get leader before possible damaging operations to
+            // reduce chance of leader migration and log truncation.
+            File dataDir = qb.getLeaderDataDir();
+            QuorumPeer leader = qb.getLeaderQuorumPeer();
+
+            try {
+                testOperations(qb.createClient(new CountdownWatcher(), QuorumPeer.ServerState.LEADING));
+            } catch (Throwable ignored) {
+                // Ignore to test database reload after check
+            }
+            qb.shutdown(leader);
+
+            FileTxnSnapLog txnSnapLog = new FileTxnSnapLog(dataDir, dataDir);
+            ZKDatabase database = new ZKDatabase(txnSnapLog);
+            database.loadDataBase();
+        } finally {
+            try {
+                qb.tearDown();
+            } catch (Exception ignored) {}
+        }
+    }
+}

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

@@ -197,6 +197,10 @@ protected TestableZooKeeper createClient(CountdownWatcher watcher) throws IOExce
     private List<ZooKeeper> allClients;
     private boolean allClientsSetup = false;
 
+    protected TestableZooKeeper createClient(String hp, int timeout) throws IOException, InterruptedException {
+        return createClient(new CountdownWatcher(), hp, timeout);
+    }
+
     protected TestableZooKeeper createClient(CountdownWatcher watcher, String hp) throws IOException, InterruptedException {
         return createClient(watcher, hp, CONNECTION_TIMEOUT);
     }

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

@@ -342,6 +342,21 @@ public QuorumPeer getLeaderQuorumPeer() {
       return null;
     }
 
+    public File getLeaderDataDir() {
+        if (s1.getPeerState() == ServerState.LEADING) {
+            return s1dir;
+        } else if (s2.getPeerState() == ServerState.LEADING) {
+            return s2dir;
+        } else if (s3.getPeerState() == ServerState.LEADING) {
+            return s3dir;
+        } else if (s4.getPeerState() == ServerState.LEADING) {
+            return s4dir;
+        } else if (s5.getPeerState() == ServerState.LEADING) {
+            return s5dir;
+        }
+        return null;
+    }
+
     public QuorumPeer getFirstObserver() {
       if (s1.getLearnerType() == LearnerType.OBSERVER) {
         return s1;