public void waitForMetrics(String namePrefix, final double expectedValue, Set<MetricType> metricTypes)
            throws InterruptedException {
        long maxAggregateWaitMs = 15000;
        long startMs = time.milliseconds();
        for (MetricType metricType : metricTypes) {
            long currentElapsedMs = time.milliseconds() - startMs;
            long thisMaxWaitMs = maxAggregateWaitMs - currentElapsedMs;
            String metricName = namePrefix + metricType.metricNameSuffix();
            if (expectedValue == 0.0) {
                double expected = expectedValue;
                if (metricType == MetricType.MAX || metricType == MetricType.AVG)
                    expected = Double.NaN;

                assertEquals(expected, metricValue(metricName), EPS, "Metric not updated " + metricName +
                    " expected:<" + expectedValue + "> but was:<" + metricValue(metricName) + ">");
            } else if (metricType == MetricType.TOTAL)
                TestUtils.waitForCondition(() -> Math.abs(metricValue(metricName) - expectedValue) <= EPS,
                        thisMaxWaitMs, () -> "Metric not updated " + metricName + " expected:<" + expectedValue
                                + "> but was:<" + metricValue(metricName) + ">");
            else
                TestUtils.waitForCondition(() -> metricValue(metricName) > 0.0, thisMaxWaitMs,
                    () -> "Metric not updated " + metricName + " expected:<a positive number> but was:<"
                                + metricValue(metricName) + ">");
        }
    }
Non-blocking EchoServer implementation that uses ChannelBuilder to create channels
with the configured security protocol.
clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java