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

    conf.setLong("hbase.splitlog.max.resubmit", 0);
    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());
    if (tot_mgr_heartbeat.sum() == 0) {
      waitForCounter(tot_mgr_heartbeat, 0, 1, to / 2);
    }
    slm.handleDeadWorker(worker1);
    if (tot_mgr_resubmit.sum() == 0) {
      waitForCounter(tot_mgr_resubmit, 0, 1, to + to / 2);
    }
    if (tot_mgr_resubmit_dead_server_task.sum() == 0) {
      waitForCounter(tot_mgr_resubmit_dead_server_task, 0, 1, to + to / 2);
    }

    int version1 = ZKUtil.checkExists(zkw, tasknode);
    assertTrue(version1 > version);
    byte[] taskstate = ZKUtil.getData(zkw, tasknode);
    slt = SplitLogTask.parseFrom(taskstate);
    assertTrue(slt.isUnassigned(master.getServerName()));
    return;
  }
Test whether the splitlog correctly creates a task in zookeeper
hbase-server/src/test/java/org/apache/hadoop/hbase/master/TestSplitLogManager.java