protected final ListenableFutureTask<Integer> task =
      ListenableFutureTask.create(
          new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
              runLatch.countDown();
              taskLatch.await();
              if (throwException) {
                throw new IllegalStateException("Fail");
              }
              return 25;
            }
          });

                        Thread.currentThread().interrupt();
                    } finally {
                        endLatch.countDown();
                    }
                }
            });
        }

        startLatch.countDown();
        endLatch.await(10, TimeUnit.SECONDS);
        executor.shutdown();

                            rejectedCount.incrementAndGet();
                        }
                    } finally {
                        completeLatch.countDown();
                    }
                });
            }

            startLatch.countDown();
            assertTrue(completeLatch.await(5, TimeUnit.SECONDS), "All tasks should complete");
            executor.shutdown();

                        startLatch.await();
                        response.received();
                        endLatch.countDown();
                    } catch (InterruptedException e) {
                        // Ignore
                    }
                }).start();
            }

            startLatch.countDown();
            assertTrue(endLatch.await(1, TimeUnit.SECONDS));
            assertTrue(response.isReceived());

        cache.getUnchecked(getKey);
        getFinishedSignal.countDown();
      }
    }.start();
    new Thread() {
      @Override
      public void run() {
        cache.refresh(refreshKey);
        getFinishedSignal.countDown();
      }
    }.start();

    // let computation complete
    letGetFinishSignal.countDown();
    getFinishedSignal.await();
    checkNothingLogged();

        cache.getUnchecked(getKey);
        getFinishedSignal.countDown();
      }
    }.start();
    new Thread() {
      @Override
      public void run() {
        cache.refresh(refreshKey);
        getFinishedSignal.countDown();
      }
    }.start();

    // let computation complete
    letGetFinishSignal.countDown();
    getFinishedSignal.await();
    checkNothingLogged();

        } finally {
          closedLatch.countDown()
        }
      }
    }.buffer()

  override val sink: BufferedSink =
    object : ForwardingSink(delegate.sink) {
      private var closed = false

      override fun close() {
        if (closed) return
        try {
          super.close()
        } finally {
          closedLatch.countDown()
        }
      }
    }.buffer()

    service.expectedShutdownState = Service.State.RUNNING;

    service.startAsync().awaitRunning();
    assertTrue(service.startUpCalled);
    assertEquals(Service.State.RUNNING, service.state());

    exitRun.countDown(); // the service will exit voluntarily
    executionThread.join();

    assertTrue(service.shutDownCalled);
    assertEquals(Service.State.TERMINATED, service.state());

    assertThat(closingFuture.cancel(false)).isTrue();
    ValueAndCloser<?> unused = finishToValueAndCloser(closingFuture);
    waitUntilClosed(closingFuture);
    futureCancelled.countDown();
  }

  private <V> ValueAndCloser<V> finishToValueAndCloser(ClosingFuture<V> closingFuture) {
    CountDownLatch valueAndCloserSet = new CountDownLatch(1);
    closingFuture.finishToValueAndCloser(

                    fail("Thread failed: " + e.getMessage());
                } finally {
                    doneLatch.countDown();
                }
            });
        }

        // Start all threads at once
        startLatch.countDown();

        // Wait for completion
        assertTrue(doneLatch.await(5, TimeUnit.SECONDS), "Threads did not complete in time");