### Question:
SplitLogManager extends ZooKeeperListener { void handleDeadWorker(ServerName workerName) { synchronized (deadWorkersLock) { if (deadWorkers == null) { deadWorkers = new HashSet<ServerName>(100); } deadWorkers.add(workerName); } LOG.info("dead splitlog worker " + workerName); } SplitLogManager(ZooKeeperWatcher zkw, final Configuration conf,
      Stoppable stopper, MasterServices master, ServerName serverName);  SplitLogManager(ZooKeeperWatcher zkw, final Configuration conf,
      Stoppable stopper, MasterServices master, ServerName serverName, boolean masterRecovery);  SplitLogManager(ZooKeeperWatcher zkw, Configuration conf,
        Stoppable stopper, MasterServices master,
        ServerName serverName, boolean masterRecovery, TaskFinisher tf); long splitLogDistributed(final Path logDir); long splitLogDistributed(final List<Path> logDirs); long splitLogDistributed(final Set<ServerName> serverNames, final List<Path> logDirs,
      PathFilter filter); @Override void nodeDataChanged(String path); void stop(); static long parseLastFlushedSequenceIdFrom(final byte[] bytes); static boolean isRegionMarkedRecoveringInZK(ZooKeeperWatcher zkw, String regionEncodedName); static RegionStoreSequenceIds getRegionFlushedSequenceId(ZooKeeperWatcher zkw,
      String serverName, String encodedRegionName); static final int DEFAULT_TIMEOUT; static final int DEFAULT_ZK_RETRIES; static final int DEFAULT_MAX_RESUBMIT; static final int DEFAULT_UNASSIGNED_TIMEOUT; public boolean ignoreZKDeleteForTesting; }### Answer:
@Test public void testDeadWorker() throws Exception { LOG.info("testDeadWorker"); conf.setLong("hbase.splitlog.max.resubmit", 0); slm = new SplitLogManager(zkw, conf, stopper, master, DUMMY_MASTER); 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.get() == 0) waitForCounter(tot_mgr_heartbeat, 0, 1, to/2); slm.handleDeadWorker(worker1); if (tot_mgr_resubmit.get() == 0) waitForCounter(tot_mgr_resubmit, 0, 1, to+to/2); if (tot_mgr_resubmit_dead_server_task.get() == 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(DUMMY_MASTER)); return; }