executor.execute(new ResolveTask(
                            classLoader, latch, artifact, session, node.getRemoteRepositories(), result));
                } else {
                    latch.countDown();
                }
            }
            try {
                latch.await();
            } catch (InterruptedException e) {
                result.addErrorArtifactException(

                  started.countDown();
                  try {
                    return method.invoke(delegate, args);
                  } catch (InvocationTargetException e) {
                    throw e.getCause();
                  } finally {
                    awaitUninterruptibly(canReturn);
                    returned.countDown();
                  }
                }

        List<String> receivedLines = new ArrayList<>();
        Consumer<String> callback = line -> {
            receivedLines.add(line);
            if (receivedLines.size() == 5000) {
                latch.countDown();
            }
        };

        InputStreamThread thread = new InputStreamThread(is, StandardCharsets.UTF_8, 100, callback);
        thread.start();

            exc -> {}
        );
        CountDownLatch latch = new CountDownLatch(shards.size());
        for (int i = 0; i < shards.size(); i++) {
            consumer.consumeResult(shards.get(i), () -> latch.countDown());
        }
        latch.await();
        SearchPhaseController.ReducedQueryPhase phase = consumer.reduce();
        executor.shutdownNow();
        return phase;
    }

    clientListener.assertOpen()
    clientListener.assertFailure(
      SocketTimeoutException::class.java,
      "sent ping but didn't receive pong within 500ms (after 0 successful ping/pongs)",
    )
    latch.countDown()
    val elapsedUntilFailure = System.nanoTime() - openAtNanos
    assertThat(TimeUnit.NANOSECONDS.toMillis(elapsedUntilFailure).toDouble())
      .isCloseTo(1000.0, 250.0)
  }

    Future<?> first = blocked.submit(() -> executor.execute(Runnables.doNothing()));
    future.get(10, SECONDS);
    assertThrows(RejectedExecutionException.class, () -> executor.execute(Runnables.doNothing()));
    latch.countDown();
    ExecutionException expected =
        assertThrows(ExecutionException.class, () -> first.get(10, SECONDS));
    assertThat(expected).hasCauseThat().isInstanceOf(RejectedExecutionException.class);
  }

        });
        writerThread.start();

        // Give threads a moment to reach the latch
        Thread.sleep(50);

        // Release all threads to start simultaneously
        startLatch.countDown();

        // Wait for writer thread to complete
        writerThread.join(1000);

        // Wait for all reader threads to complete
        for (Thread thread : readerThreads) {

    val latch = CountDownLatch(1)
    Thread {
      sleep(delayMillis)
      if (cancelMode == CANCEL) {
        call.cancel()
      } else {
        threadToCancel!!.interrupt()
      }
      latch.countDown()
    }.apply { start() }
    return latch
  }

  companion object {
    // The size of the socket buffers in bytes.
    private const val SOCKET_BUFFER_SIZE = 256 * 1024
  }
}

                execJob.jvmOptions("-Xmx" + i + "m");
            }
            latch.countDown();
        });

        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 100; i++) {
                execJob.jvmOptions("-Xms" + i + "m");
            }
            latch.countDown();
        });

        t1.start();
        t2.start();

    Thread thread =
        new Thread() {
          @Override
          public void run() {
            try {
              cacheRef.get().getUnchecked(3);
            } finally {
              doneSignal.countDown();
            }
          }
        };
    thread.setUncaughtExceptionHandler((t, throwable) -> {});
    thread.start();

    boolean done = doneSignal.await(1, SECONDS);
    if (!done) {