package org.apache.helix.zookeeper.impl.client;

/*
 * 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.lang.management.ManagementFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import javax.management.MBeanServer;
import javax.management.ObjectName;

import org.apache.helix.monitoring.mbeans.MBeanRegistrar;
import org.apache.helix.monitoring.mbeans.MonitorDomainNames;
import org.apache.helix.zookeeper.constant.ZkSystemPropertyKeys;
import org.apache.helix.zookeeper.datamodel.SessionAwareZNRecord;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.apache.helix.zookeeper.datamodel.serializer.ZNRecordSerializer;
import org.apache.helix.zookeeper.exception.ZkClientException;
import org.apache.helix.zookeeper.impl.TestHelper;
import org.apache.helix.zookeeper.impl.ZkTestBase;
import org.apache.helix.zookeeper.impl.ZkTestHelper;
import org.apache.helix.zookeeper.zkclient.IZkDataListener;
import org.apache.helix.zookeeper.zkclient.IZkStateListener;
import org.apache.helix.zookeeper.zkclient.ZkConnection;
import org.apache.helix.zookeeper.zkclient.ZkServer;
import org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks;
import org.apache.helix.zookeeper.zkclient.exception.ZkException;
import org.apache.helix.zookeeper.zkclient.exception.ZkSessionMismatchedException;
import org.apache.helix.zookeeper.zkclient.exception.ZkTimeoutException;
import org.apache.helix.zookeeper.zkclient.metric.ZkClientMonitor;
import org.apache.helix.zookeeper.zkclient.metric.ZkClientPathMonitor;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Op;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.testng.Assert;
import org.testng.AssertJUnit;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

public class TestRawZkClient extends ZkTestBase {
  private final String TEST_TAG = "test_monitor";
  private final String TEST_ROOT = "/my_cluster/IDEALSTATES";

  private ZkClient _zkClient;

  @BeforeClass
  public void beforeClass() {
    _zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
  }

  @AfterClass
  public void afterClass() {
    _zkClient.delete(TEST_ROOT);
    _zkClient.deleteRecursively("/tmp");
    _zkClient.deleteRecursively("/my_cluster");
    _zkClient.close();
  }

  @Test()
  void testGetStat() {
    String path = "/tmp/getStatTest";
    _zkClient.deleteRecursively(path);

    Stat stat, newStat;
    stat = _zkClient.getStat(path);
    AssertJUnit.assertNull(stat);
    _zkClient.createPersistent(path, true);

    stat = _zkClient.getStat(path);
    AssertJUnit.assertNotNull(stat);

    newStat = _zkClient.getStat(path);
    AssertJUnit.assertEquals(stat, newStat);

    _zkClient.writeData(path, "Test");
    newStat = _zkClient.getStat(path);
    AssertJUnit.assertNotSame(stat, newStat);
  }

  /*
   * Tests subscribing state changes for helix's IZkStateListener.
   */
  @Test
  public void testSubscribeStateChanges() {
    int numListeners = _zkClient.numberOfListeners();
    List<IZkStateListener> listeners = new ArrayList<>();

    // Subscribe multiple listeners to test that listener's hashcode works as expected.
    // Each listener is subscribed and unsubscribed successfully.
    for (int i = 0; i < 3; i++) {
      IZkStateListener listener = new IZkStateListener() {
        @Override
        public void handleStateChanged(KeeperState state) {
          System.out.println("Handle new state: " + state);
        }

        @Override
        public void handleNewSession(final String sessionId) {
          System.out.println("Handle new session: " + sessionId);
        }

        @Override
        public void handleSessionEstablishmentError(Throwable error) {
          System.out.println("Handle session establishment error: " + error);
        }
      };

      _zkClient.subscribeStateChanges(listener);
      Assert.assertEquals(_zkClient.numberOfListeners(), ++numListeners);
      listeners.add(listener);
    }

    for (IZkStateListener listener : listeners) {
      _zkClient.unsubscribeStateChanges(listener);
      Assert.assertEquals(_zkClient.numberOfListeners(), --numListeners);
    }
  }

  /*
   * Tests state changes subscription for I0Itec's IZkStateListener.
   * This is a test for backward compatibility.
   *
   * TODO: remove this test when getting rid of I0Itec.
   */
  @Test
  public void testSubscribeStateChangesForI0ItecIZkStateListener() {
    int numListeners = _zkClient.numberOfListeners();
    List<org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener> listeners =
        new ArrayList<>();

    // Subscribe multiple listeners to test that listener's hashcode works as expected.
    // Each listener is subscribed and unsubscribed successfully.
    for (int i = 0; i < 3; i++) {
      org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener listener =
          new org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener() {
            @Override
            public void handleStateChanged(KeeperState state) {
              System.out.println("Handle new state: " + state);
            }

            @Override
            public void handleNewSession() {
              System.out.println("Handle new session: ");
            }

            @Override
            public void handleSessionEstablishmentError(Throwable error) {
              System.out.println("Handle session establishment error: " + error);
            }
          };

      _zkClient.subscribeStateChanges(listener);
      Assert.assertEquals(_zkClient.numberOfListeners(), ++numListeners);

      // Try to subscribe the listener again but number of listeners should not change because the
      // listener already exists.
      _zkClient.subscribeStateChanges(listener);
      Assert.assertEquals(_zkClient.numberOfListeners(), numListeners);

      listeners.add(listener);
    }

    for (org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener listener : listeners) {
      _zkClient.unsubscribeStateChanges(listener);
      Assert.assertEquals(_zkClient.numberOfListeners(), --numListeners);
    }
  }

  /*
   * Tests session expiry for I0Itec's IZkStateListener.
   * This is a test for backward compatibility.
   *
   * TODO: remove this test when getting rid of I0Itec.
   */
  @Test
  public void testSessionExpiryForI0IItecZkStateListener()
      throws Exception {
    org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener listener =
        new org.apache.helix.zookeeper.zkclient.deprecated.IZkStateListener() {

          @Override
          public void handleStateChanged(KeeperState state) {
            System.out.println("In Old connection New state " + state);
          }

          @Override
          public void handleNewSession() {
            System.out.println("In Old connection New session");
          }

          @Override
          public void handleSessionEstablishmentError(Throwable var1) {
          }
        };

    _zkClient.subscribeStateChanges(listener);
    ZkConnection connection = ((ZkConnection) _zkClient.getConnection());
    ZooKeeper zookeeper = connection.getZookeeper();
    long oldSessionId = zookeeper.getSessionId();
    System.out.println("old sessionId= " + oldSessionId);
    Watcher watcher = event -> System.out.println("In New connection In process event:" + event);
    ZooKeeper newZookeeper =
        new ZooKeeper(connection.getServers(), zookeeper.getSessionTimeout(), watcher,
            zookeeper.getSessionId(), zookeeper.getSessionPasswd());
    Thread.sleep(3000);
    System.out.println("New sessionId= " + newZookeeper.getSessionId());
    Thread.sleep(3000);
    newZookeeper.close();
    Thread.sleep(10000);
    connection = ((ZkConnection) _zkClient.getConnection());
    zookeeper = connection.getZookeeper();
    long newSessionId = zookeeper.getSessionId();
    System.out.println("After session expiry sessionId= " + newSessionId);
    _zkClient.unsubscribeStateChanges(listener);
  }

  @Test
  public void testZkClientMonitor()
      throws Exception {
    final String TEST_KEY = "testZkClientMonitor";
    ZkClient.Builder builder = new ZkClient.Builder();
    builder.setZkServer(ZkTestBase.ZK_ADDR).setMonitorKey(TEST_KEY).setMonitorType(TEST_TAG)
        .setMonitorRootPathOnly(false);
    ZkClient zkClient = builder.build();

    String TEST_DATA = "testData";
    String TEST_NODE = "/test_zkclient_monitor";
    String TEST_PATH = TEST_ROOT + TEST_NODE;
    final long TEST_DATA_SIZE = zkClient.serialize(TEST_DATA, TEST_PATH).length;

    if (_zkClient.exists(TEST_PATH)) {
      _zkClient.delete(TEST_PATH);
    }
    if (!_zkClient.exists(TEST_ROOT)) {
      _zkClient.createPersistent(TEST_ROOT, true);
    }

    MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();

    ObjectName name = MBeanRegistrar
        .buildObjectName(MonitorDomainNames.HelixZkClient.name(), ZkClientMonitor.MONITOR_TYPE,
            TEST_TAG, ZkClientMonitor.MONITOR_KEY, TEST_KEY);
    ObjectName rootname = MBeanRegistrar
        .buildObjectName(MonitorDomainNames.HelixZkClient.name(), ZkClientMonitor.MONITOR_TYPE,
            TEST_TAG, ZkClientMonitor.MONITOR_KEY, TEST_KEY, ZkClientPathMonitor.MONITOR_PATH,
            "Root");
    ObjectName idealStatename = MBeanRegistrar
        .buildObjectName(MonitorDomainNames.HelixZkClient.name(), ZkClientMonitor.MONITOR_TYPE,
            TEST_TAG, ZkClientMonitor.MONITOR_KEY, TEST_KEY, ZkClientPathMonitor.MONITOR_PATH,
            "IdealStates");
    Assert.assertTrue(beanServer.isRegistered(rootname));
    Assert.assertTrue(beanServer.isRegistered(idealStatename));

    Assert.assertEquals((long) beanServer.getAttribute(name, "DataChangeEventCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(name, "ExpiredSessionCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(name, "OutstandingRequestGauge"), 0);

    boolean verifyResult = TestHelper.verify(()->{
      return (long) beanServer.getAttribute(rootname, "ReadAsyncCounter") == 1;
    }, TestHelper.WAIT_DURATION);
    Assert.assertTrue(verifyResult, " did not see first sync() read");

    Assert.assertEquals((long) beanServer.getAttribute(name, "StateChangeEventCounter"), 1);

    long firstLatencyCounter =  (long) beanServer.getAttribute(rootname, "ReadTotalLatencyCounter");
    long firstReadLatencyGauge = (long) beanServer.getAttribute(rootname, "ReadLatencyGauge.Max");
    Assert.assertTrue(firstLatencyCounter >= 0);
    Assert.assertTrue(firstReadLatencyGauge >= 0);
    zkClient.exists(TEST_ROOT);

    Assert.assertTrue((long) beanServer.getAttribute(rootname, "ReadCounter") == 1);

    Assert.assertTrue((long) beanServer.getAttribute(rootname, "ReadTotalLatencyCounter") >= firstLatencyCounter);
    Assert.assertTrue((long) beanServer.getAttribute(rootname, "ReadLatencyGauge.Max") >= firstReadLatencyGauge);

    // Test create
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteBytesCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteBytesCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteTotalLatencyCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteLatencyGauge.Max"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteTotalLatencyCounter"),
        0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteLatencyGauge.Max"), 0);
    zkClient.create(TEST_PATH, TEST_DATA, CreateMode.PERSISTENT);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteBytesCounter"),
        TEST_DATA_SIZE);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteBytesCounter"),
        TEST_DATA_SIZE);
    long origWriteTotalLatencyCounter =
        (long) beanServer.getAttribute(rootname, "WriteTotalLatencyCounter");
    Assert.assertTrue(origWriteTotalLatencyCounter >= 0);
    Assert.assertTrue((long) beanServer.getAttribute(rootname, "WriteLatencyGauge.Max") >= 0);
    long origIdealStatesWriteTotalLatencyCounter =
        (long) beanServer.getAttribute(idealStatename, "WriteTotalLatencyCounter");
    Assert.assertTrue(origIdealStatesWriteTotalLatencyCounter >= 0);
    Assert.assertTrue((long) beanServer.getAttribute(idealStatename, "WriteLatencyGauge.Max") >= 0);

    // Test read
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadBytesCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadCounter"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadBytesCounter"), 0);
    long origReadTotalLatencyCounter =
        (long) beanServer.getAttribute(rootname, "ReadTotalLatencyCounter");
    long origIdealStatesReadTotalLatencyCounter =
        (long) beanServer.getAttribute(idealStatename, "ReadTotalLatencyCounter");
    Assert.assertEquals(origIdealStatesReadTotalLatencyCounter, 0);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadLatencyGauge.Max"), 0);
    zkClient.readData(TEST_PATH, new Stat());
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadCounter"), 2);
    Assert
        .assertEquals((long) beanServer.getAttribute(rootname, "ReadBytesCounter"), TEST_DATA_SIZE);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadBytesCounter"),
        TEST_DATA_SIZE);
    Assert.assertTrue((long) beanServer.getAttribute(rootname, "ReadTotalLatencyCounter")
        >= origReadTotalLatencyCounter);
    Assert.assertTrue((long) beanServer.getAttribute(idealStatename, "ReadTotalLatencyCounter")
        >= origIdealStatesReadTotalLatencyCounter);
    Assert.assertTrue((long) beanServer.getAttribute(idealStatename, "ReadLatencyGauge.Max") >= 0);
    zkClient.getChildren(TEST_PATH);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadCounter"), 3);
    Assert
        .assertEquals((long) beanServer.getAttribute(rootname, "ReadBytesCounter"), TEST_DATA_SIZE);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadCounter"), 2);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadBytesCounter"),
        TEST_DATA_SIZE);
    zkClient.getStat(TEST_PATH);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadCounter"), 4);
    Assert
        .assertEquals((long) beanServer.getAttribute(rootname, "ReadBytesCounter"), TEST_DATA_SIZE);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadCounter"), 3);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "ReadBytesCounter"),
        TEST_DATA_SIZE);
    zkClient.readDataAndStat(TEST_PATH, new Stat(), true);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadCounter"), 5);

    ZkAsyncCallbacks.ExistsCallbackHandler callbackHandler =
        new ZkAsyncCallbacks.ExistsCallbackHandler();
    zkClient.asyncExists(TEST_PATH, callbackHandler);
    callbackHandler.waitForSuccess();
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "ReadAsyncCounter"), 2);

    // Test write
    zkClient.writeData(TEST_PATH, TEST_DATA);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteCounter"), 2);
    Assert.assertEquals((long) beanServer.getAttribute(rootname, "WriteBytesCounter"),
        TEST_DATA_SIZE * 2);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteCounter"), 2);
    Assert.assertEquals((long) beanServer.getAttribute(idealStatename, "WriteBytesCounter"),
        TEST_DATA_SIZE * 2);
    Assert.assertTrue((long) beanServer.getAttribute(rootname, "WriteTotalLatencyCounter")
        >= origWriteTotalLatencyCounter);
    Assert.assertTrue((long) beanServer.getAttribute(idealStatename, "WriteTotalLatencyCounter")
        >= origIdealStatesWriteTotalLatencyCounter);

    // Verify the callback baseline count because of the data sync call.
    Assert.assertEquals((long) beanServer.getAttribute(name, "TotalCallbackCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(name, "TotalCallbackHandledCounter"), 1);

    // Test data change count
    final Lock lock = new ReentrantLock();
    final Condition callbackFinish = lock.newCondition();
    zkClient.subscribeDataChanges(TEST_PATH, new IZkDataListener() {
      @Override
      public void handleDataChange(String dataPath, Object data) {
        callbackLock();
      }

      @Override
      public void handleDataDeleted(String dataPath) {
        callbackLock();
      }

      private void callbackLock() {
        lock.lock();
        try {
          callbackFinish.signal();
        } finally {
          lock.unlock();
        }
      }
    });
    lock.lock();

    _zkClient.writeData(TEST_PATH, "Test");
    Assert.assertTrue(callbackFinish.await(10, TimeUnit.SECONDS));
    Assert.assertEquals((long) beanServer.getAttribute(name, "DataChangeEventCounter"), 1);
    Assert.assertEquals((long) beanServer.getAttribute(name, "OutstandingRequestGauge"), 0);
    Assert.assertEquals((long) beanServer.getAttribute(name, "TotalCallbackCounter"), 2);
    // Processing of the event might be slightly delayed.
    Assert.assertTrue(TestHelper
        .verify(() -> (long) beanServer.getAttribute(name, "TotalCallbackHandledCounter") == 2,
            500));
    Assert.assertEquals((long) beanServer.getAttribute(name, "PendingCallbackGauge"), 0);

    // Simulate a delayed callback
    int waitTime = 10;
    Thread.sleep(waitTime);
    lock.lock();
    zkClient.process(new WatchedEvent(Watcher.Event.EventType.NodeDataChanged, null, TEST_PATH));
    Assert.assertTrue(callbackFinish.await(10, TimeUnit.SECONDS));
    Assert.assertTrue(
        (long) beanServer.getAttribute(rootname, "DataPropagationLatencyGauge.Max") >= waitTime);

    _zkClient.delete(TEST_PATH);
  }

  @Test(dependsOnMethods = "testZkClientMonitor")
  void testPendingRequestGauge()
      throws Exception {
    final String TEST_KEY = "testPendingRequestGauge";

    final MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();
    final ObjectName name = MBeanRegistrar
        .buildObjectName(MonitorDomainNames.HelixZkClient.name(), ZkClientMonitor.MONITOR_TYPE,
            TEST_TAG, ZkClientMonitor.MONITOR_KEY, TEST_KEY);

    final int zkPort = TestHelper.getRandomPort();
    final String zkAddr = String.format("localhost:%d", zkPort);
    final ZkServer zkServer = startZkServer(zkAddr);

    try {
      ZkClient.Builder builder = new ZkClient.Builder();
      builder.setZkServer(zkAddr).setMonitorKey(TEST_KEY).setMonitorType(TEST_TAG)
          .setMonitorRootPathOnly(true);
      final ZkClient zkClient = builder.build();

      zkServer.shutdown();
      zkClient.waitForKeeperState(KeeperState.Disconnected, 5000, TimeUnit.MILLISECONDS);
      Assert.assertFalse(zkClient.waitUntilConnected(0, TimeUnit.MILLISECONDS));

      Assert.assertEquals((long) beanServer.getAttribute(name, "OutstandingRequestGauge"), 0);

      // Request a read in a separate thread. This will be a pending request
      ExecutorService executorService = Executors.newSingleThreadExecutor();
      executorService.submit(() -> {
        zkClient.exists(TEST_ROOT);
      });
      Assert.assertTrue(TestHelper
          .verify(() -> (long) beanServer.getAttribute(name, "OutstandingRequestGauge") == 1,
              1000));

      zkServer.start();
      Assert.assertTrue(zkClient.waitUntilConnected(5000, TimeUnit.MILLISECONDS));
      Assert.assertTrue(TestHelper
          .verify(() -> (long) beanServer.getAttribute(name, "OutstandingRequestGauge") == 0,
              2000));
      zkClient.close();
    } finally {
      zkServer.shutdown();
    }
  }

  /*
   * Tests session expiry and session expire counter for the helix's IZkStateListener.
   */
  @Test(dependsOnMethods = "testZkClientMonitor")
  void testSessionExpireCount() throws Exception {
    final String TEST_KEY = "testSessionExpireCount";

    final MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();
    final ObjectName name = MBeanRegistrar
        .buildObjectName(MonitorDomainNames.HelixZkClient.name(), ZkClientMonitor.MONITOR_TYPE,
            TEST_TAG, ZkClientMonitor.MONITOR_KEY, TEST_KEY);

    final int zkPort = TestHelper.getRandomPort();
    final String zkAddr = String.format("localhost:%d", zkPort);
    final ZkServer zkServer = startZkServer(zkAddr);

    try {
      ZkClient.Builder builder = new ZkClient.Builder();
      builder.setZkServer(zkAddr).setMonitorKey(TEST_KEY).setMonitorType(TEST_TAG)
          .setMonitorRootPathOnly(true);
      final ZkClient zkClient = builder.build();
      long lastSessionId = zkClient.getSessionId();
      long previousSessionExpiredCount =
          (long) beanServer.getAttribute(name, "ExpiredSessionCounter");

      for (int i = 0; i < 3; i++) {
        ZkTestHelper.expireSession(zkClient);
        long newSessionId = zkClient.getSessionId();
        Assert.assertTrue(newSessionId != lastSessionId,
            "New session id should not equal to expired session id.");
        lastSessionId = newSessionId;
      }
      Assert.assertEquals((long) beanServer.getAttribute(name, "ExpiredSessionCounter"),
          previousSessionExpiredCount + 3);

      zkClient.close();
    } finally {
      zkServer.shutdown();
    }
  }

  /*
   * This test checks that a valid session can successfully create an ephemeral node.
   */
  @Test
  public void testCreateEphemeralWithValidSession()
      throws Exception {

    final String originalSessionId = Long.toHexString(_zkClient.getSessionId());
    final String path = "/" + TestHelper.getTestMethodName();
    final String data = "Hello Helix";

    // Verify the node is not existed yet.
    Assert.assertFalse(_zkClient.exists(path));

    // Wait until the ZkClient has got a new session.
    Assert.assertTrue(TestHelper
        .verify(() -> _zkClient.getConnection().getZookeeperState().isConnected(), 1000L));

    try {
      // Create ephemeral node.
      _zkClient.createEphemeral(path, data, originalSessionId);
    } catch (Exception ex) {
      Assert.fail("Failed to create ephemeral node.", ex);
    }

    // Verify the node is created and its data is correct.
    Stat stat = new Stat();
    String nodeData = _zkClient.readData(path, stat, true);

    Assert.assertNotNull(nodeData, "Failed to create ephemeral node: " + path);
    Assert.assertEquals(nodeData, data, "Data is not correct.");
    Assert.assertTrue(stat.getEphemeralOwner() != 0L,
        "Ephemeral owner should NOT be zero because the node is an ephemeral node.");
    Assert.assertEquals(Long.toHexString(stat.getEphemeralOwner()), originalSessionId,
        "Ephemeral node is created by an unexpected session");

    // Delete the node to clean up, otherwise, the ephemeral node would be existed
    // until the session of its owner expires.
    _zkClient.delete(path);
  }

  /*
   * This test validates that when ZK_AUTOSYNC_ENABLED_DEFAULT is enabled, sync() would be issued
   * before handleNewSession. ZKclient would not see stale data.
   */
  @Test
  public void testAutoSyncWithNewSessionEstablishment() throws Exception {
    final String path = "/" + TestHelper.getTestMethodName();
    final String data = "Hello Helix 2";

    // Wait until the ZkClient has got a new session.
    Assert.assertTrue(_zkClient.waitUntilConnected(1, TimeUnit.SECONDS));

    try {
      // Create node.
      _zkClient.create(path, data, CreateMode.PERSISTENT);
    } catch (Exception ex) {
      Assert.fail("Failed to create ephemeral node.", ex);
    }

    // Expire the original session.
    ZkTestHelper.expireSession(_zkClient);

    // Verify the node is created and its data is correct.
    Stat stat = new Stat();
    String nodeData = null;
    try {
       nodeData = _zkClient.readData(path, stat, true);
    } catch (ZkException e) {
      Assert.fail("fail to read data");
    }
    Assert.assertEquals(nodeData, data, "Data is not correct.");
  }

  /*
   * This test checks that ephemeral creation fails because the expected zk session does not match
   * the actual zk session.
   * How this test does is:
   * 1. Creates a zk client and gets its original session id
   * 2. Expires the original session, and a new session will be created
   * 3. Tries to create ephemeral node with the original session
   * 4. ZkSessionMismatchedException is expected for the creation and ephemeral node is not created
   */
  @Test
  public void testCreateEphemeralWithMismatchedSession()
      throws Exception {
    final String methodName = TestHelper.getTestMethodName();

    final long originalSessionId = _zkClient.getSessionId();
    final String originalHexSessionId = Long.toHexString(originalSessionId);
    final String path = "/" + methodName;

    // Verify the node is not existed.
    Assert.assertFalse(_zkClient.exists(path));

    // Expire the original session.
    ZkTestHelper.expireSession(_zkClient);

    // Wait until the ZkClient has got a new session.
    Assert.assertTrue(TestHelper.verify(() -> {
      try {
        // New session id should not equal to expired session id.
        return _zkClient.getSessionId() != originalSessionId;
      } catch (ZkClientException ex) {
        return false;
      }
    }, 1000L));

    try {
      // Try to create ephemeral node with the original session.
      // This creation should NOT be successful because the original session is already expired.
      _zkClient.createEphemeral(path, "Hello Helix", originalHexSessionId);
      Assert.fail("Ephemeral node should not be created by the expired session.");
    } catch (ZkSessionMismatchedException expected) {
      // Expected because there is a session mismatch.
    } catch (Exception unexpected) {
      Assert.fail("Should not have thrown exception: " + unexpected);
    }

    // Verify the node is not created.
    Assert.assertFalse(_zkClient.exists(path));
  }

  /*
   * Tests that when trying to create an ephemeral node, if connection to zk service is lost,
   * ConnectionLossException will be thrown in retryUntilConnected and ephemeral creation will fail.
   * Because ConnectionLossException is not thrown in createEphemeral(), operation retry timeout is
   * set to 3 seconds and then ZkTimeoutException is thrown. And retry cause message is checked to
   * see if ConnectionLossException was thrown before retry.
   */
  @Test(timeOut = 5 * 60 * 1000L)
  public void testConnectionLossWhileCreateEphemeral()
      throws Exception {
    final String methodName = TestHelper.getTestMethodName();

    final ZkClient zkClient =
        new ZkClient.Builder().setZkServer(ZkTestBase.ZK_ADDR).setOperationRetryTimeout(3000L) // 3 seconds
            .build();

    final String expectedSessionId = Long.toHexString(zkClient.getSessionId());
    final String path = "/" + methodName;
    final String data = "data";

    Assert.assertFalse(zkClient.exists(path));

    // Shutdown zk server so zk operations will fail due to disconnection.
    TestHelper.stopZkServer(_zkServerMap.get(ZK_ADDR));

    try {
      final CountDownLatch countDownLatch = new CountDownLatch(1);
      final AtomicBoolean running = new AtomicBoolean(true);

      final Thread creationThread = new Thread(() -> {
        while (running.get()) {
          try {
            // Create ephemeral node in the expected session id.
            zkClient.createEphemeral(path, data, expectedSessionId);
          } catch (ZkTimeoutException e) {
            // Verify ConnectionLossException was thrown before retry and timeout.
            if (e.getMessage().endsWith("Retry was caused by "
                + KeeperException.ConnectionLossException.Code.CONNECTIONLOSS)) {
              running.set(false);
            }
          }
        }
        countDownLatch.countDown();
      });

      creationThread.start();

      final boolean creationThreadTerminated = countDownLatch.await(10L, TimeUnit.SECONDS);
      if (!creationThreadTerminated) {
        running.set(false);
        creationThread.join(5000L);
        Assert.fail("Failed to receive a ConnectionLossException after zookeeper has shutdown.");
      }
    } finally {
      zkClient.close();
      // Recover zk server.
      _zkServerMap.get(ZK_ADDR).start();
    }
  }

  /*
   * Tests that when trying to create an ephemeral node, if connection to zk service is lost,
   * the creation operation will keep retrying until connected and finally successfully create the
   * node.
   * How this test simulates the steps is:
   * 1. Shuts down zk server
   * 2. Starts a creation thread to create an ephemeral node in the original zk session
   * 3. Creation operation loses connection and keeps retrying
   * 4. Restarts zk server and Zk session is recovered
   * 5. zk client reconnects successfully and creates an ephemeral node
   */
  @Test(timeOut = 5 * 60 * 1000L)
  public void testRetryUntilConnectedAfterConnectionLoss()
      throws Exception {
    final String methodName = TestHelper.getTestMethodName();

    final String expectedSessionId = Long.toHexString(_zkClient.getSessionId());
    final String path = "/" + methodName;
    final String data = "data";

    Assert.assertFalse(_zkClient.exists(path));

    // Shutdown zk server so zk operations will fail due to disconnection.
    TestHelper.stopZkServer(_zkServerMap.get(ZK_ADDR));

    final CountDownLatch countDownLatch = new CountDownLatch(1);
    final AtomicBoolean running = new AtomicBoolean(true);

    final Thread creationThread = new Thread(() -> {
      while (running.get()) {
        // Create ephemeral node in the expected session id.
        System.out.println("Trying to create ephemeral node...");
        _zkClient.createEphemeral(path, data, expectedSessionId);
        System.out.println("Ephemeral node created.");
        running.set(false);
      }
      countDownLatch.countDown();
    });

    creationThread.start();
    // Keep creation thread retrying to connect for 10 seconds.
    System.out.println("Keep creation thread retrying to connect for 10 seconds...");
    TimeUnit.SECONDS.sleep(10);

    System.out.println("Restarting zk server...");
    _zkServerMap.get(ZK_ADDR).start();

    // Wait for creating ephemeral node successfully.
    final boolean creationThreadTerminated = countDownLatch.await(10, TimeUnit.SECONDS);
    if (!creationThreadTerminated) {
      running.set(false);
      creationThread.join(5000L);
      Assert.fail("Failed to reconnect to zk server and create ephemeral node"
          + " after zk server is recovered.");
    }

    Stat stat = new Stat();
    String nodeData = _zkClient.readData(path, stat, true);

    Assert.assertNotNull(nodeData, "Failed to create ephemeral node: " + path);
    Assert.assertEquals(nodeData, data, "Data is not correct.");
    Assert.assertTrue(stat.getEphemeralOwner() != 0L,
        "Ephemeral owner should NOT be zero because the node is an ephemeral node.");
    Assert.assertEquals(Long.toHexString(stat.getEphemeralOwner()), expectedSessionId,
        "Ephemeral node is created by an unexpected session");

    // Delete the node to clean up, otherwise, the ephemeral node would be existed until the session
    // of its owner expires.
    _zkClient.delete(path);
  }

  @Test
  public void testWaitForEstablishedSession() {
    ZkClient zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
    Assert.assertTrue(zkClient.waitForEstablishedSession(1, TimeUnit.SECONDS) != 0L);
    TestHelper.stopZkServer(_zkServerMap.get(ZK_ADDR));
    Assert.assertTrue(zkClient.waitForKeeperState(KeeperState.Disconnected, 1, TimeUnit.SECONDS));

    try {
      zkClient.waitForEstablishedSession(3, TimeUnit.SECONDS);
      Assert.fail("Connecting to zk server should time out and ZkTimeoutException is expected.");
    } catch (ZkTimeoutException expected) {
      // Because zk server is shutdown, zkClient should not connect to zk server and a
      // ZkTimeoutException should be thrown.
    }

    zkClient.close();
    // Recover zk server for later tests.
    _zkServerMap.get(ZK_ADDR).start();
  }

  @Test
  public void testAsyncWriteOperations() {
    ZkClient zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
    String originSizeLimit =
        System.getProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES);
    System.setProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES, "2000");
    try {
      zkClient.setZkSerializer(new ZNRecordSerializer());

      ZNRecord oversizeZNRecord = new ZNRecord("Oversize");
      char[] buff = new char[1024];
      Random ran = new Random();
      for (int i = 0; i < 1024; i++) {
        buff[i] = (char) (ran.nextInt(26) + 'a');
      }
      String buffString = new String(buff);
      for (int i = 0; i < 1024; i++) {
        oversizeZNRecord.setSimpleField(Integer.toString(i), buffString);
      }

      // ensure /tmp exists for the test
      if (!zkClient.exists("/tmp")) {
        zkClient.create("/tmp", null, CreateMode.PERSISTENT);
      }

      org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks.CreateCallbackHandler
          createCallback =
          new org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks.CreateCallbackHandler();

      zkClient.asyncCreate("/tmp/async", null, CreateMode.PERSISTENT, createCallback);
      createCallback.waitForSuccess();
      Assert.assertEquals(createCallback.getRc(), 0);
      Assert.assertTrue(zkClient.exists("/tmp/async"));

      Assert.assertFalse(zkClient.exists("/tmp/asyncOversize"));
      // try to create oversize node, should fail
      zkClient.asyncCreate("/tmp/asyncOversize", oversizeZNRecord, CreateMode.PERSISTENT,
          createCallback);
      createCallback.waitForSuccess();
      Assert.assertEquals(createCallback.getRc(), KeeperException.Code.MARSHALLINGERROR.intValue());
      Assert.assertFalse(zkClient.exists("/tmp/asyncOversize"));

      ZNRecord normalZNRecord = new ZNRecord("normal");
      normalZNRecord.setSimpleField("key", buffString);

      org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks.SetDataCallbackHandler
          setDataCallbackHandler =
          new org.apache.helix.zookeeper.zkclient.callback.ZkAsyncCallbacks.SetDataCallbackHandler();

      zkClient.asyncSetData("/tmp/async", normalZNRecord, -1, setDataCallbackHandler);
      setDataCallbackHandler.waitForSuccess();
      Assert.assertEquals(setDataCallbackHandler.getRc(), 0);

      zkClient.asyncSetData("/tmp/async", oversizeZNRecord, -1, setDataCallbackHandler);
      setDataCallbackHandler.waitForSuccess();
      Assert.assertEquals(setDataCallbackHandler.getRc(),
          KeeperException.Code.MARSHALLINGERROR.intValue());
      Assert.assertEquals(zkClient.readData("/tmp/async"), normalZNRecord);
    } finally {
      TestHelper
          .resetSystemProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES,
              originSizeLimit);
      zkClient.delete("/tmp/async");
      zkClient.delete("/tmp/asyncOversize");
      zkClient.close();
    }
  }

  @Test
  public void testAsyncWriteByExpectedSession() throws Exception {
    ZkClient zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
    zkClient.setZkSerializer(new ZNRecordSerializer());
    String sessionId = Long.toHexString(zkClient.getSessionId());
    String path = "/" + TestHelper.getTestClassName() + "_" + TestHelper.getTestMethodName();
    SessionAwareZNRecord record = new SessionAwareZNRecord("test");

    // Set a dummy session id string to be mismatched with the real session id in ZkClient.
    record.setExpectedSessionId("ExpectedSession");
    ZkAsyncCallbacks.CreateCallbackHandler createCallback =
        new ZkAsyncCallbacks.CreateCallbackHandler();

    try {
      zkClient.asyncCreate(path, record, CreateMode.PERSISTENT, createCallback);
      Assert.fail("Invalid session should not create znode");
    } catch (ZkSessionMismatchedException expected) {
      Assert.assertEquals(expected.getMessage(),
          "Failed to get expected zookeeper instance! There is a session id mismatch. Expected: "
              + "ExpectedSession. Actual: " + sessionId);

      // Ensure the async callback is cancelled because of the exception
      Assert.assertTrue(createCallback.waitForSuccess(), "Callback operation should be done");
      Assert.assertEquals(createCallback.getRc(), KeeperException.Code.APIERROR.intValue());
    }

    Assert.assertFalse(zkClient.exists(path));

    // A valid session should be able to create the znode.
    record.setExpectedSessionId(sessionId);
    createCallback = new ZkAsyncCallbacks.CreateCallbackHandler();
    zkClient.asyncCreate(path, record, CreateMode.PERSISTENT, createCallback);

    Assert.assertTrue(createCallback.waitForSuccess(), "Callback operation should be done");
    Assert.assertEquals(createCallback.getRc(), 0);
    Assert.assertTrue(zkClient.exists(path));

    // Test asyncSetData() failed by mismatched session
    record.setExpectedSessionId("ExpectedSession");
    ZkAsyncCallbacks.SetDataCallbackHandler setDataCallback =
        new ZkAsyncCallbacks.SetDataCallbackHandler();
    try {
      zkClient.asyncSetData(path, record, 0, setDataCallback);
      Assert.fail("Invalid session should not change znode data");
    } catch (ZkSessionMismatchedException expected) {
      Assert.assertEquals(expected.getMessage(),
          "Failed to get expected zookeeper instance! There is a session id mismatch. Expected: "
              + "ExpectedSession. Actual: " + sessionId);

      // Ensure the async callback is cancelled because of the exception
      Assert.assertTrue(setDataCallback.waitForSuccess(), "Callback operation should be done");
      Assert.assertEquals(setDataCallback.getRc(), KeeperException.Code.APIERROR.intValue());
    }

    TestHelper.verify(() -> zkClient.delete(path), TestHelper.WAIT_DURATION);
    zkClient.close();
  }

  /*
   * Tests getChildren() when there are an excessive number of children and connection loss happens,
   * the operation should terminate and exit retry loop.
   */
  @Test(timeOut = 30 * 1000L)
  public void testGetChildrenOnLargeNumChildren() throws Exception {
    final String methodName = TestHelper.getTestMethodName();
    System.out.println("Start test: " + methodName);
    // Create 110K children to make packet length of children exceed 4 MB
    // and cause connection loss for getChildren() operation
    String path = "/" + methodName;

    _zkClient.createPersistent(path);

    for (int i = 0; i < 110; i++) {
      List<Op> ops = new ArrayList<>(1000);
      for (int j = 0; j < 1000; j++) {
        String childPath = path + "/" + UUID.randomUUID().toString();
        // Create ephemeral nodes so closing zkClient deletes them for cleanup
        ops.add(
            Op.create(childPath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL));
      }
      // Reduce total creation time by batch creating znodes
      _zkClient.multi(ops);
    }

    try {
      _zkClient.getChildren(path);
      Assert.fail("Should not successfully get children because of connection loss.");
    } catch (ZkException expected) {
      Assert.assertEquals(expected.getMessage(),
          "org.apache.zookeeper.KeeperException$MarshallingErrorException: "
              + "KeeperErrorCode = MarshallingError");
    } finally {
      // Delete children ephemeral znodes
      _zkClient.close();
      _zkClient = new ZkClient(ZkTestBase.ZK_ADDR);

      Assert.assertTrue(TestHelper.verify(() -> {
        try {
          return _zkClient.delete(path);
        } catch (ZkException e) {
          return false;
        }
      }, TestHelper.WAIT_DURATION));
    }
    System.out.println("End test: " + methodName);
  }

  @Test(expectedExceptions = ZkClientException.class,
      expectedExceptionsMessageRegExp = "Data size of path .* is greater than write size limit 1024000 bytes")
  public void testDataSizeGreaterThanLimit() {
    // Creating should fail because size is greater than limit.
    _zkClient.createPersistent("/" + TestHelper.getTestMethodName(), new byte[1001 * 1024]);
  }

  @Test
  public void testDataSizeLessThanLimit() throws Exception {
    String path = "/" + TestHelper.getTestMethodName();
    Assert.assertFalse(_zkClient.exists(path));
    // Creating znode is successful.
    _zkClient.createPersistent(path, new byte[1024]);

    Assert.assertTrue(_zkClient.exists(path));

    TestHelper.verify(() -> _zkClient.delete(path), TestHelper.WAIT_DURATION);
  }

  // Tests znrecord serializer write size limit is invalid: greater than size limit in ZkClient
  @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp =
      "ZNRecord serializer write size limit .* is greater than ZkClient size limit .*")
  public void testInvalidWriteSizeLimitConfig() {
    String originSerializerLimit =
        System.getProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES);
    System.setProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES,
        String.valueOf(ZNRecord.SIZE_LIMIT + 1024));

    ZkClient zkClient = null;
    try {
      // Constructing ZkClient should throw exception because of invalid write size limit config
      zkClient = new ZkClient(ZkTestBase.ZK_ADDR);
    } finally {
      TestHelper
          .resetSystemProperty(ZkSystemPropertyKeys.ZK_SERIALIZER_ZNRECORD_WRITE_SIZE_LIMIT_BYTES,
              originSerializerLimit);
      if (zkClient != null) {
        zkClient.close();
      }
    }
  }
}