@Test(timeout=60000) 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(zkw, TEST_UTIL.getConfiguration(), mockedRS, neverEndingTask); slw.start(); Thread.yield(); Thread.sleep(100); String task = ZKSplitLog.getEncodedNodeName(zkw, "task"); SplitLogTask slt = new SplitLogTask.Unassigned(MANAGER, this.mode); zkw.getRecoverableZooKeeper().create(task,slt.toByteArray(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); waitForCounter(SplitLogCounters.tot_wkr_task_acquired, 0, 1, WAIT_TIME); ZKUtil.setData(zkw, task, slt.toByteArray()); waitForCounter(SplitLogCounters.tot_wkr_preempt_task, 0, 1, WAIT_TIME); 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); 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.splitLogZNode); LOG.debug(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, ZKUtil.joinZNode(zkw.splitLogZNode, fn)); slt = SplitLogTask.parseFrom(data); assertTrue(slt.toString(), slt.isDone(SRV)); } } assertEquals(2, num); }
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; }