package org.apache.helix.tools;

/*
 * 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.io.FileWriter;
import java.io.PrintWriter;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

import org.apache.helix.BaseDataAccessor;
import org.apache.helix.HelixAdmin;
import org.apache.helix.HelixDataAccessor;
import org.apache.helix.TestHelper;
import org.apache.helix.common.ZkTestBase;
import org.apache.helix.integration.manager.ClusterDistributedController;
import org.apache.helix.integration.manager.MockParticipantManager;
import org.apache.helix.manager.zk.ZKHelixDataAccessor;
import org.apache.helix.manager.zk.ZKUtil;
import org.apache.helix.manager.zk.ZkBaseDataAccessor;
import org.apache.helix.model.IdealState;
import org.apache.helix.model.InstanceConfig;
import org.apache.helix.model.LiveInstance;
import org.apache.helix.store.ZNRecordJsonSerializer;
import org.apache.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;
import org.apache.helix.tools.ClusterStateVerifier.MasterNbInExtViewVerifier;
import org.apache.helix.tools.ClusterVerifiers.BestPossibleExternalViewVerifier;
import org.apache.helix.zookeeper.datamodel.ZNRecord;
import org.testng.Assert;
import org.testng.ITestContext;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.Test;

public class TestHelixAdminCli extends ZkTestBase {
  private static final long SLEEP_DURATION = 1000L;
  private String clusterName = TestHelper.getTestClassName();
  private String grandClusterName = clusterName + "_grand";

  @AfterMethod
  public void endTest(Method testMethod, ITestContext testContext) {
    try {
      deleteCluster(clusterName);
      deleteCluster(grandClusterName);
    } catch (Exception ignored) {
      // OK
    }
    super.endTest(testMethod, testContext);
  }

  @Test
  public void testAddCluster() throws Exception {
    String command = "--zkSvr localhost:2183 -addCluster clusterTest";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // malformed cluster name
    command = "--zkSvr localhost:2183 -addCluster /ClusterTest";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("ClusterSetup should fail since /ClusterTest is not a valid name");
    } catch (Exception e) {
      // OK
    }

    // Add the grand cluster
    // " is ignored by zk
    command = "--zkSvr localhost:2183 -addCluster \"Klazt3rz";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "--zkSvr localhost:2183 -addCluster \\ClusterTest";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // Add already exist cluster
    command = "--zkSvr localhost:2183 -addCluster clusterTest";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // make sure clusters are properly setup
    Assert.assertTrue(ZKUtil.isClusterSetup("Klazt3rz", _gZkClient));
    Assert.assertTrue(ZKUtil.isClusterSetup("clusterTest", _gZkClient));
    Assert.assertTrue(ZKUtil.isClusterSetup("\\ClusterTest", _gZkClient));

    // delete cluster without resource and instance
    command = "-zkSvr localhost:2183 -dropCluster \"Klazt3rz";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -dropCluster \\ClusterTest";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -dropCluster clusterTest1";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -dropCluster clusterTest";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    Assert.assertFalse(_gZkClient.exists("/Klazt3rz"));
    Assert.assertFalse(_gZkClient.exists("/clusterTest"));
    Assert.assertFalse(_gZkClient.exists("/\\ClusterTest"));
    Assert.assertFalse(_gZkClient.exists("/clusterTest1"));
  }

  @Test(dependsOnMethods = "testAddCluster")
  public void testAddResource() throws Exception {
    String command = "-zkSvr localhost:2183 -addCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_22 144 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_11 44 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // Add duplicate resource
    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_22 55 OnlineOffline";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("ClusterSetup should fail since resource db_22 already exists");
    } catch (Exception e) {
      // OK
    }

    // drop resource now
    command = "-zkSvr localhost:2183 -dropResource " + clusterName + " db_11 ";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));
  }

  @Test(dependsOnMethods = "testAddResource")
  public void testAddInstance() throws Exception {
    String command = "-zkSvr localhost:2183 -addCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    for (int i = 0; i < 3; i++) {
      command = "-zkSvr localhost:2183 -addNode " + clusterName + " localhost:123" + i;
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    command = "-zkSvr localhost:2183 -addNode " + clusterName
        + " localhost:1233;localhost:1234;localhost:1235;localhost:1236";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // delete enabled node
    command = "-zkSvr localhost:2183 -dropNode " + clusterName + " localhost:1236";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("delete node localhost:1236 should fail since it's not disabled");
    } catch (Exception e) {
      // OK
    }

    // delete non-exist node
    command = "-zkSvr localhost:2183 -dropNode " + clusterName + " localhost:12367";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("delete node localhost:1237 should fail since it doesn't exist");
    } catch (Exception e) {
      // OK
    }

    // disable node
    command = "-zkSvr localhost:2183 -enableInstance " + clusterName + " localhost:1236 false";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -dropNode " + clusterName + " localhost:1236";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // add a duplicated host
    command = "-zkSvr localhost:2183 -addNode " + clusterName + " localhost:1234";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("add node localhost:1234 should fail since it already exists");
    } catch (Exception e) {
      // OK
    }
  }

  @Test(dependsOnMethods = "testAddInstance")
  public void testRebalanceResource() throws Exception {
    String command = "-zkSvr localhost:2183 -addCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_11 12 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    for (int i = 0; i < 6; i++) {
      command = "-zkSvr localhost:2183 -addNode " + clusterName + " localhost:123" + i;
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    command = "-zkSvr localhost:2183 -rebalance " + clusterName + " db_11 3";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -dropResource " + clusterName + " db_11 ";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // re-add and rebalance
    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_11 48 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -rebalance " + clusterName + " db_11 3";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // rebalance with key prefix
    command = "-zkSvr localhost:2183 -rebalance " + clusterName + " db_11 2 -key alias";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));
  }

  @Test(dependsOnMethods = "testRebalanceResource")
  public void testStartCluster() throws Exception {
    final int n = 6;

    MockParticipantManager[] participants = new MockParticipantManager[n];
    ClusterDistributedController[] controllers = new ClusterDistributedController[2];
    setupCluster(clusterName, grandClusterName, n, participants, controllers);

    // activate clusters
    // wrong grand clusterName
    String command =
        "-zkSvr localhost:2183 -activateCluster " + clusterName + " nonExistGrandCluster true";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail(
          "add " + clusterName + " to grandCluster should fail since grandCluster doesn't exists");
    } catch (Exception e) {
      // OK
    }

    // wrong cluster name
    command =
        "-zkSvr localhost:2183 -activateCluster nonExistCluster " + grandClusterName + " true";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("add nonExistCluster to " + grandClusterName
          + " should fail since nonExistCluster doesn't exists");
    } catch (Exception e) {
      // OK
    }

    activateCluster();

    // drop a running cluster
    command = "-zkSvr localhost:2183 -dropCluster " + clusterName;
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("drop " + clusterName + " should fail since it's still running");
    } catch (Exception e) {
      // OK
    }

    // verify leader node
    BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    HelixDataAccessor accessor = new ZKHelixDataAccessor(grandClusterName, baseAccessor);
    LiveInstance controllerLeader = accessor.getProperty(accessor.keyBuilder().controllerLeader());
    Assert.assertNotNull(controllerLeader,
        "controllerLeader should be either controller_9000 or controller_9001");
    Assert.assertTrue(controllerLeader.getInstanceName().startsWith("controller_900"));

    accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
    LiveInstance leader = accessor.getProperty(accessor.keyBuilder().controllerLeader());
    for (int i = 0; i < 20; i++) {
      if (leader != null) {
        break;
      }
      Thread.sleep(200);
      leader = accessor.getProperty(accessor.keyBuilder().controllerLeader());
    }
    Assert.assertNotNull(leader);
    Assert.assertTrue(leader.getInstanceName().startsWith("controller_900"));

    boolean verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new MasterNbInExtViewVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    // clean up
    for (ClusterDistributedController controller : controllers) {
      controller.syncStop();
    }
    for (ClusterDistributedController controller : controllers) {
      if (controller.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
    for (MockParticipantManager participant : participants) {
      participant.syncStop();
    }
    for (MockParticipantManager participant : participants) {
      if (participant.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
  }

  @Test(dependsOnMethods = "testStartCluster")
  public void testDropAddResource() throws Exception {
    final int n = 6;

    MockParticipantManager[] participants = new MockParticipantManager[n];
    ClusterDistributedController[] controllers = new ClusterDistributedController[2];
    setupCluster(clusterName, grandClusterName, n, participants, controllers);
    activateCluster();

    // save ideal state
    BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
    IdealState idealState = accessor.getProperty(accessor.keyBuilder().idealStates("db_11"));
    ZNRecordJsonSerializer serializer = new ZNRecordJsonSerializer();

    String tmpDir = System.getProperty("java.io.tmpdir");
    if (tmpDir == null) {
      tmpDir = "/tmp";
    }
    final String tmpIdealStateFile = tmpDir + "/" + clusterName + "_idealState.log";
    FileWriter fos = new FileWriter(tmpIdealStateFile);
    PrintWriter pw = new PrintWriter(fos);
    pw.write(new String(serializer.serialize(idealState.getRecord())));
    pw.close();

    String command = "-zkSvr " + ZK_ADDR + " -dropResource " + clusterName + " db_11 ";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    boolean verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    command =
        "-zkSvr " + ZK_ADDR + " -addIdealState " + clusterName + " db_11 " + tmpIdealStateFile;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    IdealState idealState2 = accessor.getProperty(accessor.keyBuilder().idealStates("db_11"));
    Assert.assertEquals(idealState.getRecord(), idealState2.getRecord());

    // clean up
    for (ClusterDistributedController controller : controllers) {
      controller.syncStop();
    }
    for (MockParticipantManager participant : participants) {
      participant.syncStop();
    }
  }

  private void setupCluster(String clusterName, String grandClusterName, final int n,
      MockParticipantManager[] participants, ClusterDistributedController[] controllers)
      throws Exception {
    // add cluster
    String command = "-zkSvr " + ZK_ADDR + " -addCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // add grand cluster
    command = "-zkSvr " + ZK_ADDR + " -addCluster " + grandClusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // add nodes
    for (int i = 0; i < n; i++) {
      command = "-zkSvr " + ZK_ADDR + " -addNode " + clusterName + " localhost:123" + i;
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    // add resource
    command = "-zkSvr " + ZK_ADDR + " -addResource " + clusterName + " db_11 48 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // rebalance with key prefix
    command = "-zkSvr " + ZK_ADDR + " -rebalance " + clusterName + " db_11 2 -key alias";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // add nodes to grand cluster
    command =
        "-zkSvr " + ZK_ADDR + " -addNode " + grandClusterName + " controller:9000;controller:9001";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // start mock nodes
    for (int i = 0; i < n; i++) {
      String instanceName = "localhost_123" + i;
      participants[i] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
      participants[i].syncStart();
    }

    // start controller nodes
    for (int i = 0; i < 2; i++) {
      controllers[i] =
          new ClusterDistributedController(ZK_ADDR, grandClusterName, "controller_900" + i);
      controllers[i].syncStart();
    }

    Thread.sleep(SLEEP_DURATION);
  }

  @Test(dependsOnMethods = "testDropAddResource")
  public void testInstanceOperations() throws Exception {
    final int n = 6;

    MockParticipantManager[] participants = new MockParticipantManager[n];
    ClusterDistributedController[] controllers = new ClusterDistributedController[2];
    setupCluster(clusterName, grandClusterName, n, participants, controllers);
    activateCluster();

    // drop node should fail if the node is not disabled
    String command = "-zkSvr " + ZK_ADDR + " -dropNode " + clusterName + " localhost:1232";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("dropNode should fail since the node is not disabled");
    } catch (Exception e) {
      // OK
    }

    // disabled node
    command = "-zkSvr " + ZK_ADDR + " -enableInstance " + clusterName + " localhost:1232 false";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // Cannot dropNode if the node is not disconnected
    command = "-zkSvr " + ZK_ADDR + " -dropNode " + clusterName + " localhost:1232";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("dropNode should fail since the node is not disconnected");
    } catch (Exception e) {
      // OK
    }

    // Cannot swapNode if the node is not disconnected
    command =
        "-zkSvr " + ZK_ADDR + " -swapInstance " + clusterName + " localhost_1232 localhost_12320";
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("swapInstance should fail since the node is not disconnected");
    } catch (Exception e) {
      // OK
    }

    // disconnect localhost_1232
    participants[2].syncStop();

    // add new node then swap instance
    command = "-zkSvr " + ZK_ADDR + " -addNode " + clusterName + " localhost:12320";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // swap instance. The instance get swapped out should not exist anymore
    command =
        "-zkSvr " + ZK_ADDR + " -swapInstance " + clusterName + " localhost_1232 localhost_12320";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
    String path = accessor.keyBuilder().instanceConfig("localhost_1232").getPath();
    Assert.assertFalse(_gZkClient.exists(path),
        path + " should not exist since localhost_1232 has been swapped by localhost_12320");

    // clean up
    for (ClusterDistributedController controller : controllers) {
      controller.syncStop();
    }
    for (ClusterDistributedController controller : controllers) {
      if (controller.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
    for (MockParticipantManager participant : participants) {
      participant.syncStop();
    }
    for (MockParticipantManager participant : participants) {
      if (participant.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
  }

  @Test(dependsOnMethods = "testInstanceOperations")
  public void testExpandCluster() throws Exception {
    final int n = 6;

    System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));

    MockParticipantManager[] participants = new MockParticipantManager[n];
    ClusterDistributedController[] controllers = new ClusterDistributedController[2];
    setupCluster(clusterName, grandClusterName, n, participants, controllers);
    activateCluster();

    String command = "-zkSvr " + ZK_ADDR + " -addNode " + clusterName
        + " localhost:12331;localhost:12341;localhost:12351;localhost:12361";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -expandCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    MockParticipantManager[] newParticipants = new MockParticipantManager[4];
    for (int i = 3; i <= 6; i++) {
      String instanceName = "localhost_123" + i + "1";
      newParticipants[i - 3] = new MockParticipantManager(ZK_ADDR, clusterName, instanceName);
      newParticipants[i - 3].syncStart();
    }

    boolean verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new MasterNbInExtViewVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    verifyResult = ClusterStateVerifier
        .verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR, clusterName));
    Assert.assertTrue(verifyResult);

    // clean up
    for (ClusterDistributedController controller : controllers) {
      controller.syncStop();
    }
    for (ClusterDistributedController controller : controllers) {
      if (controller.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
    for (MockParticipantManager participant : participants) {
      participant.syncStop();
    }
    for (MockParticipantManager participant : participants) {
      if (participant.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
    for (MockParticipantManager newParticipant : newParticipants) {
      newParticipant.syncStop();
    }
    for (MockParticipantManager newParticipant : newParticipants) {
      if (newParticipant.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }
  }

  @Test(dependsOnMethods = "testExpandCluster")
  public void testDeactivateCluster() throws Exception {
    final int n = 6;

    System.out.println("START " + clusterName + " at " + new Date(System.currentTimeMillis()));

    MockParticipantManager[] participants = new MockParticipantManager[n];
    ClusterDistributedController[] controllers = new ClusterDistributedController[2];
    setupCluster(clusterName, grandClusterName, n, participants, controllers);
    activateCluster();
    // wait till grand_cluster converge
    BestPossibleExternalViewVerifier verifier =
        new BestPossibleExternalViewVerifier.Builder(grandClusterName).setZkClient(_gZkClient)
            .setWaitTillVerify(TestHelper.DEFAULT_REBALANCE_PROCESSING_WAIT_TIME).build();

    boolean result = verifier.verifyByPolling();
    Assert.assertTrue(result, "grand cluster not converging.");

    // deactivate cluster
    String command =
        "-zkSvr " + ZK_ADDR + " -activateCluster " + clusterName + " " + grandClusterName
            + " false";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);
    final String path = accessor.keyBuilder().controllerLeader().getPath();
    TestHelper.verify(() -> !_gZkClient.exists(path), 10000L);

    Assert.assertFalse(_gZkClient.exists(path),
        "leader should be gone after deactivate the cluster");

    command = "-zkSvr " + ZK_ADDR + " -dropCluster " + clusterName;
    try {
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
      Assert.fail("dropCluster should fail since there are still instances running");
    } catch (Exception e) {
      // OK
    }

    for (MockParticipantManager participant : participants) {
      participant.syncStop();
    }
    for (MockParticipantManager participant : participants) {
      if (participant.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }

    command = "-zkSvr localhost:2183 -dropCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s"));

    boolean leaderNotExists = TestHelper.verify(() -> {
      boolean isLeaderExists = _gZkClient.exists(path);
      return isLeaderExists == false;
    }, TestHelper.WAIT_DURATION);
    Assert.assertTrue(leaderNotExists, " mysterious leader out!");

    for (ClusterDistributedController controller : controllers) {
      controller.syncStop();
    }
    for (ClusterDistributedController controller : controllers) {
      if (controller.isConnected()) {
        Thread.sleep(SLEEP_DURATION);
      }
    }

    command = "-zkSvr localhost:2183 -dropCluster " + grandClusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));
  }

  @Test(dependsOnMethods = "testDeactivateCluster")
  public void testInstanceGroupTags() throws Exception {
    BaseDataAccessor<ZNRecord> baseAccessor = new ZkBaseDataAccessor<>(_gZkClient);
    HelixDataAccessor accessor = new ZKHelixDataAccessor(clusterName, baseAccessor);

    String command = "-zkSvr " + ZK_ADDR + " -addCluster " + clusterName;
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command = "-zkSvr localhost:2183 -addResource " + clusterName + " db_11 12 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    for (int i = 0; i < 6; i++) {
      command = "-zkSvr " + ZK_ADDR + " -addNode " + clusterName + " localhost:123" + i;
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    for (int i = 0; i < 2; i++) {
      command =
          "-zkSvr " + ZK_ADDR + " -addInstanceTag " + clusterName + " localhost_123" + i + "  tag1";
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }
    for (int i = 2; i < 6; i++) {
      command =
          "-zkSvr " + ZK_ADDR + " -addInstanceTag " + clusterName + " localhost_123" + i + "  tag2";
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    command =
        "-zkSvr " + ZK_ADDR + " -rebalance " + clusterName + " db_11 2 -instanceGroupTag tag1";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    IdealState dbIs = accessor.getProperty(accessor.keyBuilder().idealStates("db_11"));
    Set<String> hosts = new HashSet<>();
    for (String p : dbIs.getPartitionSet()) {
      for (String hostName : dbIs.getInstanceStateMap(p).keySet()) {
        InstanceConfig config =
            accessor.getProperty(accessor.keyBuilder().instanceConfig(hostName));
        Assert.assertTrue(config.containsTag("tag1"));
        hosts.add(hostName);
      }
    }
    Assert.assertEquals(hosts.size(), 2);

    command = "-zkSvr " + ZK_ADDR + " -dropResource " + clusterName + " db_11 ";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    // re-add and rebalance
    command = "-zkSvr " + ZK_ADDR + " -addResource " + clusterName + " db_11 48 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command =
        "-zkSvr " + ZK_ADDR + " -rebalance " + clusterName + " db_11 3 -instanceGroupTag tag2";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    dbIs = accessor.getProperty(accessor.keyBuilder().idealStates("db_11"));
    hosts = new HashSet<>();
    for (String p : dbIs.getPartitionSet()) {
      for (String hostName : dbIs.getInstanceStateMap(p).keySet()) {
        InstanceConfig config =
            accessor.getProperty(accessor.keyBuilder().instanceConfig(hostName));
        Assert.assertTrue(config.containsTag("tag2"));
        hosts.add(hostName);
      }
    }
    Assert.assertEquals(hosts.size(), 4);

    command = "-zkSvr " + ZK_ADDR + " -dropResource " + clusterName + " db_11 ";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    for (int i = 3; i <= 3; i++) {
      command = "-zkSvr " + ZK_ADDR + " -removeInstanceTag " + clusterName + " localhost_123" + i
          + " tag2";
      ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    }

    // re-add and rebalance
    command = "-zkSvr " + ZK_ADDR + " -addResource " + clusterName + " db_11 48 MasterSlave";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    command =
        "-zkSvr " + ZK_ADDR + " -rebalance " + clusterName + " db_11 3 -instanceGroupTag tag2";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));
    dbIs = accessor.getProperty(accessor.keyBuilder().idealStates("db_11"));
    hosts = new HashSet<>();
    for (String p : dbIs.getPartitionSet()) {
      for (String hostName : dbIs.getInstanceStateMap(p).keySet()) {
        InstanceConfig config =
            accessor.getProperty(accessor.keyBuilder().instanceConfig(hostName));
        Assert.assertTrue(config.containsTag("tag2"));
        hosts.add(hostName);
      }
    }
    Assert.assertEquals(hosts.size(), 3);

    // rebalance with key prefix
    command = "-zkSvr " + ZK_ADDR + " -rebalance " + clusterName + " db_11 2 -key alias";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));
  }

  private void activateCluster() throws Exception {
    String command =
        "-zkSvr localhost:2183 -activateCluster " + clusterName + " " + grandClusterName + " true";
    ClusterSetup.processCommandLineArgs(command.split("\\s+"));

    HelixAdmin admin = _gSetupTool.getClusterManagementTool();
    TestHelper.verify(() -> {
      if (admin.getResourceExternalView(grandClusterName, clusterName) == null) {
        return false;
      }
      return true;
    }, TestHelper.WAIT_DURATION);
  }
}
