duration: Long,
    unit: TimeUnit,
  ) = apply {
    actions += { Thread.sleep(unit.toMillis(duration)) }
  }

  override fun handle(stream: Stream) {
    val task = serviceStreamTask(stream)
    results.add(task)
    task.run()
  }

  /** Returns a task that processes both request and response from [stream]. */
  private fun serviceStreamTask(stream: Stream): FutureTask<Void> {
    return FutureTask<Void> {

    // During execution:
    //   * task:0 processes main function `VariableAddFunctionWithGraphError`
    //     and places the 'read0_maybe_with_graph_error' op on task:2
    //   * task:0 partitions the main function with a subgraph containing
    //     'read0_maybe_with_graph_error' sent to task:2
    //   * task:2 graph pass reports an error when it sees

  }

  /**
   * Submits a blocking task for the given {@link Future} to provide {@link ListenableFuture}
   * functionality.
   *
   * <p><b>Warning:</b> If the input future does not already implement {@code ListenableFuture}, the
   * returned future will emulate {@link ListenableFuture#addListener} by submitting a task to the
   * given executor at the first call to {@code addListener}. The task must be started by the

  var nanoTime = 0L
    private set

  /** Backlog of tasks to run. Only one task runs at a time. Guarded by [TaskRunner.lock]. */
  private val serialTaskQueue = ArrayDeque<SerialTask>()

  /** The task that's currently executing. Guarded by [TaskRunner.lock]. */
  private var currentTask: SerialTask = TestThreadSerialTask

  /** The coordinator task if it's waiting, and how it will resume. Guarded by [TaskRunner.lock]. */

  }

  @Test fun taskFailsWithUncheckedException() {
    queue.schedule("task", TimeUnit.MILLISECONDS.toNanos(100)) {
      log.put("failing task running")
      throw RuntimeException("boom!")
    }

    queue.schedule("task", TimeUnit.MILLISECONDS.toNanos(200)) {
      log.put("normal task running")
      return@schedule -1L
    }

    queue.idleLatch().await(500, TimeUnit.MILLISECONDS)

                    PluginVersionResolutionException {
        List<TaskSegment> taskSegments = new ArrayList<>(tasks.size());

        TaskSegment currentSegment = null;

        for (String task : tasks) {
            if (isGoalSpecification(task)) {
                // "pluginPrefix[:version]:goal" or "groupId:artifactId[:version]:goal"

                lifecyclePluginResolver.resolveMissingPluginVersions(session.getTopLevelProject(), session);

							task.event.StorageClass, task.objInfo.Bucket, task.objInfo.Name, task.objInfo.VersionID, err))
					}
				}
			} else {
				ts := tierStats{
					TotalSize:   uint64(task.objInfo.Size),
					NumVersions: 1,
				}
				if task.objInfo.IsLatest {
					ts.NumObjects = 1
				}
				t.addLastDayStats(task.event.StorageClass, ts)
			}
			t.activeTasks.Add(-1)

    private final List<Task> tasks;

    private final boolean aggregating;

    public TaskSegment(boolean aggregating) {
        this.aggregating = aggregating;
        tasks = new ArrayList<>();
    }

    public TaskSegment(boolean aggregating, Task... tasks) {
        this.aggregating = aggregating;

            logger.warn("Failed to generate a thumbnail of {}: {}", thumbnailId, commandList, e);
        }
        if (task != null && !task.isExecuted()) {
            task.cancel();
        }
    }

    protected static class ProcessDestroyer extends TimerTask {

        private final Process p;

        private final InputStreamThread ist;

        private final AtomicBoolean executed = new AtomicBoolean(false);

                logger.debug("Job {} is running.", id);
            }
            return;
        }

        TimeoutTask task = null;
        try {
            if (scheduledJob.isLoggingEnabled()) {
                jobHelper.store(jobLog);
                task = jobHelper.startMonitorTask(jobLog);
            }

            if (logger.isDebugEnabled()) {