Enable ZooKeeper client to establish connection in read-only mode (#4244

)

### Motivation

If the system property `readonlymode.enabled` is set to true on a ZooKeeper server, read-only mode is enabled. Data can be read from the server in read-only mode even if that server is split from the quorum.
https://zookeeper.apache.org/doc/current/zookeeperAdmin.html#Experimental+Options%2FFeatures

To connect to the server in read-only mode, the client must also allow read-only mode. The `ZooKeeperClient` class in the bookkeeper repository also has an option called `allowReadOnlyMode`.
https://github.com/apache/bookkeeper/blob/15171e1904f7196d8e9f4116ab2aecdf582e0032/bookkeeper-server/src/main/java/org/apache/bookkeeper/zookeeper/ZooKeeperClient.java#L219-L222

However, even if this option is set to true, the connection to the server in read-only mode will actually fail. The cause is in the `ZooKeeperWatcherBase` class. When the `ZooKeeperWatcherBase` class receives the `SyncConnected` event, it releases `clientConnectLatch` and assumes that the connection is complete.
https://github.com/apache/bookkeeper/blob/15171e1904f7196d8e9f4116ab2aecdf582e0032/bookkeeper-server/src/main/java/org/apache/bookkeeper/zookeeper/ZooKeeperWatcherBase.java#L128-L144

However, if the server is in read-only mode, it will receive `ConnectedReadOnly` instead of `SyncConnected`. This causes the connection to time out without being completed.

### Changes

Modified the switch statement in the `ZooKeeperWatcherBase` class to release `clientConnectLatch` when `ConnectedReadOnly` is received if the `allowReadOnlyMode` option is true.

By the way, `allowReadOnlyMode` is never set to true in BookKeeper. So this change would be useless for BookKeeper. However, it is useful for Pulsar. Because Pulsar also uses `ZooKeeperWatcherBase` and needs to be able to connect to ZooKeeper in read-only mode.
https://github.com/apache/pulsar/blob/cba1600d0f6a82f1ea194f3214a80f283fe8dc27/pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/PulsarZooKeeperClient.java#L242-L244

bookkeeper-server/src/main/java/org/apache/bookkeeper/zookeeper/ZooKeeperClient.java

@@ -241,9 +241,9 @@ public ZooKeeperClient build() throws IOException, KeeperException, InterruptedE
 
  // Create a watcher manager
  StatsLogger watcherStatsLogger = statsLogger.scope("watcher");
