helix-core/src/test/java/org/apache/helix/store/zk/TestZkHelixPropertyStore.java - helix - Git at Google

 package org.apache.helix.store.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.lang.management.ManagementFactory;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import javax.management.JMException;
 import javax.management.MBeanServer;
 import javax.management.ObjectName;

 import org.apache.helix.AccessOption;
 import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.helix.ZkUnitTestBase;
 import org.apache.helix.manager.zk.ZkBaseDataAccessor;
 import org.apache.helix.zookeeper.api.client.HelixZkClient;
 import org.apache.helix.monitoring.mbeans.MBeanRegistrar;
 import org.apache.helix.monitoring.mbeans.MonitorDomainNames;
 import org.apache.helix.monitoring.mbeans.ZkClientMonitor;
 import org.apache.helix.monitoring.mbeans.ZkClientPathMonitor;
 import org.apache.helix.store.HelixPropertyListener;
 import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
 import org.apache.helix.zookeeper.zkclient.serialize.SerializableSerializer;
 import org.testng.Assert;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.Test;

 public class TestZkHelixPropertyStore extends ZkUnitTestBase {
   final String _root = "/" + getShortClassName();
   final int bufSize = 128;
   final int mapNr = 10;
   final int firstLevelNr = 10;
   final int secondLevelNr = 10;

   // final int totalNodes = firstLevelNr * secondLevelNr;

   class TestListener implements HelixPropertyListener {
     Map<String, Long> _changeKeys = new HashMap<>();
     Map<String, Long> _createKeys = new HashMap<>();
     Map<String, Long> _deleteKeys = new HashMap<>();

     public void reset() {
       _changeKeys.clear();
       _createKeys.clear();
       _deleteKeys.clear();
     }

     @Override
     public void onDataChange(String path) {
       _changeKeys.put(path, System.currentTimeMillis());
     }

     @Override
     public void onDataCreate(String path) {
       _createKeys.put(path, System.currentTimeMillis());
     }

     @Override
     public void onDataDelete(String path) {
       _deleteKeys.put(path, System.currentTimeMillis());
     }
   }

   @AfterClass
   public void afterClass() {
     deleteCluster(getShortClassName());
   }

   @Test
   public void testSet() {
     // Logger.getRootLogger().setLevel(Level.INFO);

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

     String subRoot = _root + "/" + "set";
     List<String> subscribedPaths = new ArrayList<>();
     subscribedPaths.add(subRoot);
     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

     // test set
     setNodes(store, 'a', false);
     for (int i = 0; i < 10; i++) {
       for (int j = 0; j < 10; j++) {
         String nodeId = getNodeId(i, j);
         String key = getSecondLevelKey(i, j);
         ZNRecord record = store.get(key, null, 0);
         Assert.assertEquals(record.getId(), nodeId);
       }
     }

     // test get from cache
     long startT = System.currentTimeMillis();
     for (int i = 0; i < 1000; i++) {
       ZNRecord record = store.get("/node_0/childNode_0_0", null, 0);
       Assert.assertNotNull(record);
     }
     long endT = System.currentTimeMillis();
     System.out.println("1000 Get() time used: " + (endT - startT) + "ms");
     long latency = endT - startT;
     Assert.assertTrue(latency < 200,
         "1000 Gets should be finished within 200ms, but was " + latency + " ms");

     store.stop();
     System.out.println("END testSet() at " + new Date(System.currentTimeMillis()));
   }

   @Test
   public void testSetInvalidPath() {
     String subRoot = _root + "/" + "setInvalidPath";
     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, null);
     try {
       store.set("abc/xyz", new ZNRecord("testInvalid"), AccessOption.PERSISTENT);
       Assert.fail("Should throw illegal-argument-exception since path doesn't start with /");
     } catch (IllegalArgumentException e) {
       // e.printStackTrace();
       // OK
     } catch (Exception e) {
       Assert.fail("Should not throw exceptions other than illegal-argument");
     }
   }

   @Test
   public void testLocalTriggeredCallback() throws Exception {
     System.out
         .println("START testLocalTriggeredCallback() at " + new Date(System.currentTimeMillis()));

     String subRoot = _root + "/" + "localCallback";
     List<String> subscribedPaths = new ArrayList<>();
     subscribedPaths.add(subRoot);
     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

     // change nodes via property store interface
     // and verify all notifications have been received
     TestListener listener = new TestListener();
     store.subscribe("/", listener);

     // test dataCreate callbacks
     listener.reset();
     setNodes(store, 'a', true);

     // wait until all callbacks have been received
     Thread.sleep(500);
     int expectCreateNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertEquals(expectCreateNodes, listener._createKeys.size(),
         "Should receive " + expectCreateNodes + " create callbacks");

     // test dataChange callbacks
     listener.reset();
     setNodes(store, 'b', true);

     // wait until all callbacks have been received
     Thread.sleep(500);
     int expectChangeNodes = firstLevelNr * secondLevelNr;
     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertTrue(listener._changeKeys.size() >= expectChangeNodes,
         "Should receive at least " + expectChangeNodes + " change callbacks");

     // test delete callbacks
     listener.reset();
     int expectDeleteNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
     store.remove("/", 0);
     // wait until all callbacks have been received
     for (int i = 0; i < 10; i++) {
       if (listener._deleteKeys.size() == expectDeleteNodes)
         break;
       Thread.sleep(500);
     }

     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertEquals(expectDeleteNodes, listener._deleteKeys.size(),
         "Should receive " + expectDeleteNodes + " delete callbacks");

     store.stop();
     System.out
         .println("END testLocalTriggeredCallback() at " + new Date(System.currentTimeMillis()));
   }

   @Test
   public void testZkTriggeredCallback() throws Exception {
     System.out
         .println("START testZkTriggeredCallback() at " + new Date(System.currentTimeMillis()));

     String subRoot = _root + "/" + "zkCallback";
     List<String> subscribedPaths = Collections.singletonList(subRoot);
     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

     // change nodes via property store interface
     // and verify all notifications have been received
     TestListener listener = new TestListener();
     store.subscribe("/", listener);

     // test create callbacks
     listener.reset();
     setNodes(_gZkClient, subRoot, 'a', true);
     int expectCreateNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
     Thread.sleep(500);

     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertEquals(expectCreateNodes, listener._createKeys.size(),
         "Should receive " + expectCreateNodes + " create callbacks");

     // test change callbacks
     listener.reset();
     setNodes(_gZkClient, subRoot, 'b', true);
     int expectChangeNodes = firstLevelNr * secondLevelNr;
     for (int i = 0; i < 10; i++) {
       if (listener._changeKeys.size() >= expectChangeNodes)
         break;
       Thread.sleep(500);
     }

     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertTrue(listener._changeKeys.size() >= expectChangeNodes,
         "Should receive at least " + expectChangeNodes + " change callbacks");

     // test delete callbacks
     listener.reset();
     int expectDeleteNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
     _gZkClient.deleteRecursively(subRoot);
     Thread.sleep(1000);

     System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
         + listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
     Assert.assertEquals(expectDeleteNodes, listener._deleteKeys.size(),
         "Should receive " + expectDeleteNodes + " delete callbacks");

     store.stop();
     System.out.println("END testZkTriggeredCallback() at " + new Date(System.currentTimeMillis()));
   }

   @Test
   public void testBackToBackRemoveAndSet() throws Exception {
     System.out
         .println("START testBackToBackRemoveAndSet() at " + new Date(System.currentTimeMillis()));

     String subRoot = _root + "/" + "backToBackRemoveAndSet";
     List<String> subscribedPaths = new ArrayList<>();
     subscribedPaths.add(subRoot);
     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

     store.set("/child0", new ZNRecord("child0"), AccessOption.PERSISTENT);

     ZNRecord record = store.get("/child0", null, 0); // will put the record in cache
     Assert.assertEquals(record.getId(), "child0");
     // System.out.println("1:get:" + record);

     String child0Path = subRoot + "/child0";
     for (int i = 0; i < 2; i++) {
       _gZkClient.delete(child0Path);
       _gZkClient.createPersistent(child0Path, new ZNRecord("child0-new-" + i));
     }

     Thread.sleep(500); // should wait for zk callback to add "/child0" into cache
     record = store.get("/child0", null, 0);
     Assert.assertEquals(record.getId(), "child0-new-1",
         "Cache shoulde be updated to latest create");
     // System.out.println("2:get:" + record);

     _gZkClient.delete(child0Path);
     Thread.sleep(500); // should wait for zk callback to remove "/child0" from cache
     try {
       record = store.get("/child0", null, AccessOption.THROW_EXCEPTION_IFNOTEXIST);
       Assert.fail("/child0 should have been removed");
     } catch (ZkNoNodeException e) {
       // OK.
     }
     // System.out.println("3:get:" + record);

     store.stop();
     System.out
         .println("END testBackToBackRemoveAndSet() at " + new Date(System.currentTimeMillis()));
   }

   private String getNodeId(int i, int j) {
     return "childNode_" + i + "_" + j;
   }

   private String getSecondLevelKey(int i, int j) {
     return "/node_" + i + "/" + getNodeId(i, j);
   }

   private String getFirstLevelKey(int i) {
     return "/node_" + i;
   }

   private void setNodes(ZkHelixPropertyStore<ZNRecord> store, char c, boolean needTimestamp) {
     char[] data = new char[bufSize];

     for (int i = 0; i < bufSize; i++) {
       data[i] = c;
     }

     Map<String, String> map = new TreeMap<>();
     for (int i = 0; i < mapNr; i++) {
       map.put("key_" + i, new String(data));
     }

     for (int i = 0; i < firstLevelNr; i++) {
       for (int j = 0; j < secondLevelNr; j++) {
         String nodeId = getNodeId(i, j);
         ZNRecord record = new ZNRecord(nodeId);
         record.setSimpleFields(map);
         if (needTimestamp) {
           long now = System.currentTimeMillis();
           record.setSimpleField("SetTimestamp", Long.toString(now));
         }
         String key = getSecondLevelKey(i, j);
         store.set(key, record, AccessOption.PERSISTENT);
       }
     }
   }

   private void setNodes(HelixZkClient zkClient, String root, char c, boolean needTimestamp) {
     char[] data = new char[bufSize];

     for (int i = 0; i < bufSize; i++) {
       data[i] = c;
     }

     Map<String, String> map = new TreeMap<>();
     for (int i = 0; i < mapNr; i++) {
       map.put("key_" + i, new String(data));
     }

     for (int i = 0; i < firstLevelNr; i++) {
       String firstLevelKey = getFirstLevelKey(i);

       for (int j = 0; j < secondLevelNr; j++) {
         String nodeId = getNodeId(i, j);
         ZNRecord record = new ZNRecord(nodeId);
         record.setSimpleFields(map);
         if (needTimestamp) {
           long now = System.currentTimeMillis();
           record.setSimpleField("SetTimestamp", Long.toString(now));
         }
         String key = getSecondLevelKey(i, j);
         try {
           zkClient.writeData(root + key, record);
         } catch (ZkNoNodeException e) {
           zkClient.createPersistent(root + firstLevelKey, true);
           zkClient.createPersistent(root + key, record);
         }
       }
     }
   }

   @Test
   public void testZkClientMonitor() throws JMException {
     final String TEST_ROOT = "/test_root";

     ZkHelixPropertyStore<ZNRecord> store =
         new ZkHelixPropertyStore<>(ZK_ADDR, new SerializableSerializer(), TEST_ROOT);

     ObjectName name = MBeanRegistrar.buildObjectName(MonitorDomainNames.HelixZkClient.name(),
         ZkClientMonitor.MONITOR_TYPE, ZkHelixPropertyStore.MONITOR_TYPE,
         ZkClientMonitor.MONITOR_KEY, TEST_ROOT, ZkClientPathMonitor.MONITOR_PATH, "Root");
     MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();
     Assert.assertTrue(beanServer.isRegistered(name));

     store.getStat("/", AccessOption.PERSISTENT);

     Assert.assertEquals((long) beanServer.getAttribute(name, "ReadCounter"), 1);
   }
 }
	package org.apache.helix.store.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.lang.management.ManagementFactory;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Date;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.TreeMap;
	import javax.management.JMException;
	import javax.management.MBeanServer;
	import javax.management.ObjectName;

	import org.apache.helix.AccessOption;
	import org.apache.helix.zookeeper.datamodel.ZNRecord;
	import org.apache.helix.ZkUnitTestBase;
	import org.apache.helix.manager.zk.ZkBaseDataAccessor;
	import org.apache.helix.zookeeper.api.client.HelixZkClient;
	import org.apache.helix.monitoring.mbeans.MBeanRegistrar;
	import org.apache.helix.monitoring.mbeans.MonitorDomainNames;
	import org.apache.helix.monitoring.mbeans.ZkClientMonitor;
	import org.apache.helix.monitoring.mbeans.ZkClientPathMonitor;
	import org.apache.helix.store.HelixPropertyListener;
	import org.apache.helix.zookeeper.zkclient.exception.ZkNoNodeException;
	import org.apache.helix.zookeeper.zkclient.serialize.SerializableSerializer;
	import org.testng.Assert;
	import org.testng.annotations.AfterClass;
	import org.testng.annotations.Test;

	public class TestZkHelixPropertyStore extends ZkUnitTestBase {
	final String _root = "/" + getShortClassName();
	final int bufSize = 128;
	final int mapNr = 10;
	final int firstLevelNr = 10;
	final int secondLevelNr = 10;

	// final int totalNodes = firstLevelNr * secondLevelNr;

	class TestListener implements HelixPropertyListener {
	Map<String, Long> _changeKeys = new HashMap<>();
	Map<String, Long> _createKeys = new HashMap<>();
	Map<String, Long> _deleteKeys = new HashMap<>();

	public void reset() {
	_changeKeys.clear();
	_createKeys.clear();
	_deleteKeys.clear();
	}

	@Override
	public void onDataChange(String path) {
	_changeKeys.put(path, System.currentTimeMillis());
	}

	@Override
	public void onDataCreate(String path) {
	_createKeys.put(path, System.currentTimeMillis());
	}

	@Override
	public void onDataDelete(String path) {
	_deleteKeys.put(path, System.currentTimeMillis());
	}
	}

	@AfterClass
	public void afterClass() {
	deleteCluster(getShortClassName());
	}

	@Test
	public void testSet() {
	// Logger.getRootLogger().setLevel(Level.INFO);

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

	String subRoot = _root + "/" + "set";
	List<String> subscribedPaths = new ArrayList<>();
	subscribedPaths.add(subRoot);
	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

	// test set
	setNodes(store, 'a', false);
	for (int i = 0; i < 10; i++) {
	for (int j = 0; j < 10; j++) {
	String nodeId = getNodeId(i, j);
	String key = getSecondLevelKey(i, j);
	ZNRecord record = store.get(key, null, 0);
	Assert.assertEquals(record.getId(), nodeId);
	}
	}

	// test get from cache
	long startT = System.currentTimeMillis();
	for (int i = 0; i < 1000; i++) {
	ZNRecord record = store.get("/node_0/childNode_0_0", null, 0);
	Assert.assertNotNull(record);
	}
	long endT = System.currentTimeMillis();
	System.out.println("1000 Get() time used: " + (endT - startT) + "ms");
	long latency = endT - startT;
	Assert.assertTrue(latency < 200,
	"1000 Gets should be finished within 200ms, but was " + latency + " ms");

	store.stop();
	System.out.println("END testSet() at " + new Date(System.currentTimeMillis()));
	}

	@Test
	public void testSetInvalidPath() {
	String subRoot = _root + "/" + "setInvalidPath";
	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, null);
	try {
	store.set("abc/xyz", new ZNRecord("testInvalid"), AccessOption.PERSISTENT);
	Assert.fail("Should throw illegal-argument-exception since path doesn't start with /");
	} catch (IllegalArgumentException e) {
	// e.printStackTrace();
	// OK
	} catch (Exception e) {
	Assert.fail("Should not throw exceptions other than illegal-argument");
	}
	}

	@Test
	public void testLocalTriggeredCallback() throws Exception {
	System.out
	.println("START testLocalTriggeredCallback() at " + new Date(System.currentTimeMillis()));

	String subRoot = _root + "/" + "localCallback";
	List<String> subscribedPaths = new ArrayList<>();
	subscribedPaths.add(subRoot);
	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

	// change nodes via property store interface
	// and verify all notifications have been received
	TestListener listener = new TestListener();
	store.subscribe("/", listener);

	// test dataCreate callbacks
	listener.reset();
	setNodes(store, 'a', true);

	// wait until all callbacks have been received
	Thread.sleep(500);
	int expectCreateNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertEquals(expectCreateNodes, listener._createKeys.size(),
	"Should receive " + expectCreateNodes + " create callbacks");

	// test dataChange callbacks
	listener.reset();
	setNodes(store, 'b', true);

	// wait until all callbacks have been received
	Thread.sleep(500);
	int expectChangeNodes = firstLevelNr * secondLevelNr;
	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertTrue(listener._changeKeys.size() >= expectChangeNodes,
	"Should receive at least " + expectChangeNodes + " change callbacks");

	// test delete callbacks
	listener.reset();
	int expectDeleteNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
	store.remove("/", 0);
	// wait until all callbacks have been received
	for (int i = 0; i < 10; i++) {
	if (listener._deleteKeys.size() == expectDeleteNodes)
	break;
	Thread.sleep(500);
	}

	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertEquals(expectDeleteNodes, listener._deleteKeys.size(),
	"Should receive " + expectDeleteNodes + " delete callbacks");

	store.stop();
	System.out
	.println("END testLocalTriggeredCallback() at " + new Date(System.currentTimeMillis()));
	}

	@Test
	public void testZkTriggeredCallback() throws Exception {
	System.out
	.println("START testZkTriggeredCallback() at " + new Date(System.currentTimeMillis()));

	String subRoot = _root + "/" + "zkCallback";
	List<String> subscribedPaths = Collections.singletonList(subRoot);
	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

	// change nodes via property store interface
	// and verify all notifications have been received
	TestListener listener = new TestListener();
	store.subscribe("/", listener);

	// test create callbacks
	listener.reset();
	setNodes(_gZkClient, subRoot, 'a', true);
	int expectCreateNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
	Thread.sleep(500);

	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertEquals(expectCreateNodes, listener._createKeys.size(),
	"Should receive " + expectCreateNodes + " create callbacks");

	// test change callbacks
	listener.reset();
	setNodes(_gZkClient, subRoot, 'b', true);
	int expectChangeNodes = firstLevelNr * secondLevelNr;
	for (int i = 0; i < 10; i++) {
	if (listener._changeKeys.size() >= expectChangeNodes)
	break;
	Thread.sleep(500);
	}

	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertTrue(listener._changeKeys.size() >= expectChangeNodes,
	"Should receive at least " + expectChangeNodes + " change callbacks");

	// test delete callbacks
	listener.reset();
	int expectDeleteNodes = 1 + firstLevelNr + firstLevelNr * secondLevelNr;
	_gZkClient.deleteRecursively(subRoot);
	Thread.sleep(1000);

	System.out.println("createKey#:" + listener._createKeys.size() + ", changeKey#:"
	+ listener._changeKeys.size() + ", deleteKey#:" + listener._deleteKeys.size());
	Assert.assertEquals(expectDeleteNodes, listener._deleteKeys.size(),
	"Should receive " + expectDeleteNodes + " delete callbacks");

	store.stop();
	System.out.println("END testZkTriggeredCallback() at " + new Date(System.currentTimeMillis()));
	}

	@Test
	public void testBackToBackRemoveAndSet() throws Exception {
	System.out
	.println("START testBackToBackRemoveAndSet() at " + new Date(System.currentTimeMillis()));

	String subRoot = _root + "/" + "backToBackRemoveAndSet";
	List<String> subscribedPaths = new ArrayList<>();
	subscribedPaths.add(subRoot);
	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(new ZkBaseDataAccessor<>(_gZkClient), subRoot, subscribedPaths);

	store.set("/child0", new ZNRecord("child0"), AccessOption.PERSISTENT);

	ZNRecord record = store.get("/child0", null, 0); // will put the record in cache
	Assert.assertEquals(record.getId(), "child0");
	// System.out.println("1:get:" + record);

	String child0Path = subRoot + "/child0";
	for (int i = 0; i < 2; i++) {
	_gZkClient.delete(child0Path);
	_gZkClient.createPersistent(child0Path, new ZNRecord("child0-new-" + i));
	}

	Thread.sleep(500); // should wait for zk callback to add "/child0" into cache
	record = store.get("/child0", null, 0);
	Assert.assertEquals(record.getId(), "child0-new-1",
	"Cache shoulde be updated to latest create");
	// System.out.println("2:get:" + record);

	_gZkClient.delete(child0Path);
	Thread.sleep(500); // should wait for zk callback to remove "/child0" from cache
	try {
	record = store.get("/child0", null, AccessOption.THROW_EXCEPTION_IFNOTEXIST);
	Assert.fail("/child0 should have been removed");
	} catch (ZkNoNodeException e) {
	// OK.
	}
	// System.out.println("3:get:" + record);

	store.stop();
	System.out
	.println("END testBackToBackRemoveAndSet() at " + new Date(System.currentTimeMillis()));
	}

	private String getNodeId(int i, int j) {
	return "childNode_" + i + "_" + j;
	}

	private String getSecondLevelKey(int i, int j) {
	return "/node_" + i + "/" + getNodeId(i, j);
	}

	private String getFirstLevelKey(int i) {
	return "/node_" + i;
	}

	private void setNodes(ZkHelixPropertyStore<ZNRecord> store, char c, boolean needTimestamp) {
	char[] data = new char[bufSize];

	for (int i = 0; i < bufSize; i++) {
	data[i] = c;
	}

	Map<String, String> map = new TreeMap<>();
	for (int i = 0; i < mapNr; i++) {
	map.put("key_" + i, new String(data));
	}

	for (int i = 0; i < firstLevelNr; i++) {
	for (int j = 0; j < secondLevelNr; j++) {
	String nodeId = getNodeId(i, j);
	ZNRecord record = new ZNRecord(nodeId);
	record.setSimpleFields(map);
	if (needTimestamp) {
	long now = System.currentTimeMillis();
	record.setSimpleField("SetTimestamp", Long.toString(now));
	}
	String key = getSecondLevelKey(i, j);
	store.set(key, record, AccessOption.PERSISTENT);
	}
	}
	}

	private void setNodes(HelixZkClient zkClient, String root, char c, boolean needTimestamp) {
	char[] data = new char[bufSize];

	for (int i = 0; i < bufSize; i++) {
	data[i] = c;
	}

	Map<String, String> map = new TreeMap<>();
	for (int i = 0; i < mapNr; i++) {
	map.put("key_" + i, new String(data));
	}

	for (int i = 0; i < firstLevelNr; i++) {
	String firstLevelKey = getFirstLevelKey(i);

	for (int j = 0; j < secondLevelNr; j++) {
	String nodeId = getNodeId(i, j);
	ZNRecord record = new ZNRecord(nodeId);
	record.setSimpleFields(map);
	if (needTimestamp) {
	long now = System.currentTimeMillis();
	record.setSimpleField("SetTimestamp", Long.toString(now));
	}
	String key = getSecondLevelKey(i, j);
	try {
	zkClient.writeData(root + key, record);
	} catch (ZkNoNodeException e) {
	zkClient.createPersistent(root + firstLevelKey, true);
	zkClient.createPersistent(root + key, record);
	}
	}
	}
	}

	@Test
	public void testZkClientMonitor() throws JMException {
	final String TEST_ROOT = "/test_root";

	ZkHelixPropertyStore<ZNRecord> store =
	new ZkHelixPropertyStore<>(ZK_ADDR, new SerializableSerializer(), TEST_ROOT);

	ObjectName name = MBeanRegistrar.buildObjectName(MonitorDomainNames.HelixZkClient.name(),
	ZkClientMonitor.MONITOR_TYPE, ZkHelixPropertyStore.MONITOR_TYPE,
	ZkClientMonitor.MONITOR_KEY, TEST_ROOT, ZkClientPathMonitor.MONITOR_PATH, "Root");
	MBeanServer beanServer = ManagementFactory.getPlatformMBeanServer();
	Assert.assertTrue(beanServer.isRegistered(name));

	store.getStat("/", AccessOption.PERSISTENT);

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