ProcessorTasklet implements Tasklet { @Override @Nonnull public ProgressState call() { assert state != END : "already in terminal state"; progTracker.reset(); progTracker.notDone(); outbox.reset(); stateMachineStep(); return progTracker.toProgressState(); } @SuppressWarnings("checkstyle:ExecutableStatementCount")  ProcessorTasklet(
            @Nonnull Context context,
            @Nonnull ExecutorService executionService,
            @Nonnull SerializationService serializationService,
            @Nonnull Processor processor,
            @Nonnull List<? extends InboundEdgeStream> instreams,
            @Nonnull List<? extends OutboundEdgeStream> outstreams,
            @Nonnull SnapshotContext ssContext,
            @Nonnull OutboundCollector ssCollector,
            boolean isSource
    ); @Override void init(); @Override @Nonnull ProgressState call(); @Override MetricsContext getMetricsContext(); @Override String toString(); @Override boolean isCooperative(); @Override void close(); @Override void provideDynamicMetrics(MetricDescriptor descriptor, MetricsCollectionContext context);  }
@Test public void when_closeBlocked_then_waitUntilDone() { processor.doneLatch = new CountDownLatch(1); ProcessorTasklet tasklet = createTasklet(ForkJoinPool.commonPool()); callUntil(tasklet, NO_PROGRESS); processor.doneLatch.countDown(); assertTrueEventually(() -> assertEquals(DONE, tasklet.call()), 2); }