public void maybeBlockOn(String block) {
            if (block.equals(this.block)) {
                log.info("Will block on {}", block);
                CountDownLatch blockLatch;
                synchronized (Block.class) {
                    assertNotNull(
                            awaitBlockLatch,
                            "Block was reset prematurely"
                    );
                    awaitBlockLatch.countDown();
                    blockLatch = newBlockLatch();
                    BLOCKED_THREADS.add(Thread.currentThread());
                }
                while (true) {
                    try {
                        blockLatch.await();
                        log.debug("Instructed to stop blocking; will resume normal execution");
                        return;
                    } catch (InterruptedException e) {
                        log.debug("Interrupted while blocking; will continue blocking until instructed to stop");
                    }
                }
            } else {
                log.debug("Will not block on {}", block);
            }
        }
{@link Thread#join(long millis) Await} the termination of all threads that have been
intentionally blocked either since the last invocation of this method or, if this method
has never been invoked, all threads that have ever been blocked.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/BlockingConnectorTest.java