hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestSplitLogWorker.java - hbase - Git at Google

 /**
  * 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.
  */
 package org.apache.hadoop.hbase.regionserver;

 import static org.apache.hadoop.hbase.HConstants.HBASE_SPLIT_WAL_MAX_SPLITTER;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;

 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.atomic.LongAdder;

 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.CoordinatedStateManager;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.SplitLogCounters;
 import org.apache.hadoop.hbase.SplitLogTask;
 import org.apache.hadoop.hbase.Waiter;
 import org.apache.hadoop.hbase.coordination.ZkCoordinatedStateManager;
 import org.apache.hadoop.hbase.executor.ExecutorService;
 import org.apache.hadoop.hbase.executor.ExecutorType;
 import org.apache.hadoop.hbase.testclassification.MediumTests;
 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
 import org.apache.hadoop.hbase.util.CancelableProgressable;
 import org.apache.hadoop.hbase.util.MockServer;
 import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
 import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
 import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 @Category({RegionServerTests.class, MediumTests.class})
 public class TestSplitLogWorker {

   @ClassRule
   public static final HBaseClassTestRule CLASS_RULE =
       HBaseClassTestRule.forClass(TestSplitLogWorker.class);

   private static final Logger LOG = LoggerFactory.getLogger(TestSplitLogWorker.class);
   private static final int WAIT_TIME = 15000;
   private final ServerName MANAGER = ServerName.valueOf("manager,1,1");
   private final static HBaseTestingUtility TEST_UTIL =
     new HBaseTestingUtility();
   private DummyServer ds;
   private ZKWatcher zkw;
   private SplitLogWorker slw;
   private ExecutorService executorService;

   static class DummyServer extends MockServer {
     private ZKWatcher zkw;
     private Configuration conf;
     private CoordinatedStateManager cm;

     public DummyServer(ZKWatcher zkw, Configuration conf) {
       this.zkw = zkw;
       this.conf = conf;
       cm = new ZkCoordinatedStateManager(this);
     }

     @Override
     public Configuration getConfiguration() {
       return conf;
     }

     @Override
     public ZKWatcher getZooKeeper() {
       return zkw;
     }

     @Override
     public CoordinatedStateManager getCoordinatedStateManager() {
       return cm;
     }
   }

   private void waitForCounter(LongAdder ctr, long oldval, long newval, long timems)
       throws Exception {
     assertTrue("ctr=" + ctr.sum() + ", oldval=" + oldval + ", newval=" + newval,
       waitForCounterBoolean(ctr, oldval, newval, timems));
   }

   private boolean waitForCounterBoolean(final LongAdder ctr, final long oldval, long newval,
       long timems) throws Exception {

     return waitForCounterBoolean(ctr, oldval, newval, timems, true);
   }

   private boolean waitForCounterBoolean(final LongAdder ctr, final long oldval, final long newval,
       long timems, boolean failIfTimeout) throws Exception {

     long timeWaited = TEST_UTIL.waitFor(timems, 10, failIfTimeout,
       new Waiter.Predicate<Exception>() {
       @Override
       public boolean evaluate() throws Exception {
             return (ctr.sum() >= newval);
       }
     });

     if( timeWaited > 0) {
       // when not timed out
       assertEquals(newval, ctr.sum());
     }
     return true;
   }

   @Before
   public void setup() throws Exception {
     TEST_UTIL.startMiniZKCluster();
     Configuration conf = TEST_UTIL.getConfiguration();
     zkw = new ZKWatcher(TEST_UTIL.getConfiguration(),
         "split-log-worker-tests", null);
     ds = new DummyServer(zkw, conf);
     ZKUtil.deleteChildrenRecursively(zkw, zkw.getZNodePaths().baseZNode);
     ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().baseZNode);
     assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().baseZNode), not(is(-1)));
     LOG.debug(zkw.getZNodePaths().baseZNode + " created");
     ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().splitLogZNode);
     assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().splitLogZNode), not(is(-1)));

     LOG.debug(zkw.getZNodePaths().splitLogZNode + " created");
     ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().rsZNode);
     assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().rsZNode), not(is(-1)));

     SplitLogCounters.resetCounters();
     executorService = new ExecutorService("TestSplitLogWorker");
     executorService.startExecutorService(executorService.new ExecutorConfig().setExecutorType(
         ExecutorType.RS_LOG_REPLAY_OPS).setCorePoolSize(10));
   }

   @After
   public void teardown() throws Exception {
     if (executorService != null) {
       executorService.shutdown();
     }
     TEST_UTIL.shutdownMiniZKCluster();
   }

   SplitLogWorker.TaskExecutor neverEndingTask =
     new SplitLogWorker.TaskExecutor() {

       @Override
       public Status exec(String name, CancelableProgressable p) {
         while (true) {
           try {
             Thread.sleep(1000);
           } catch (InterruptedException e) {
             return Status.PREEMPTED;
           }
           if (!p.progress()) {
             return Status.PREEMPTED;
           }
         }
       }

   };

   @Test
   public void testAcquireTaskAtStartup() throws Exception {
     LOG.info("testAcquireTaskAtStartup");
     SplitLogCounters.resetCounters();
     final String TATAS = "tatas";
     final ServerName RS = ServerName.valueOf("rs,1,1");
     RegionServerServices mockedRS = getRegionServer(RS);
     zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TATAS),
       new SplitLogTask.Unassigned(ServerName.valueOf("mgr,1,1")).toByteArray(),
         Ids.OPEN_ACL_UNSAFE,
         CreateMode.PERSISTENT);

     SplitLogWorker slw =
         new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
     slw.start();
     try {
       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
       byte [] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TATAS));
       SplitLogTask slt = SplitLogTask.parseFrom(bytes);
       assertTrue(slt.isOwned(RS));
     } finally {
      stopSplitLogWorker(slw);
     }
   }

   private void stopSplitLogWorker(final SplitLogWorker slw)
   throws InterruptedException {
     if (slw != null) {
       slw.stop();
       slw.worker.join(WAIT_TIME);
       if (slw.worker.isAlive()) {
         assertTrue(("Could not stop the worker thread slw=" + slw) == null);
       }
     }
   }

   @Test
   public void testRaceForTask() throws Exception {
     LOG.info("testRaceForTask");
     SplitLogCounters.resetCounters();
     final String TRFT = "trft";
     final ServerName SVR1 = ServerName.valueOf("svr1,1,1");
     final ServerName SVR2 = ServerName.valueOf("svr2,1,1");
     zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TRFT),
       new SplitLogTask.Unassigned(MANAGER).toByteArray(),
         Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
     RegionServerServices mockedRS1 = getRegionServer(SVR1);
     RegionServerServices mockedRS2 = getRegionServer(SVR2);
     SplitLogWorker slw1 =
         new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS1, neverEndingTask);
     SplitLogWorker slw2 =
         new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS2, neverEndingTask);
     slw1.start();
     slw2.start();
     try {
       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
       // Assert that either the tot_wkr_failed_to_grab_task_owned count was set of if
       // not it, that we fell through to the next counter in line and it was set.
       assertTrue(waitForCounterBoolean(SplitLogCounters.tot_wkr_failed_to_grab_task_owned, 0, 1,
           WAIT_TIME, false) ||
         SplitLogCounters.tot_wkr_failed_to_grab_task_lost_race.sum() == 1);
       byte [] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TRFT));
       SplitLogTask slt = SplitLogTask.parseFrom(bytes);
       assertTrue(slt.isOwned(SVR1) || slt.isOwned(SVR2));
     } finally {
       stopSplitLogWorker(slw1);
       stopSplitLogWorker(slw2);
     }
   }

   @Test
   public void testPreemptTask() throws Exception {
     LOG.info("testPreemptTask");
     SplitLogCounters.resetCounters();
     final ServerName SRV = ServerName.valueOf("tpt_svr,1,1");
     final String PATH = ZKSplitLog.getEncodedNodeName(zkw, "tpt_task");
     RegionServerServices mockedRS = getRegionServer(SRV);
     SplitLogWorker slw =
         new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
     slw.start();
     try {
       Thread.yield(); // let the worker start
       Thread.sleep(1000);
       waitForCounter(SplitLogCounters.tot_wkr_task_grabing, 0, 1, WAIT_TIME);

       // this time create a task node after starting the splitLogWorker
       zkw.getRecoverableZooKeeper().create(PATH,
         new SplitLogTask.Unassigned(MANAGER).toByteArray(),
         Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
       assertEquals(1, slw.getTaskReadySeq());
       byte [] bytes = ZKUtil.getData(zkw, PATH);
       SplitLogTask slt = SplitLogTask.parseFrom(bytes);
       assertTrue(slt.isOwned(SRV));
       slt = new SplitLogTask.Owned(MANAGER);
       ZKUtil.setData(zkw, PATH, slt.toByteArray());
       waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);
     } finally {
       stopSplitLogWorker(slw);
     }
   }

   @Test
   public void testMultipleTasks() throws Exception {
     LOG.info("testMultipleTasks");
     SplitLogCounters.resetCounters();
     final ServerName SRV = ServerName.valueOf("tmt_svr,1,1");
     final String PATH1 = ZKSplitLog.getEncodedNodeName(zkw, "tmt_task");
     RegionServerServices mockedRS = getRegionServer(SRV);
     SplitLogWorker slw =
         new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
     slw.start();
     try {
       Thread.yield(); // let the worker start
       Thread.sleep(100);
       waitForCounter(SplitLogCounters.tot_wkr_task_grabing, 0, 1, WAIT_TIME);

       SplitLogTask unassignedManager =
         new SplitLogTask.Unassigned(MANAGER);
       zkw.getRecoverableZooKeeper().create(PATH1, unassignedManager.toByteArray(),
         Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
       // now the worker is busy doing the above task

       // create another task
       final String PATH2 = ZKSplitLog.getEncodedNodeName(zkw, "tmt_task_2");
       zkw.getRecoverableZooKeeper().create(PATH2, unassignedManager.toByteArray(),
         Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

       // preempt the first task, have it owned by another worker
       final ServerName anotherWorker = ServerName.valueOf("another-worker,1,1");
       SplitLogTask slt = new SplitLogTask.Owned(anotherWorker);
       ZKUtil.setData(zkw, PATH1, slt.toByteArray());
       waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);

       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 1, 2, WAIT_TIME);
       assertEquals(2, slw.getTaskReadySeq());
       byte [] bytes = ZKUtil.getData(zkw, PATH2);
       slt = SplitLogTask.parseFrom(bytes);
       assertTrue(slt.isOwned(SRV));
     } finally {
       stopSplitLogWorker(slw);
     }
   }

   @Test
   public void testRescan() throws Exception {
     LOG.info("testRescan");
     SplitLogCounters.resetCounters();
     final ServerName SRV = ServerName.valueOf("svr,1,1");
     RegionServerServices mockedRS = getRegionServer(SRV);
     slw = new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
     slw.start();
     Thread.yield(); // let the worker start
     Thread.sleep(100);

     String task = ZKSplitLog.getEncodedNodeName(zkw, "task");
     SplitLogTask slt = new SplitLogTask.Unassigned(MANAGER);
     zkw.getRecoverableZooKeeper().create(task,slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
       CreateMode.PERSISTENT);

     waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
     // now the worker is busy doing the above task

     // preempt the task, have it owned by another worker
     ZKUtil.setData(zkw, task, slt.toByteArray());
     waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);

     // create a RESCAN node
     String rescan = ZKSplitLog.getEncodedNodeName(zkw, "RESCAN");
     rescan = zkw.getRecoverableZooKeeper().create(rescan, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
       CreateMode.PERSISTENT_SEQUENTIAL);

     waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 1, 2, WAIT_TIME);
     // RESCAN node might not have been processed if the worker became busy
     // with the above task. preempt the task again so that now the RESCAN
     // node is processed
     ZKUtil.setData(zkw, task, slt.toByteArray());
     waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 1, 2, WAIT_TIME);
     waitForCounter(SplitLogCounters.tot_wkr_task_acquired_rescan, 0, 1, WAIT_TIME);

     List<String> nodes = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().splitLogZNode);
     LOG.debug(Objects.toString(nodes));
     int num = 0;
     for (String node : nodes) {
       num++;
       if (node.startsWith("RESCAN")) {
         String name = ZKSplitLog.getEncodedNodeName(zkw, node);
         String fn = ZKSplitLog.getFileName(name);
         byte [] data = ZKUtil.getData(zkw,
                 ZNodePaths.joinZNode(zkw.getZNodePaths().splitLogZNode, fn));
         slt = SplitLogTask.parseFrom(data);
         assertTrue(slt.toString(), slt.isDone(SRV));
       }
     }
     assertEquals(2, num);
   }

   @Test
   public void testAcquireMultiTasks() throws Exception {
     LOG.info("testAcquireMultiTasks");
     SplitLogCounters.resetCounters();
     final String TATAS = "tatas";
     final ServerName RS = ServerName.valueOf("rs,1,1");
     final int maxTasks = 3;
     Configuration testConf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
     testConf.setInt(HBASE_SPLIT_WAL_MAX_SPLITTER, maxTasks);
     RegionServerServices mockedRS = getRegionServer(RS);
     for (int i = 0; i < maxTasks; i++) {
       zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TATAS + i),
         new SplitLogTask.Unassigned(ServerName.valueOf("mgr,1,1")).toByteArray(),
           Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
     }

     SplitLogWorker slw = new SplitLogWorker(ds, testConf, mockedRS, neverEndingTask);
     slw.start();
     try {
       waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, maxTasks, WAIT_TIME);
       for (int i = 0; i < maxTasks; i++) {
         byte[] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TATAS + i));
         SplitLogTask slt = SplitLogTask.parseFrom(bytes);
         assertTrue(slt.isOwned(RS));
       }
     } finally {
       stopSplitLogWorker(slw);
     }
   }

   /**
    * Create a mocked region server service instance
    */
   private RegionServerServices getRegionServer(ServerName name) {

     RegionServerServices mockedServer = mock(RegionServerServices.class);
     when(mockedServer.getConfiguration()).thenReturn(TEST_UTIL.getConfiguration());
     when(mockedServer.getServerName()).thenReturn(name);
     when(mockedServer.getZooKeeper()).thenReturn(zkw);
     when(mockedServer.isStopped()).thenReturn(false);
     when(mockedServer.getExecutorService()).thenReturn(executorService);

     return mockedServer;
   }

 }
	/**
	* 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.
	*/
	package org.apache.hadoop.hbase.regionserver;

	import static org.apache.hadoop.hbase.HConstants.HBASE_SPLIT_WAL_MAX_SPLITTER;
	import static org.hamcrest.CoreMatchers.is;
	import static org.hamcrest.CoreMatchers.not;
	import static org.hamcrest.MatcherAssert.assertThat;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertTrue;
	import static org.mockito.Mockito.mock;
	import static org.mockito.Mockito.when;

	import java.util.List;
	import java.util.Objects;
	import java.util.concurrent.atomic.LongAdder;

	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.hbase.CoordinatedStateManager;
	import org.apache.hadoop.hbase.HBaseClassTestRule;
	import org.apache.hadoop.hbase.HBaseConfiguration;
	import org.apache.hadoop.hbase.HBaseTestingUtility;
	import org.apache.hadoop.hbase.ServerName;
	import org.apache.hadoop.hbase.SplitLogCounters;
	import org.apache.hadoop.hbase.SplitLogTask;
	import org.apache.hadoop.hbase.Waiter;
	import org.apache.hadoop.hbase.coordination.ZkCoordinatedStateManager;
	import org.apache.hadoop.hbase.executor.ExecutorService;
	import org.apache.hadoop.hbase.executor.ExecutorType;
	import org.apache.hadoop.hbase.testclassification.MediumTests;
	import org.apache.hadoop.hbase.testclassification.RegionServerTests;
	import org.apache.hadoop.hbase.util.CancelableProgressable;
	import org.apache.hadoop.hbase.util.MockServer;
	import org.apache.hadoop.hbase.zookeeper.ZKSplitLog;
	import org.apache.hadoop.hbase.zookeeper.ZKUtil;
	import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
	import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
	import org.apache.zookeeper.CreateMode;
	import org.apache.zookeeper.ZooDefs.Ids;
	import org.junit.After;
	import org.junit.Before;
	import org.junit.ClassRule;
	import org.junit.Test;
	import org.junit.experimental.categories.Category;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	@Category({RegionServerTests.class, MediumTests.class})
	public class TestSplitLogWorker {

	@ClassRule
	public static final HBaseClassTestRule CLASS_RULE =
	HBaseClassTestRule.forClass(TestSplitLogWorker.class);

	private static final Logger LOG = LoggerFactory.getLogger(TestSplitLogWorker.class);
	private static final int WAIT_TIME = 15000;
	private final ServerName MANAGER = ServerName.valueOf("manager,1,1");
	private final static HBaseTestingUtility TEST_UTIL =
	new HBaseTestingUtility();
	private DummyServer ds;
	private ZKWatcher zkw;
	private SplitLogWorker slw;
	private ExecutorService executorService;

	static class DummyServer extends MockServer {
	private ZKWatcher zkw;
	private Configuration conf;
	private CoordinatedStateManager cm;

	public DummyServer(ZKWatcher zkw, Configuration conf) {
	this.zkw = zkw;
	this.conf = conf;
	cm = new ZkCoordinatedStateManager(this);
	}

	@Override
	public Configuration getConfiguration() {
	return conf;
	}

	@Override
	public ZKWatcher getZooKeeper() {
	return zkw;
	}

	@Override
	public CoordinatedStateManager getCoordinatedStateManager() {
	return cm;
	}
	}

	private void waitForCounter(LongAdder ctr, long oldval, long newval, long timems)
	throws Exception {
	assertTrue("ctr=" + ctr.sum() + ", oldval=" + oldval + ", newval=" + newval,
	waitForCounterBoolean(ctr, oldval, newval, timems));
	}

	private boolean waitForCounterBoolean(final LongAdder ctr, final long oldval, long newval,
	long timems) throws Exception {

	return waitForCounterBoolean(ctr, oldval, newval, timems, true);
	}

	private boolean waitForCounterBoolean(final LongAdder ctr, final long oldval, final long newval,
	long timems, boolean failIfTimeout) throws Exception {

	long timeWaited = TEST_UTIL.waitFor(timems, 10, failIfTimeout,
	new Waiter.Predicate<Exception>() {
	@Override
	public boolean evaluate() throws Exception {
	return (ctr.sum() >= newval);
	}
	});

	if( timeWaited > 0) {
	// when not timed out
	assertEquals(newval, ctr.sum());
	}
	return true;
	}

	@Before
	public void setup() throws Exception {
	TEST_UTIL.startMiniZKCluster();
	Configuration conf = TEST_UTIL.getConfiguration();
	zkw = new ZKWatcher(TEST_UTIL.getConfiguration(),
	"split-log-worker-tests", null);
	ds = new DummyServer(zkw, conf);
	ZKUtil.deleteChildrenRecursively(zkw, zkw.getZNodePaths().baseZNode);
	ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().baseZNode);
	assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().baseZNode), not(is(-1)));
	LOG.debug(zkw.getZNodePaths().baseZNode + " created");
	ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().splitLogZNode);
	assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().splitLogZNode), not(is(-1)));

	LOG.debug(zkw.getZNodePaths().splitLogZNode + " created");
	ZKUtil.createAndFailSilent(zkw, zkw.getZNodePaths().rsZNode);
	assertThat(ZKUtil.checkExists(zkw, zkw.getZNodePaths().rsZNode), not(is(-1)));

	SplitLogCounters.resetCounters();
	executorService = new ExecutorService("TestSplitLogWorker");
	executorService.startExecutorService(executorService.new ExecutorConfig().setExecutorType(
	ExecutorType.RS_LOG_REPLAY_OPS).setCorePoolSize(10));
	}

	@After
	public void teardown() throws Exception {
	if (executorService != null) {
	executorService.shutdown();
	}
	TEST_UTIL.shutdownMiniZKCluster();
	}

	SplitLogWorker.TaskExecutor neverEndingTask =
	new SplitLogWorker.TaskExecutor() {

	@Override
	public Status exec(String name, CancelableProgressable p) {
	while (true) {
	try {
	Thread.sleep(1000);
	} catch (InterruptedException e) {
	return Status.PREEMPTED;
	}
	if (!p.progress()) {
	return Status.PREEMPTED;
	}
	}
	}

	};

	@Test
	public void testAcquireTaskAtStartup() throws Exception {
	LOG.info("testAcquireTaskAtStartup");
	SplitLogCounters.resetCounters();
	final String TATAS = "tatas";
	final ServerName RS = ServerName.valueOf("rs,1,1");
	RegionServerServices mockedRS = getRegionServer(RS);
	zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TATAS),
	new SplitLogTask.Unassigned(ServerName.valueOf("mgr,1,1")).toByteArray(),
	Ids.OPEN_ACL_UNSAFE,
	CreateMode.PERSISTENT);

	SplitLogWorker slw =
	new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
	slw.start();
	try {
	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
	byte [] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TATAS));
	SplitLogTask slt = SplitLogTask.parseFrom(bytes);
	assertTrue(slt.isOwned(RS));
	} finally {
	stopSplitLogWorker(slw);
	}
	}

	private void stopSplitLogWorker(final SplitLogWorker slw)
	throws InterruptedException {
	if (slw != null) {
	slw.stop();
	slw.worker.join(WAIT_TIME);
	if (slw.worker.isAlive()) {
	assertTrue(("Could not stop the worker thread slw=" + slw) == null);
	}
	}
	}

	@Test
	public void testRaceForTask() throws Exception {
	LOG.info("testRaceForTask");
	SplitLogCounters.resetCounters();
	final String TRFT = "trft";
	final ServerName SVR1 = ServerName.valueOf("svr1,1,1");
	final ServerName SVR2 = ServerName.valueOf("svr2,1,1");
	zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TRFT),
	new SplitLogTask.Unassigned(MANAGER).toByteArray(),
	Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	RegionServerServices mockedRS1 = getRegionServer(SVR1);
	RegionServerServices mockedRS2 = getRegionServer(SVR2);
	SplitLogWorker slw1 =
	new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS1, neverEndingTask);
	SplitLogWorker slw2 =
	new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS2, neverEndingTask);
	slw1.start();
	slw2.start();
	try {
	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
	// Assert that either the tot_wkr_failed_to_grab_task_owned count was set of if
	// not it, that we fell through to the next counter in line and it was set.
	assertTrue(waitForCounterBoolean(SplitLogCounters.tot_wkr_failed_to_grab_task_owned, 0, 1,
	WAIT_TIME, false) \|\|
	SplitLogCounters.tot_wkr_failed_to_grab_task_lost_race.sum() == 1);
	byte [] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TRFT));
	SplitLogTask slt = SplitLogTask.parseFrom(bytes);
	assertTrue(slt.isOwned(SVR1) \|\| slt.isOwned(SVR2));
	} finally {
	stopSplitLogWorker(slw1);
	stopSplitLogWorker(slw2);
	}
	}

	@Test
	public void testPreemptTask() throws Exception {
	LOG.info("testPreemptTask");
	SplitLogCounters.resetCounters();
	final ServerName SRV = ServerName.valueOf("tpt_svr,1,1");
	final String PATH = ZKSplitLog.getEncodedNodeName(zkw, "tpt_task");
	RegionServerServices mockedRS = getRegionServer(SRV);
	SplitLogWorker slw =
	new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
	slw.start();
	try {
	Thread.yield(); // let the worker start
	Thread.sleep(1000);
	waitForCounter(SplitLogCounters.tot_wkr_task_grabing, 0, 1, WAIT_TIME);

	// this time create a task node after starting the splitLogWorker
	zkw.getRecoverableZooKeeper().create(PATH,
	new SplitLogTask.Unassigned(MANAGER).toByteArray(),
	Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
	assertEquals(1, slw.getTaskReadySeq());
	byte [] bytes = ZKUtil.getData(zkw, PATH);
	SplitLogTask slt = SplitLogTask.parseFrom(bytes);
	assertTrue(slt.isOwned(SRV));
	slt = new SplitLogTask.Owned(MANAGER);
	ZKUtil.setData(zkw, PATH, slt.toByteArray());
	waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);
	} finally {
	stopSplitLogWorker(slw);
	}
	}

	@Test
	public void testMultipleTasks() throws Exception {
	LOG.info("testMultipleTasks");
	SplitLogCounters.resetCounters();
	final ServerName SRV = ServerName.valueOf("tmt_svr,1,1");
	final String PATH1 = ZKSplitLog.getEncodedNodeName(zkw, "tmt_task");
	RegionServerServices mockedRS = getRegionServer(SRV);
	SplitLogWorker slw =
	new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
	slw.start();
	try {
	Thread.yield(); // let the worker start
	Thread.sleep(100);
	waitForCounter(SplitLogCounters.tot_wkr_task_grabing, 0, 1, WAIT_TIME);

	SplitLogTask unassignedManager =
	new SplitLogTask.Unassigned(MANAGER);
	zkw.getRecoverableZooKeeper().create(PATH1, unassignedManager.toByteArray(),
	Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
	// now the worker is busy doing the above task

	// create another task
	final String PATH2 = ZKSplitLog.getEncodedNodeName(zkw, "tmt_task_2");
	zkw.getRecoverableZooKeeper().create(PATH2, unassignedManager.toByteArray(),
	Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

	// preempt the first task, have it owned by another worker
	final ServerName anotherWorker = ServerName.valueOf("another-worker,1,1");
	SplitLogTask slt = new SplitLogTask.Owned(anotherWorker);
	ZKUtil.setData(zkw, PATH1, slt.toByteArray());
	waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);

	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 1, 2, WAIT_TIME);
	assertEquals(2, slw.getTaskReadySeq());
	byte [] bytes = ZKUtil.getData(zkw, PATH2);
	slt = SplitLogTask.parseFrom(bytes);
	assertTrue(slt.isOwned(SRV));
	} finally {
	stopSplitLogWorker(slw);
	}
	}

	@Test
	public void testRescan() throws Exception {
	LOG.info("testRescan");
	SplitLogCounters.resetCounters();
	final ServerName SRV = ServerName.valueOf("svr,1,1");
	RegionServerServices mockedRS = getRegionServer(SRV);
	slw = new SplitLogWorker(ds, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask);
	slw.start();
	Thread.yield(); // let the worker start
	Thread.sleep(100);

	String task = ZKSplitLog.getEncodedNodeName(zkw, "task");
	SplitLogTask slt = new SplitLogTask.Unassigned(MANAGER);
	zkw.getRecoverableZooKeeper().create(task,slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
	CreateMode.PERSISTENT);

	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME);
	// now the worker is busy doing the above task

	// preempt the task, have it owned by another worker
	ZKUtil.setData(zkw, task, slt.toByteArray());
	waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME);

	// create a RESCAN node
	String rescan = ZKSplitLog.getEncodedNodeName(zkw, "RESCAN");
	rescan = zkw.getRecoverableZooKeeper().create(rescan, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
	CreateMode.PERSISTENT_SEQUENTIAL);

	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 1, 2, WAIT_TIME);
	// RESCAN node might not have been processed if the worker became busy
	// with the above task. preempt the task again so that now the RESCAN
	// node is processed
	ZKUtil.setData(zkw, task, slt.toByteArray());
	waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 1, 2, WAIT_TIME);
	waitForCounter(SplitLogCounters.tot_wkr_task_acquired_rescan, 0, 1, WAIT_TIME);

	List<String> nodes = ZKUtil.listChildrenNoWatch(zkw, zkw.getZNodePaths().splitLogZNode);
	LOG.debug(Objects.toString(nodes));
	int num = 0;
	for (String node : nodes) {
	num++;
	if (node.startsWith("RESCAN")) {
	String name = ZKSplitLog.getEncodedNodeName(zkw, node);
	String fn = ZKSplitLog.getFileName(name);
	byte [] data = ZKUtil.getData(zkw,
	ZNodePaths.joinZNode(zkw.getZNodePaths().splitLogZNode, fn));
	slt = SplitLogTask.parseFrom(data);
	assertTrue(slt.toString(), slt.isDone(SRV));
	}
	}
	assertEquals(2, num);
	}

	@Test
	public void testAcquireMultiTasks() throws Exception {
	LOG.info("testAcquireMultiTasks");
	SplitLogCounters.resetCounters();
	final String TATAS = "tatas";
	final ServerName RS = ServerName.valueOf("rs,1,1");
	final int maxTasks = 3;
	Configuration testConf = HBaseConfiguration.create(TEST_UTIL.getConfiguration());
	testConf.setInt(HBASE_SPLIT_WAL_MAX_SPLITTER, maxTasks);
	RegionServerServices mockedRS = getRegionServer(RS);
	for (int i = 0; i < maxTasks; i++) {
	zkw.getRecoverableZooKeeper().create(ZKSplitLog.getEncodedNodeName(zkw, TATAS + i),
	new SplitLogTask.Unassigned(ServerName.valueOf("mgr,1,1")).toByteArray(),
	Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
	}

	SplitLogWorker slw = new SplitLogWorker(ds, testConf, mockedRS, neverEndingTask);
	slw.start();
	try {
	waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, maxTasks, WAIT_TIME);
	for (int i = 0; i < maxTasks; i++) {
	byte[] bytes = ZKUtil.getData(zkw, ZKSplitLog.getEncodedNodeName(zkw, TATAS + i));
	SplitLogTask slt = SplitLogTask.parseFrom(bytes);
	assertTrue(slt.isOwned(RS));
	}
	} finally {
	stopSplitLogWorker(slw);
	}
	}

	/**
	* Create a mocked region server service instance
	*/
	private RegionServerServices getRegionServer(ServerName name) {

	RegionServerServices mockedServer = mock(RegionServerServices.class);
	when(mockedServer.getConfiguration()).thenReturn(TEST_UTIL.getConfiguration());
	when(mockedServer.getServerName()).thenReturn(name);
	when(mockedServer.getZooKeeper()).thenReturn(zkw);
	when(mockedServer.isStopped()).thenReturn(false);
	when(mockedServer.getExecutorService()).thenReturn(executorService);

	return mockedServer;
	}

	}