public static void retryOnExceptionWithTimeout(final long timeoutMs,
                                                   final long pollIntervalMs,
                                                   final ValuelessCallable runnable) throws InterruptedException {
        final long expectedEnd = System.currentTimeMillis() + timeoutMs;

        while (true) {
            try {
                runnable.call();
                return;
            } catch (final NoRetryException e) {
                throw e;
            } catch (final AssertionError t) {
                if (expectedEnd <= System.currentTimeMillis()) {
                    throw t;
                }
            } catch (final Exception e) {
                if (expectedEnd <= System.currentTimeMillis()) {
                    throw new AssertionError(String.format("Assertion failed with an exception after %s ms", timeoutMs), e);
                }
            }
            Thread.sleep(Math.min(pollIntervalMs, timeoutMs));
        }
    }
Wait for the given runnable to complete successfully, i.e. throw now {@link Exception}s or
{@link AssertionError}s, or for the given timeout to expire. If the timeout expires then the
last exception or assertion failure will be thrown thus providing context for the failure.

@param timeoutMs the total time in milliseconds to wait for {@code runnable} to complete successfully.
@param pollIntervalMs the interval in milliseconds to wait between invoking {@code runnable}.
@param runnable the code to attempt to execute successfully.
@throws InterruptedException if the current thread is interrupted while waiting for {@code runnable} to complete successfully.
clients/src/test/java/org/apache/kafka/test/TestUtils.java