assertFalse(task.isDone());
    assertFalse(task.isCancelled());

    // Start the task to put it in the RUNNING state.  Have to use a separate
    // thread because the task will block on the task latch after unblocking
    // the run latch.
    exec.execute(task);
    runLatch.await();
    assertEquals(1, listenerLatch.getCount());
    assertFalse(task.isDone());
    assertFalse(task.isCancelled());

            AbstractService service, ScheduledExecutorService executor, Runnable task) {
          return new FutureAsCancellable(
              executor.scheduleWithFixedDelay(task, initialDelay, delay, unit));
        }
      };
    }

    /**
     * Returns a {@link Scheduler} that schedules the task using the {@link
     * ScheduledExecutorService#scheduleAtFixedRate} method.
     *

      startAnotherThread()
    }
  }

  private fun beforeRun(task: Task) {
    assertLockHeld()

    task.nextExecuteNanoTime = -1L
    val queue = task.queue!!
    queue.futureTasks.remove(task)
    readyQueues.remove(queue)
    queue.activeTask = task
    busyQueues.add(queue)
  }

  private fun afterRun(
    task: Task,
    delayNanos: Long,
    completedNormally: Boolean,
  ) {

    return delegate().submit(task);
  }

  @Override
  public ListenableFuture<?> submit(Runnable task) {
    return delegate().submit(task);
  }

  @Override
  public <T extends @Nullable Object> ListenableFuture<T> submit(
      Runnable task, @ParametricNullness T result) {
    return delegate().submit(task, result);
  }

  @Throws(IOException::class)
  internal fun removeEntry(entry: Entry): Boolean {
    // If we can't delete files that are still open, mark this entry as a zombie so its files will
    // be deleted when those files are closed.
    if (!civilizedFileSystem) {
      if (entry.lockingSourceCount > 0) {
        // Mark this entry as 'DIRTY' so that if the process crashes this entry won't be used.
        journalWriter?.let {
          it.writeUtf8(DIRTY)

    return delegate().submit(task);
  }

  @Override
  public ListenableFuture<?> submit(Runnable task) {
    return delegate().submit(task);
  }

  @Override
  public <T extends @Nullable Object> ListenableFuture<T> submit(
      Runnable task, @ParametricNullness T result) {
    return delegate().submit(task, result);
  }

  public <T> ListenableFuture<T> submit(Callable<T> task) {
    Preconditions.checkNotNull(task, "task must not be null!");
    return delegate.submit(task);
  }

  @Override
  public <T> ListenableFuture<T> submit(Runnable task, T result) {
    Preconditions.checkNotNull(task, "task must not be null!");
    Preconditions.checkNotNull(result, "result must not be null!");
    return delegate.submit(task, result);
  }

  @Override

  public <T> ListenableFuture<T> submit(Callable<T> task) {
    Preconditions.checkNotNull(task, "task must not be null!");
    return delegate.submit(task);
  }

  @Override
  public <T> ListenableFuture<T> submit(Runnable task, T result) {
    Preconditions.checkNotNull(task, "task must not be null!");
    Preconditions.checkNotNull(result, "result must not be null!");
    return delegate.submit(task, result);
  }

  @Override

        assertNotNull(task.getValue3());
        assertEquals("task-id", task.getValue1());
        assertEquals(path, task.getValue2());
        assertEquals("application/pdf", task.getValue3());
    }

    public void test_createTask_withNullPath() {
        // Test task creation with null path
        Map<String, Object> docMap = new HashMap<>();
        docMap.put("mimetype", "application/pdf");

  @Override
  public ListenableFuture<?> submit(Runnable task) {
    return (ListenableFuture<?>) super.submit(task);
  }

  @CanIgnoreReturnValue // TODO(kak): consider removing this
  @Override
  public <T extends @Nullable Object> ListenableFuture<T> submit(
      Runnable task, @ParametricNullness T result) {
    return (ListenableFuture<T>) super.submit(task, result);
  }