- ZooKeeperWatcherBase watcherManager =
- null == watchers ? new ZooKeeperWatcherBase(sessionTimeoutMs, watcherStatsLogger) :
-  new ZooKeeperWatcherBase(sessionTimeoutMs, watchers, watcherStatsLogger);
+ ZooKeeperWatcherBase watcherManager = (null == watchers)
+ ? new ZooKeeperWatcherBase(sessionTimeoutMs, allowReadOnlyMode, watcherStatsLogger)
+ : new ZooKeeperWatcherBase(sessionTimeoutMs, allowReadOnlyMode, watchers, watcherStatsLogger);
  ZooKeeperClient client = new ZooKeeperClient(
  connectString,
  sessionTimeoutMs,

bookkeeper-server/src/main/java/org/apache/bookkeeper/zookeeper/ZooKeeperWatcherBase.java

@@ -44,6 +44,7 @@ public class ZooKeeperWatcherBase implements Watcher {
  .getLogger(ZooKeeperWatcherBase.class);
 
  private final int zkSessionTimeOut;
+ private final boolean allowReadOnlyMode;
  private volatile CountDownLatch clientConnectLatch = new CountDownLatch(1);
  private final CopyOnWriteArraySet<Watcher> childWatchers =
  new CopyOnWriteArraySet<Watcher>();
@@ -53,18 +54,20 @@ public class ZooKeeperWatcherBase implements Watcher {
  private final ConcurrentHashMap<EventType, Counter> eventCounters =
  new ConcurrentHashMap<EventType, Counter>();
 
- public ZooKeeperWatcherBase(int zkSessionTimeOut) {
- this(zkSessionTimeOut, NullStatsLogger.INSTANCE);
+ public ZooKeeperWatcherBase(int zkSessionTimeOut, boolean allowReadOnlyMode) {
+ this(zkSessionTimeOut, allowReadOnlyMode, NullStatsLogger.INSTANCE);
  }
 
- public ZooKeeperWatcherBase(int zkSessionTimeOut, StatsLogger statsLogger) {
- this(zkSessionTimeOut, new HashSet<Watcher>(), statsLogger);
+ public ZooKeeperWatcherBase(int zkSessionTimeOut, boolean allowReadOnlyMode, StatsLogger statsLogger) {
+ this(zkSessionTimeOut, allowReadOnlyMode, new HashSet<Watcher>(), statsLogger);
  }
 
  public ZooKeeperWatcherBase(int zkSessionTimeOut,
+ boolean allowReadOnlyMode,
  Set<Watcher> childWatchers,
  StatsLogger statsLogger) {
  this.zkSessionTimeOut = zkSessionTimeOut;
+ this.allowReadOnlyMode = allowReadOnlyMode;
  this.childWatchers.addAll(childWatchers);
  this.statsLogger = statsLogger;
  }
@@ -130,6 +133,14 @@ public void process(WatchedEvent event) {
  LOG.info("ZooKeeper client is connected now.");
  clientConnectLatch.countDown();
  break;
+ case ConnectedReadOnly:
+ if (allowReadOnlyMode) {
+ LOG.info("ZooKeeper client is connected in read-only mode now.");
+ clientConnectLatch.countDown();
+ } else {
+ LOG.warn("ZooKeeper client is connected in read-only mode, which is not allowed.");
+ }
+ break;
  case Disconnected:
  LOG.info("ZooKeeper client is disconnected from zookeeper now,"
  + " but it is OK unless we received EXPIRED event.");

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperClientZKSessionExpiry.java

@@ -51,7 +51,7 @@ public void run() {
  byte[] sessionPasswd = bkc.getZkHandle().getSessionPasswd();
 
  try {
- ZooKeeperWatcherBase watcher = new ZooKeeperWatcherBase(10000);
+ ZooKeeperWatcherBase watcher = new ZooKeeperWatcherBase(10000, false);
  ZooKeeper zk = new ZooKeeper(zkUtil.getZooKeeperConnectString(), 10000,
  watcher, sessionId, sessionPasswd);
  zk.close();

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/BookKeeperTest.java

@@ -1072,7 +1072,7 @@ protected ZooKeeper createZooKeeper() throws IOException {
  public void testZKConnectionLossForLedgerCreation() throws Exception {
  int zkSessionTimeOut = 10000;
  AtomicLong ledgerIdToInjectFailure = new AtomicLong(INVALID_LEDGERID);
- ZooKeeperWatcherBase zooKeeperWatcherBase = new ZooKeeperWatcherBase(zkSessionTimeOut,
+ ZooKeeperWatcherBase zooKeeperWatcherBase = new ZooKeeperWatcherBase(zkSessionTimeOut, false,
  NullStatsLogger.INSTANCE);
  MockZooKeeperClient zkFaultInjectionWrapper = new MockZooKeeperClient(zkUtil.getZooKeeperConnectString(),
  zkSessionTimeOut, zooKeeperWatcherBase, ledgerIdToInjectFailure);

bookkeeper-server/src/test/java/org/apache/bookkeeper/replication/TestReplicationWorker.java

@@ -1103,7 +1103,7 @@ public void testRWShutDownInTheCaseOfZKOperationFailures() throws Exception {
  * create MockZooKeeperClient instance and wait for it to be connected.
  */
  int zkSessionTimeOut = 10000;
- ZooKeeperWatcherBase zooKeeperWatcherBase = new ZooKeeperWatcherBase(zkSessionTimeOut,
+ ZooKeeperWatcherBase zooKeeperWatcherBase = new ZooKeeperWatcherBase(zkSessionTimeOut, false,
  NullStatsLogger.INSTANCE);
  MockZooKeeperClient zkFaultInjectionWrapper = new MockZooKeeperClient(zkUtil.getZooKeeperConnectString(),
  zkSessionTimeOut, zooKeeperWatcherBase);

bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperCluster.java

@@ -64,7 +64,7 @@ void sleepCluster(int time, TimeUnit timeUnit, CountDownLatch l)
  default void expireSession(ZooKeeper zk) throws Exception {
  long id = zk.getSessionId();
  byte[] password = zk.getSessionPasswd();
- ZooKeeperWatcherBase w = new ZooKeeperWatcherBase(10000);
+ ZooKeeperWatcherBase w = new ZooKeeperWatcherBase(10000, false);
  ZooKeeper zk2 = new ZooKeeper(getZooKeeperConnectString(), zk.getSessionTimeout(), w, id, password);
  w.waitForConnection();
  zk2.close();

bookkeeper-server/src/test/java/org/apache/bookkeeper/test/ZooKeeperClusterUtil.java

@@ -139,4 +139,12 @@ public void killCluster() throws Exception {
  public void sleepCluster(int time, TimeUnit timeUnit, CountDownLatch l) throws InterruptedException, IOException {
  throw new UnsupportedOperationException("sleepServer operation is not supported for ZooKeeperClusterUtil");
  }
+
+ public void stopPeer(int id) throws Exception {
+ quorumUtil.shutdown(id);
+ }
+
+ public void enableLocalSession(boolean localSessionEnabled) {
+ quorumUtil.enableLocalSession(localSessionEnabled);
+ }
 }

bookkeeper-server/src/test/java/org/apache/bookkeeper/zookeeper/TestZooKeeperClient.java

@@ -171,7 +171,7 @@ public void process(WatchedEvent event) {
  };
  final int timeout = 2000;
  ZooKeeperWatcherBase watcherManager =
- new ZooKeeperWatcherBase(timeout).addChildWatcher(testWatcher);
+ new ZooKeeperWatcherBase(timeout, false).addChildWatcher(testWatcher);
  List<Watcher> watchers = new ArrayList<Watcher>(1);
  watchers.add(testWatcher);
  ZooKeeperClient client = new ShutdownZkServerClient(
@@ -895,4 +895,32 @@ public void processResult(int rc, String path, Object ctx) {
  logger.info("Delete children from znode " + path);
  }
 
+ @Test
+ public void testAllowReadOnlyMode() throws Exception {
+ if (zkUtil instanceof ZooKeeperClusterUtil) {
+ System.setProperty("readonlymode.enabled", "true");
+ ((ZooKeeperClusterUtil) zkUtil).enableLocalSession(true);
+ zkUtil.restartCluster();
+ Thread.sleep(2000);
+ ((ZooKeeperClusterUtil) zkUtil).stopPeer(2);
+ ((ZooKeeperClusterUtil) zkUtil).stopPeer(3);
+ }
+
+ try (ZooKeeperClient client = ZooKeeperClient.newBuilder()
+ .connectString(zkUtil.getZooKeeperConnectString())
+ .sessionTimeoutMs(30000)
+ .watchers(new HashSet<Watcher>())
+ .operationRetryPolicy(retryPolicy)
+ .allowReadOnlyMode(true)
+ .build()) {
+ Assert.assertTrue("Client failed to connect a ZooKeeper in read-only mode.",
+ client.getState().isConnected());
+ } finally {
+ if (zkUtil instanceof ZooKeeperClusterUtil) {
+ System.setProperty("readonlymode.enabled", "false");
+ ((ZooKeeperClusterUtil) zkUtil).enableLocalSession(false);
+ }
+ }
+ }
+
 }