package com.linkedin.helix.integration;

import java.util.Date;

import org.testng.Assert;
import org.testng.annotations.Test;

import com.linkedin.helix.HelixDataAccessor;
import com.linkedin.helix.TestHelper;
import com.linkedin.helix.ZNRecord;
import com.linkedin.helix.manager.zk.ZKHelixDataAccessor;
import com.linkedin.helix.manager.zk.ZkBaseDataAccessor;
import com.linkedin.helix.mock.controller.StandaloneController;
import com.linkedin.helix.mock.storage.MockParticipant;
import com.linkedin.helix.model.PauseSignal;
import com.linkedin.helix.tools.ClusterSetup;
import com.linkedin.helix.tools.ClusterStateVerifier;
import com.linkedin.helix.tools.ClusterStateVerifier.BestPossAndExtViewZkVerifier;

public class TestPauseSignal extends ZkIntegrationTestBase
{
  @Test()
  public void testPauseSignal() throws Exception
  {
    // Logger.getRootLogger().setLevel(Level.INFO);
    String className = TestHelper.getTestClassName();
    String methodName = TestHelper.getTestMethodName();
    String clusterName = className + "_" + methodName;

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

    MockParticipant[] participants = new MockParticipant[5];

    TestHelper.setupCluster(clusterName, ZK_ADDR, 12918, // participant port
                            "localhost", // participant name prefix
                            "TestDB", // resource name prefix
                            1, // resources
                            10, // partitions per resource
                            5, // number of nodes
                            3, // replicas
                            "MasterSlave",
                            true); // do rebalance

    // start controller
    StandaloneController controller =
        new StandaloneController(clusterName, "controller_0", ZK_ADDR);
    controller.syncStart();

    // start participants
    for (int i = 0; i < 5; i++)
    {
      String instanceName = "localhost_" + (12918 + i);

      participants[i] = new MockParticipant(clusterName, instanceName, ZK_ADDR, null);
      participants[i].syncStart();
    }

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

    // pause the cluster and add a new resource group
    HelixDataAccessor accessor =
        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor<ZNRecord>(_gZkClient));
    accessor.setProperty(accessor.keyBuilder().pause(), new PauseSignal("pause"));
    
    // wait for controller to be signaled by pause
    Thread.sleep(1000);
    
    ClusterSetup setupTool = new ClusterSetup(ZK_ADDR);
    setupTool.addResourceToCluster(clusterName, "TestDB1", 10, "MasterSlave");
    setupTool.rebalanceStorageCluster(clusterName, "TestDB1", 3);

    // make sure TestDB1 external view is empty
    TestHelper.verifyWithTimeout("verifyEmptyCurStateAndExtView",
                                 1000,
                                 clusterName,
                                 "TestDB1",
                                 TestHelper.<String>setOf("localhost_12918", "localhost_12919",
                                                          "localhost_12920", "localhost_12921",
                                                          "localhost_12922"),
                                 ZK_ADDR);

    // resume controller
    accessor.removeProperty(accessor.keyBuilder().pause());
    result =
        ClusterStateVerifier.verifyByZkCallback(new BestPossAndExtViewZkVerifier(ZK_ADDR,
                                                                                clusterName));
    Assert.assertTrue(result);
    
    // clean up
    for (int i = 0; i < 5; i++)
    {
      participants[i].syncStop();
    }

    Thread.sleep(2000);
    controller.syncStop();

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

@@ -336,33 +336,137 @@
 ger.zk.Z
-kBaseDataAccessor
+NRecordSerializer;%0Aimport com.linkedin.helix.manager.zk.ZkBaseDataAccessor;%0Aimport com.linkedin.helix.manager.zk.ZkClient
 ;%0Aimport
@@ -1094,24 +1094,30 @@
 Name();%0A    
+final 
 String clust
@@ -2417,32 +2417,33 @@
                 
+ 
 clusterName));%0A 
@@ -2506,36 +2506,150 @@
 and 
-add a new resource group%0A   
+make sure pause is persistent%0A    ZkClient zkClient = new ZkClient(ZK_ADDR);%0A    zkClient.setZkSerializer(new ZNRecordSerializer());%0A    final
  Hel
@@ -2663,17 +2663,20 @@
 ccessor 
-a
+tmpA
 ccessor 
@@ -2759,11 +2759,9 @@
 rd%3E(
-_gZ
+z
 kCli
@@ -2767,25 +2767,28 @@
 ient));%0A    
-a
+tmpA
 ccessor.setP
@@ -2787,33 +2787,36 @@
 sor.setProperty(
-a
+tmpA
 ccessor.keyBuild
@@ -2852,24 +2852,46 @@
 (%22pause%22));%0A
+    zkClient.close();%0A
     %0A    // 
@@ -2958,20 +2958,48 @@
 (1000);%0A
+%0A
     
+// add a new resource group
 %0A    Clu
@@ -3488,16 +3488,17 @@
 %3CString%3E
+ 
 setOf(%22l
@@ -3513,16 +3513,75 @@
 _12918%22,
+%0A                                                          
  %22localh
@@ -3642,32 +3642,33 @@
                 
+ 
 %22localhost_12920
@@ -3669,16 +3669,75 @@
 _12920%22,
+%0A                                                          
  %22localh
@@ -3744,24 +3744,25 @@
 ost_12921%22,%0A
+ 
             
@@ -3892,24 +3892,156 @@
  controller%0A
+    final HelixDataAccessor accessor =%0A        new ZKHelixDataAccessor(clusterName, new ZkBaseDataAccessor%3CZNRecord%3E(_gZkClient));%0A%0A
     accessor
@@ -4263,32 +4263,33 @@
                 
+ 
 clusterName));%0A 
@@ -4318,20 +4318,16 @@
 esult);%0A
-    
 %0A    // 

refactor pause signal test case to make sure pause is persistent on zk