@Test(timeout=60000) 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(zkw, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask); slw.start(); try { Thread.yield(); Thread.sleep(100); waitForCounter(SplitLogCounters.tot_wkr_task_grabing, 0, 1, WAIT_TIME); SplitLogTask unassignedManager = new SplitLogTask.Unassigned(MANAGER, this.mode); zkw.getRecoverableZooKeeper().create(PATH1, unassignedManager.toByteArray(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME); final String PATH2 = ZKSplitLog.getEncodedNodeName(zkw, "tmt_task_2"); zkw.getRecoverableZooKeeper().create(PATH2, unassignedManager.toByteArray(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); final ServerName anotherWorker = ServerName.valueOf("another-worker,1,1"); SplitLogTask slt = new SplitLogTask.Owned(anotherWorker, this.mode); 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.taskReadySeq); byte [] bytes = ZKUtil.getData(zkw, PATH2); slt = SplitLogTask.parseFrom(bytes); assertTrue(slt.isOwned(SRV)); } finally { stopSplitLogWorker(slw); } }
public void start() { worker = new Thread(null, this, "SplitLogWorker-" + serverName); exitWorker = false; worker.start(); }
SplitLogWorker extends ZooKeeperListener implements Runnable { public void start() { worker = new Thread(null, this, "SplitLogWorker-" + serverName); exitWorker = false; worker.start(); } }
SplitLogWorker extends ZooKeeperListener implements Runnable { public void start() { worker = new Thread(null, this, "SplitLogWorker-" + serverName); exitWorker = false; worker.start(); } SplitLogWorker(ZooKeeperWatcher watcher, Configuration conf, RegionServerServices server,
      TaskExecutor splitTaskExecutor);  SplitLogWorker(final ZooKeeperWatcher watcher, final Configuration conf,
      final RegionServerServices server, final LastSequenceId sequenceIdChecker); }
SplitLogWorker extends ZooKeeperListener implements Runnable { public void start() { worker = new Thread(null, this, "SplitLogWorker-" + serverName); exitWorker = false; worker.start(); } SplitLogWorker(ZooKeeperWatcher watcher, Configuration conf, RegionServerServices server,
      TaskExecutor splitTaskExecutor);  SplitLogWorker(final ZooKeeperWatcher watcher, final Configuration conf,
      final RegionServerServices server, final LastSequenceId sequenceIdChecker); @Override void run(); @Override void nodeDataChanged(String path); @Override void nodeChildrenChanged(String path); void start(); void stop(); }
SplitLogWorker extends ZooKeeperListener implements Runnable { public void start() { worker = new Thread(null, this, "SplitLogWorker-" + serverName); exitWorker = false; worker.start(); } SplitLogWorker(ZooKeeperWatcher watcher, Configuration conf, RegionServerServices server,
      TaskExecutor splitTaskExecutor);  SplitLogWorker(final ZooKeeperWatcher watcher, final Configuration conf,
      final RegionServerServices server, final LastSequenceId sequenceIdChecker); @Override void run(); @Override void nodeDataChanged(String path); @Override void nodeChildrenChanged(String path); void start(); void stop(); static final int DEFAULT_MAX_SPLITTERS; }