package org.apache.helix.metaclient.impl.zk;

/*
 * 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.
 */

import java.util.ConcurrentModificationException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.NotImplementedException;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.helix.metaclient.api.AsyncCallback;
import org.apache.helix.metaclient.api.ChildChangeListener;
import org.apache.helix.metaclient.api.ConnectStateChangeListener;
import org.apache.helix.metaclient.api.DataChangeListener;
import org.apache.helix.metaclient.api.DataUpdater;
import org.apache.helix.metaclient.api.DirectChildChangeListener;
import org.apache.helix.metaclient.api.DirectChildSubscribeResult;
import org.apache.helix.metaclient.api.MetaClientInterface;
import org.apache.helix.metaclient.api.Op;
import org.apache.helix.metaclient.api.OpResult;
import org.apache.helix.metaclient.exception.MetaClientBadVersionException;
import org.apache.helix.metaclient.exception.MetaClientException;
import org.apache.helix.metaclient.exception.MetaClientNoNodeException;
import org.apache.helix.metaclient.exception.MetaClientNodeExistsException;
import org.apache.helix.metaclient.impl.zk.adapter.ChildListenerAdapter;
import org.apache.helix.metaclient.impl.zk.adapter.DataListenerAdapter;
import org.apache.helix.metaclient.impl.zk.adapter.DirectChildListenerAdapter;
import org.apache.helix.metaclient.impl.zk.adapter.StateChangeListenerAdapter;
import org.apache.helix.metaclient.impl.zk.adapter.ZkMetaClientCreateCallbackHandler;
import org.apache.helix.metaclient.impl.zk.adapter.ZkMetaClientDeleteCallbackHandler;
import org.apache.helix.metaclient.impl.zk.adapter.ZkMetaClientExistCallbackHandler;
import org.apache.helix.metaclient.impl.zk.adapter.ZkMetaClientGetCallbackHandler;
import org.apache.helix.metaclient.impl.zk.adapter.ZkMetaClientSetCallbackHandler;
import org.apache.helix.metaclient.impl.zk.factory.ZkMetaClientConfig;
import org.apache.helix.metaclient.impl.zk.util.ZkMetaClientUtil;
import org.apache.helix.zookeeper.api.client.ChildrenSubscribeResult;
import org.apache.helix.zookeeper.impl.client.ZkClient;
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
import org.apache.helix.zookeeper.zkclient.ZkConnection;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.helix.metaclient.impl.zk.util.ZkMetaClientUtil.separateIntoUniqueNodePaths;
import static org.apache.helix.metaclient.impl.zk.util.ZkMetaClientUtil.translateZkExceptionToMetaclientException;


public class ZkMetaClient<T> implements MetaClientInterface<T>, AutoCloseable {
  private static final Logger LOG = LoggerFactory.getLogger(ZkMetaClient.class);
  private final ZkClient _zkClient;
  private final long _initConnectionTimeout;
  private final long _reconnectTimeout;

  // After ZkClient gets disconnected from ZK server, it keeps retrying connection until connection
  // is re-established or ZkClient is closed. We need a separate thread to monitor ZkClient
  // reconnect and close ZkClient if it not able to reconnect within user specified timeout.
  private final ScheduledExecutorService _zkClientReconnectMonitor;
  private ScheduledFuture<?> _reconnectMonitorFuture;
  private ReconnectStateChangeListener _reconnectStateChangeListener;
  // Lock all activities related to ZkClient connection
  private ReentrantLock _zkClientConnectionMutex = new ReentrantLock();

  public ZkMetaClient(ZkMetaClientConfig config) {
    _initConnectionTimeout = config.getConnectionInitTimeoutInMillis();
    _reconnectTimeout = config.getMetaClientReconnectPolicy().getAutoReconnectTimeout();
    // TODO: Right new ZkClient reconnect using exp backoff with fixed max backoff interval. We should
    // Allow user to config reconnect policy
    _zkClient = new ZkClient(new ZkConnection(config.getConnectionAddress(), (int) config.getSessionTimeoutInMillis()),
        (int) _initConnectionTimeout, _reconnectTimeout /*use reconnect timeout for retry timeout*/,
        config.getZkSerializer(), config.getMonitorType(), config.getMonitorKey(), config.getMonitorInstanceName(),
        config.getMonitorRootPathOnly(), false, true);
    _zkClientReconnectMonitor = Executors.newSingleThreadScheduledExecutor();
    _reconnectStateChangeListener = new ReconnectStateChangeListener();
  }

