public void submit(Zombie zombie) {
    Task<?> reapTask = zombie.reap().onFailure(e -> {
      if (ZKUtil.isRecoverable(e)) {
        submit(zombie);
      } else {
        LOG.error("failed to delete zombie node: {}", zombie);
      }
    });

    _engine.run(reapTask);
  }
@Test
  public void testReaperRetry()
      throws InterruptedException {
    final int MAX_RETRY = 3;
    final CountDownLatch failure = new CountDownLatch(MAX_RETRY);
    final CountDownLatch success = new CountDownLatch(1);

    Reaper.Zombie zombie = new Reaper.Zombie() {
      private int count = 0;

      @Override
      public Task<Void> reap() {
        if (count++ < MAX_RETRY) {
          return Task.action(() -> failure.countDown())
              .andThen(Task.failure(KeeperException.create(KeeperException.Code.CONNECTIONLOSS)));
        } else {
          return Task.action(() -> success.countDown());
        }
      }
    };
    _reaper.submit(zombie);

    Assert.assertTrue(failure.await(10, TimeUnit.SECONDS));
    Assert.assertTrue(success.await(10, TimeUnit.SECONDS));
  }