docs: public void finishHandoff(final String workUnit) throws InterruptedException
      {
          String unitId = cluster.workUnitMap.get(workUnit);
          LOG.info("Handoff of {} to me acknowledged. Deleting claim ZNode for {} and waiting for {} to shutdown work.",
                  workUnit, workUnit,
                  ((unitId == null) ? "(None)" : unitId));
  
          final String path = cluster.workUnitClaimPath(workUnit);
  
          Stat stat = ZKUtils.exists(cluster.zk, path, new Watcher() {
              @Override
              public void process(WatchedEvent event) {
                  // Don't really care about the type of event here - call unconditionally to clean up state
                  completeHandoff(workUnit, path);
              }
          });
          // Unlikely that peer will have already deleted znode, but handle it regardless
          if (stat == null) {
              LOG.warn("Peer already deleted znode of {}", workUnit);
              completeHandoff(workUnit, path);
          }
      }