@Test
  public void testOrphanTaskAcquisition() throws Exception {
    LOG.info("TestOrphanTaskAcquisition");

    String tasknode = ZKSplitLog.getEncodedNodeName(zkw, "orphan/test/slash");
    SplitLogTask slt = new SplitLogTask.Owned(master.getServerName());
    zkw.getRecoverableZooKeeper().create(tasknode, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE,
      CreateMode.PERSISTENT);

    slm = new SplitLogManager(master, conf);
    waitForCounter(tot_mgr_orphan_task_acquired, 0, 1, to / 2);
    Task task = findOrCreateOrphanTask(tasknode);
    assertTrue(task.isOrphan());
    waitForCounter(tot_mgr_heartbeat, 0, 1, to / 2);
    assertFalse(task.isUnassigned());
    long curt = EnvironmentEdgeManager.currentTime();
    assertTrue((task.last_update <= curt) && (task.last_update > (curt - 1000)));
    LOG.info("waiting for manager to resubmit the orphan task");
    waitForCounter(tot_mgr_resubmit, 0, 1, to + to / 2);
    assertTrue(task.isUnassigned());
    waitForCounter(tot_mgr_rescan, 0, 1, to + to / 2);
  }
Test whether the splitlog correctly creates a task in zookeeper
hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java