@Test
  public void testRescanCleanup() throws Exception {
    LOG.info("TestRescanCleanup - ensure RESCAN nodes are cleaned up");

    slm = new SplitLogManager(master, conf);
    TaskBatch batch = new TaskBatch();

    String tasknode = submitTaskAndWait(batch, "foo/1");
    int version = ZKUtil.checkExists(zkw, tasknode);
    final ServerName worker1 = ServerName.valueOf("worker1,1,1");
    SplitLogTask slt = new SplitLogTask.Owned(worker1);
    ZKUtil.setData(zkw, tasknode, slt.toByteArray());
    waitForCounter(tot_mgr_heartbeat, 0, 1, to / 2);
    waitForCounter(new Expr() {
      @Override
      public long eval() {
        return (tot_mgr_resubmit.sum() + tot_mgr_resubmit_failed.sum());
      }
    }, 0, 1, 5 * 60000); // wait long enough
    Assert.assertEquals("Could not run test. Lost ZK connection?", 0,
      tot_mgr_resubmit_failed.sum());
    int version1 = ZKUtil.checkExists(zkw, tasknode);
    assertTrue(version1 > version);
    byte[] taskstate = ZKUtil.getData(zkw, tasknode);
    slt = SplitLogTask.parseFrom(taskstate);
    assertTrue(slt.isUnassigned(master.getServerName()));

    waitForCounter(tot_mgr_rescan_deleted, 0, 1, to / 2);
  }
Test whether the splitlog correctly creates a task in zookeeper
hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java