@Test public void testOrphanTaskAcquisition() throws Exception { LOG.info("TestOrphanTaskAcquisition"); String tasknode = ZKSplitLog.getEncodedNodeName(zkw, "orphan/test/slash"); SplitLogTask slt = new SplitLogTask.Owned(DUMMY_MASTER, this.mode); zkw.getRecoverableZooKeeper().create(tasknode, slt.toByteArray(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); slm = new SplitLogManager(zkw, conf, stopper, master, DUMMY_MASTER); waitForCounter(tot_mgr_orphan_task_acquired, 0, 1, to/2); Task task = slm.findOrCreateOrphanTask(tasknode); assertTrue(task.isOrphan()); waitForCounter(tot_mgr_heartbeat, 0, 1, to/2); assertFalse(task.isUnassigned()); long curt = System.currentTimeMillis(); 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); }
Task findOrCreateOrphanTask(String path) { Task orphanTask = new Task(); Task task; task = tasks.putIfAbsent(path, orphanTask); if (task == null) { LOG.info("creating orphan task " + path); SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet(); task = orphanTask; } return task; }
SplitLogManager extends ZooKeeperListener { Task findOrCreateOrphanTask(String path) { Task orphanTask = new Task(); Task task; task = tasks.putIfAbsent(path, orphanTask); if (task == null) { LOG.info("creating orphan task " + path); SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet(); task = orphanTask; } return task; } }
SplitLogManager extends ZooKeeperListener { Task findOrCreateOrphanTask(String path) { Task orphanTask = new Task(); Task task; task = tasks.putIfAbsent(path, orphanTask); if (task == null) { LOG.info("creating orphan task " + path); SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet(); task = orphanTask; } return task; } SplitLogManager(ZooKeeperWatcher zkw, final Configuration conf,
      Stoppable stopper, MasterServices master, ServerName serverName);  SplitLogManager(ZooKeeperWatcher zkw, Configuration conf, Stoppable stopper,
      MasterServices master, ServerName serverName, TaskFinisher tf); }
SplitLogManager extends ZooKeeperListener { Task findOrCreateOrphanTask(String path) { Task orphanTask = new Task(); Task task; task = tasks.putIfAbsent(path, orphanTask); if (task == null) { LOG.info("creating orphan task " + path); SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet(); task = orphanTask; } return task; } SplitLogManager(ZooKeeperWatcher zkw, final Configuration conf,
      Stoppable stopper, MasterServices master, ServerName serverName);  SplitLogManager(ZooKeeperWatcher zkw, Configuration conf, Stoppable stopper,
      MasterServices master, ServerName serverName, 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); static void deleteRecoveringRegionZNodes(ZooKeeperWatcher watcher, List<String> regions); @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); void setRecoveryMode(boolean isForInitialization); RecoveryMode getRecoveryMode(); }
SplitLogManager extends ZooKeeperListener { Task findOrCreateOrphanTask(String path) { Task orphanTask = new Task(); Task task; task = tasks.putIfAbsent(path, orphanTask); if (task == null) { LOG.info("creating orphan task " + path); SplitLogCounters.tot_mgr_orphan_task_acquired.incrementAndGet(); task = orphanTask; } return task; } SplitLogManager(ZooKeeperWatcher zkw, final Configuration conf,
      Stoppable stopper, MasterServices master, ServerName serverName);  SplitLogManager(ZooKeeperWatcher zkw, Configuration conf, Stoppable stopper,
      MasterServices master, ServerName serverName, 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); static void deleteRecoveringRegionZNodes(ZooKeeperWatcher watcher, List<String> regions); @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); void setRecoveryMode(boolean isForInitialization); RecoveryMode getRecoveryMode(); 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; }