  @Override
  public void create(String key, Object data) {
    try {
      create(key, data, EntryMode.PERSISTENT);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public void create(String key, Object data, EntryMode mode) {

    try {
      _zkClient.create(key, data, ZkMetaClientUtil.convertMetaClientMode(mode));
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    } catch (KeeperException e) {
      throw new MetaClientException(e);
    }
  }

  @Override
  public void recursiveCreate(String key, T data, EntryMode mode) {
    // Function named recursiveCreate to match naming scheme, but actual work is iterative
    iterativeCreate(key, data, mode, -1);
  }

  @Override
  public void recursiveCreateWithTTL(String key, T data, long ttl) {
    iterativeCreate(key, data, EntryMode.TTL, ttl);
  }

  private void iterativeCreate(String key, T data, EntryMode mode, long ttl) {
    List<String> nodePaths = separateIntoUniqueNodePaths(key);
    int i = 0;
    // Ephemeral nodes cant have children, so change mode when creating parents
    EntryMode parentMode = (EntryMode.EPHEMERAL.equals(mode) ?
        EntryMode.PERSISTENT : mode);

    // Iterate over paths, starting with full key then attempting each successive parent
    // Try /a/b/c, if parent /a/b, does not exist, then try to create parent, etc..
    while (i < nodePaths.size()) {
      // If parent exists or there is no parent node, then try to create the node
      // and break out of loop on successful create
      if (i == nodePaths.size() - 1 || _zkClient.exists(nodePaths.get(i+1))) {
        try {
          if (EntryMode.TTL.equals(mode)) {
            createWithTTL(nodePaths.get(i), data, ttl);
          } else {
            create(nodePaths.get(i), data, i == 0 ? mode : parentMode);
          }
        // Race condition may occur where a  node is created by another thread in between loops.
        // We should not throw error if this occurs for parent nodes, only for the full node path.
        } catch (MetaClientNodeExistsException e) {
          if (i == 0) {
            throw e;
          }
        }
        break;
      // Else try to create parent in next loop iteration
      } else {
       i++;
      }
    }

    // Reattempt creation of children that failed due to parent not existing
    while (--i >= 0) {
      try {
        if (EntryMode.TTL.equals(mode)) {
          createWithTTL(nodePaths.get(i), data, ttl);
        } else {
          create(nodePaths.get(i), data, i == 0 ? mode : parentMode);
        }
      // Catch same race condition as above
      } catch (MetaClientNodeExistsException e) {
        if (i == 0) {
          throw e;
        }
      }
    }
  }

  @Override
  public void createWithTTL(String key, T data, long ttl) {
    try {
      _zkClient.createPersistentWithTTL(key, data, ttl);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public void renewTTLNode(String key) {
    T oldData = get(key);
    if (oldData == null) {
      throw new MetaClientNoNodeException("Node at " + key + " does not exist.");
    }
    set(key, oldData, _zkClient.getStat(key).getVersion());
  }

  @Override
  public void set(String key, T data, int version) {
    try {
      _zkClient.writeData(key, data, version);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public T update(String key, DataUpdater<T> updater) {
    return update(key, updater, false, false);
  }

  @Override
  public T update(String key, DataUpdater<T> updater, boolean retryOnFailure, boolean createIfAbsent) {
    final int MAX_RETRY_ATTEMPTS = 3;
    int retryAttempts = 0;
    boolean retry;
    T updatedData = null;
    do {
      retry = false;
      retryAttempts++;
      try {
        ImmutablePair<T, Stat> tup = getDataAndStat(key);
        Stat stat = tup.right;
        T oldData = tup.left;
        T newData = updater.update(oldData);
        set(key, newData, stat.getVersion());
        updatedData = newData;
      } catch (MetaClientBadVersionException badVersionException) {
        // If exceeded max retry attempts, re-throw exception
        if (retryAttempts >= MAX_RETRY_ATTEMPTS) {
          LOG.error("Failed to update node at {} after {} attempts.", key, MAX_RETRY_ATTEMPTS);
          throw badVersionException;
        }
        // Retry on bad version
        retry = true;
      } catch (MetaClientNoNodeException noNodeException) {
        if (!createIfAbsent) {
          LOG.error("Failed to update node at {} as node does not exist. createIfAbsent was {}.", key, createIfAbsent);
          throw noNodeException;
        }
        // If node does not exist, attempt to create it - pass null to updater
        T newData = updater.update(null);
        if (newData != null) {
          try {
            create(key, newData);
            updatedData = newData;
          // If parent node for key does not exist, then updater will immediately fail due to uncaught NoNodeException
          } catch (MetaClientNodeExistsException nodeExistsException) {
            // If exceeded max retry attempts, cast to ConcurrentModification exception and re-throw.
            if (retryAttempts >= MAX_RETRY_ATTEMPTS) {
              LOG.error("Failed to update node at {} after {} attempts.", key, MAX_RETRY_ATTEMPTS);
              throw new ConcurrentModificationException("Failed to update node at " + key + " after " +
                  MAX_RETRY_ATTEMPTS + " attempts.", nodeExistsException);
            }
            // If node now exists, then retry update
            retry = true;
          } catch (ZkException e) {
            throw translateZkExceptionToMetaclientException(e);
          }
        }
      }
    } while (retryOnFailure && retry);
    return updatedData;
  }

  //TODO: Get Expiry Time in Stat
  @Override
  public Stat exists(String key) {
    org.apache.zookeeper.data.Stat zkStats;
    try {
      zkStats = _zkClient.getStat(key);
      if (zkStats == null) {
        return null;
      }
      return ZkMetaClientUtil.convertZkStatToStat(zkStats);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public T get(String key) {
    return _zkClient.readData(key, true);
  }

  @Override
  public ImmutablePair<T, Stat> getDataAndStat(final String key) {
    try {
      org.apache.zookeeper.data.Stat zkStat = new org.apache.zookeeper.data.Stat();
      T data = _zkClient.readData(key, zkStat);
      return ImmutablePair.of(data, ZkMetaClientUtil.convertZkStatToStat(zkStat));
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public List<String> getDirectChildrenKeys(String key) {
    try {
      return _zkClient.getChildren(key);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public int countDirectChildren(String key) {
    return _zkClient.countChildren(key);
  }

  @Override
  public boolean delete(String key) {
    try {
      return _zkClient.delete(key);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    }
  }

  @Override
  public boolean recursiveDelete(String key) {
    _zkClient.deleteRecursively(key);
    return true;
  }

  // In Current ZkClient, Async CRUD do auto retry when connection lost or session mismatch using
  // existing retry handling logic in zkClient. (defined in ZkAsyncCallbacks)
  // ZkClient execute async callbacks at zkClient main thead, retry is handles in a separate retry
  // thread. In our first version of implementation, we will keep similar behavior and have
  // callbacks executed in ZkClient event thread, and reuse zkClient retry logic.

  // It is highly recommended *NOT* to perform any blocking operation inside the callbacks.
  // If you block the thread the meta client won't process other events.

  // corresponding callbacks for each operation are invoked in order.
  @Override
  public void setAsyncExecPoolSize(int poolSize) {
    throw new UnsupportedOperationException("All async calls are executed in a single thread to maintain sequence.");
  }

  @Override
  public void asyncCreate(String key, Object data, EntryMode mode, AsyncCallback.VoidCallback cb) {
    CreateMode entryMode;
    try {
      entryMode = ZkMetaClientUtil.convertMetaClientMode(mode);
    } catch (ZkException | KeeperException e) {
      throw new MetaClientException(e);
    }
    _zkClient.asyncCreate(key, data, entryMode, new ZkMetaClientCreateCallbackHandler(cb));
  }

  @Override
  public void asyncUpdate(String key, DataUpdater<T> updater, AsyncCallback.DataCallback cb) {
    throw new NotImplementedException("Currently asyncUpdate is not supported in ZkMetaClient.");
    /*
     * TODO:  Only Helix has potential using this API as of now.  (ZkBaseDataAccessor.update())
     *  Will move impl from ZkBaseDataAccessor to here when retiring ZkBaseDataAccessor.
     */
  }

  @Override
  public void asyncGet(String key, AsyncCallback.DataCallback cb) {
    _zkClient.asyncGetData(key, new ZkMetaClientGetCallbackHandler(cb));
  }

  @Override
  public void asyncCountChildren(String key, AsyncCallback.DataCallback cb) {
    throw new NotImplementedException("Currently asyncCountChildren is not supported in ZkMetaClient.");
    /*
     * TODO:  Only Helix has potential using this API as of now. (ZkBaseDataAccessor.getChildren())
     *  Will move impl from ZkBaseDataAccessor to here when retiring ZkBaseDataAccessor.
     */

  }

  @Override
  public void asyncExist(String key, AsyncCallback.StatCallback cb) {
    _zkClient.asyncExists(key, new ZkMetaClientExistCallbackHandler(cb));
  }

  public void asyncDelete(String key, AsyncCallback.VoidCallback cb) {
    _zkClient.asyncDelete(key, new ZkMetaClientDeleteCallbackHandler(cb));
  }

  @Override
  public void asyncTransaction(Iterable<Op> ops, AsyncCallback.TransactionCallback cb) {
    throw new NotImplementedException("Currently asyncTransaction is not supported in ZkMetaClient.");

    //TODO: There is no active use case for Async transaction.
  }

  @Override
  public void asyncSet(String key, T data, int version, AsyncCallback.StatCallback cb) {
    _zkClient.asyncSetData(key, data, version, new ZkMetaClientSetCallbackHandler(cb));
  }

  @Override
  public void connect() {
    try {
      _zkClientConnectionMutex.lock();
      _zkClient.connect(_initConnectionTimeout, _zkClient);
      // register _reconnectStateChangeListener as state change listener to react to ZkClient connect
      // state change event. When ZkClient disconnected from ZK, it still auto reconnect until
      // ZkClient is closed or connection re-established.
      // We will need to close ZkClient when user set retry connection timeout.
      _zkClient.subscribeStateChanges(_reconnectStateChangeListener);
    } catch (ZkException e) {
      throw translateZkExceptionToMetaclientException(e);
    } finally {
      _zkClientConnectionMutex.unlock();
    }
  }

  @Override
  public void disconnect() {
    cleanUpAndClose(true, true);
    _zkClientReconnectMonitor.shutdownNow();
  }

  @Override
  public ConnectState getClientConnectionState() {
    return null;
  }

  @Override
  public boolean subscribeDataChange(String key, DataChangeListener listener, boolean skipWatchingNonExistNode) {
    _zkClient.subscribeDataChanges(key, new DataListenerAdapter(listener));
    return true;
  }

  @Override
  public DirectChildSubscribeResult subscribeDirectChildChange(String key, DirectChildChangeListener listener,
      boolean skipWatchingNonExistNode) {
    ChildrenSubscribeResult result =
        _zkClient.subscribeChildChanges(key, new DirectChildListenerAdapter(listener), skipWatchingNonExistNode);
    return new DirectChildSubscribeResult(result.getChildren(), result.isInstalled());
  }

  @Override
  public boolean subscribeStateChanges(ConnectStateChangeListener listener) {
    _zkClient.subscribeStateChanges(new StateChangeListenerAdapter(listener));
    return true;
  }

  @Override
  public boolean subscribeChildChanges(String key, ChildChangeListener listener, boolean skipWatchingNonExistNode) {
    if (skipWatchingNonExistNode && exists(key) == null) {
      return false;
    }
    _zkClient.subscribePersistRecursiveListener(key, new ChildListenerAdapter(listener));
    return true;
  }

  @Override
  public void unsubscribeDataChange(String key, DataChangeListener listener) {
    _zkClient.unsubscribeDataChanges(key, new DataListenerAdapter(listener));
  }

  @Override
  public void unsubscribeDirectChildChange(String key, DirectChildChangeListener listener) {
    _zkClient.unsubscribeChildChanges(key, new DirectChildListenerAdapter(listener));
  }

  // TODO: add impl and remove UnimplementedException
  @Override
  public void unsubscribeChildChanges(String key, ChildChangeListener listener) {
    _zkClient.unsubscribePersistRecursiveListener(key, new ChildListenerAdapter(listener));
  }

  @Override
  public void unsubscribeConnectStateChanges(ConnectStateChangeListener listener) {
    _zkClient.subscribeStateChanges(new StateChangeListenerAdapter(listener));
  }

  @Override
  public boolean waitUntilExists(String key, TimeUnit timeUnit, long time) {
    return false;
  }

  @Override
  public boolean[] create(List<String> key, List<T> data, List<EntryMode> mode) {
    return new boolean[0];
  }

  @Override
  public boolean[] create(List<String> key, List<T> data) {
    return new boolean[0];
  }

  @Override
  public boolean[] delete(List<String> keys) {
    return new boolean[0];
  }

  @Override
  public List<Stat> exists(List<String> keys) {
    return null;
  }

  @Override
  public List<T> get(List<String> keys) {
    return null;
  }

  @Override
  public List<T> update(List<String> keys, List<DataUpdater<T>> updater) {
    return null;
  }

  @Override
  public boolean[] set(List<String> keys, List<T> datas, List<Integer> version) {
    return new boolean[0];
  }

  @Override
  public void close() {
    disconnect();
  }

  @Override
  public boolean isClosed() {
    return _zkClient.isClosed();
  }

  @Override
  public List<OpResult> transactionOP(Iterable<Op> ops) {
    // Convert list of MetaClient Ops to Zk Ops
    List<org.apache.zookeeper.Op> zkOps = ZkMetaClientUtil.metaClientOpsToZkOps(ops);
    // Execute Zk transactional support
    List<org.apache.zookeeper.OpResult> zkResult = _zkClient.multi(zkOps);
    // Convert list of Zk OpResults to MetaClient OpResults
    return ZkMetaClientUtil.zkOpResultToMetaClientOpResults(zkResult);
  }

  @Override
  public byte[] serialize(T data, String path) {
    return _zkClient.serialize(data, path);
  }

  @Override
  public T deserialize(byte[] bytes, String path) {
    return _zkClient.deserialize(bytes, path);
  }

  /**
   * A clean up method called when connect state change or MetaClient is closing.
   * @param cancel If we want to cancel the reconnect monitor thread.
   * @param close If we want to close ZkClient.
   */
  private void cleanUpAndClose(boolean cancel, boolean close) {
    _zkClientConnectionMutex.lock();
    try {
      if (close && !_zkClient.isClosed()) {
        _zkClient.close();
        // TODO: need to unsubscribe all persist watcher from ZK
        // Add this in ZkClient when persist watcher change is in
        // Also need to manually send CLOSED state change to state
        // change listener (in change adapter)
        LOG.info("ZkClient is closed");
      }

      if (cancel && _reconnectMonitorFuture != null) {
        _reconnectMonitorFuture.cancel(true);
        LOG.info("ZkClient reconnect monitor thread is canceled");
      }
    } finally {
      _zkClientConnectionMutex.unlock();
    }
  }

  /**
   * MetaClient uses Helix ZkClient (@see org.apache.helix.zookeeper.impl.client.ZkClient) to connect
   * to ZK. Current implementation of ZkClient auto-reconnects infinitely. We use monitor thread
   * in ZkMetaClient to monitor reconnect status and close ZkClient when the client still is in
   * disconnected state when it reach reconnect timeout.
   *
   *
   * case 1: Start the monitor thread when ZkMetaClient gets disconnected even to check connect state
   *         when timeout reached. If not re-connected when timed out, kill the monitor thread
   *         and close ZkClient.
   * [MetaClient thread]        ---------------------------------------------------------------
   *                              ( When disconnected, schedule a event
   *                              to check connect state after timeout)
   * [Reconnect monitor thread]          --------------------------------------
   *                                   ^                                     |  not reconnected when timed out
   *                                  /                                      |
   *                                 | disconnected event                    v
   * [ZkClient]               -------X---------------------------------------X zkClient.close()
   * [ZkClient exp back              |         X            X
   *  -off retry connection]         |--------|--------------|--------------
   *
   *
   * case 2: Start the monitor thread when ZkMetaClient gets disconnected even to check connect state
   *         when timeout reached. If re-connected before timed out, cancel the delayed monitor thread.
   *
   * [MetaClient thread]       ---------------------------------------------------------------
   *                            (cancel scheduled task when reconnected)
   * [Reconnect monitor]               ---------------------------------X
   *                                  ^                                ^
   *                                 /                                /
   *                                | disconnected event             |  reconnected event
   * [ZkClient]                -----X------------------------------------------------------
   * [ZkClient exp back             |        X                      Y  Reconnected before timed out
   *  -off retry connection]        |--------| ---------------------|
   *
   *
   * case 3: Start the monitor thread when ZkMetaClient gets disconnected even to check connect state
   *         when timeout reached. If re-connected errored, kill the monitor thread  and cancel the
   *         delayed monitor thread.
   * [MetaClient thread]       ---------------------------------------------------------------
   *                          (cancel scheduled task and close ZkClient when reconnected error)
   * [Reconnect monitor]              ----------------------------------X
   *                                 ^                               ^  |
   *                                /                           err /   |
   *                               | disconnected event            |    v close ZkClient
   * [ZkClient]               -----X-------------------------------X ---X
   * [ZkClient exp back            |        X                     ^ Reconnect error
   *  -off retry connection]       |--------| --------------------|
   *
   */

  private class ReconnectStateChangeListener implements IZkStateListener {
    // Schedule a monitor to track ZkClient auto reconnect when Disconnected
    // Cancel the monitor thread when connected.
    @Override
    public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
      if (state == Watcher.Event.KeeperState.Disconnected) {                        // ------case 1
        // Expired. start a new event monitoring retry
        _zkClientConnectionMutex.lockInterruptibly();
        try {
          if (_reconnectMonitorFuture == null || _reconnectMonitorFuture.isCancelled()
              || _reconnectMonitorFuture.isDone()) {
            _reconnectMonitorFuture = _zkClientReconnectMonitor.schedule(() -> {
              if (!_zkClient.getConnection().getZookeeperState().isConnected()) {
                cleanUpAndClose(false, true);
              }
            }, _reconnectTimeout, TimeUnit.MILLISECONDS);
            LOG.info("ZkClient is Disconnected, schedule a reconnect monitor after {}", _reconnectTimeout);
          }
        } finally {
          _zkClientConnectionMutex.unlock();
        }
      } else if (state == Watcher.Event.KeeperState.SyncConnected
          || state == Watcher.Event.KeeperState.ConnectedReadOnly) {               // ------ case 2
        cleanUpAndClose(true, false);
        LOG.info("ZkClient is SyncConnected, reconnect monitor thread is canceled (if any)");
      }
    }

    // Cancel the monitor thread when connected.
    @Override
    public void handleNewSession(String sessionId) throws Exception {             // ------ case 2
      cleanUpAndClose(true, false);
      LOG.info("New session initiated in ZkClient, reconnect monitor thread is canceled (if any)");
    }

    // Cancel the monitor thread and close ZkClient when connect error.
    @Override
    public void handleSessionEstablishmentError(Throwable error) throws Exception {    // -- case 3
      cleanUpAndClose(true, true);
      LOG.info("New session initiated in ZkClient, reconnect monitor thread is canceled (if any)");
    }
  }

  @VisibleForTesting
  ZkClient getZkClient() {
    return _zkClient;
  }
}
