From c1d57e0f9d35b1fb434683f316f54b59a59817c1 Mon Sep 17 00:00:00 2001 From: PapaCharlie Date: Wed, 8 Mar 2023 09:52:21 -0800 Subject: [PATCH 1/3] Fix a race condition in WatcherCleanerTest.testDeadWatcherMetrics Because the metrics were updated _after_ the listener is invoked, the listener does not always see the fresh metric value. This fixes it so that the test waits for the value to become what we expect. --- .../server/watch/WatcherCleanerTest.java | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java index 17e44eb9b01..c88a0fdf95c 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.HashSet; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -31,6 +32,7 @@ import org.apache.zookeeper.common.Time; import org.apache.zookeeper.metrics.MetricsUtils; import org.apache.zookeeper.server.ServerMetrics; +import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -139,8 +141,10 @@ public void testMaxInProcessingDeadWatchers() { assertTrue(listener.wait(5000)); } - @Test - public void testDeadWatcherMetrics() { + // There used to be a race condition surrounding this test which was reproducible by running the test multiple + // times. This test is kept as repeated to flag if the race condition reappears. + @RepeatedTest(5) + public void testDeadWatcherMetrics() throws InterruptedException { ServerMetrics.getMetrics().resetAll(); MyDeadWatcherListener listener = new MyDeadWatcherListener(); WatcherCleaner cleaner = new WatcherCleaner(listener, 1, 1, 1, 1); @@ -158,7 +162,9 @@ public void testDeadWatcherMetrics() { Map values = MetricsUtils.currentServerMetrics(); assertThat("Adding dead watcher should be stalled twice", (Long) values.get("add_dead_watcher_stall_time"), greaterThan(0L)); assertEquals(3L, values.get("dead_watchers_queued"), "Total dead watchers added to the queue should be 3"); - assertEquals(3L, values.get("dead_watchers_cleared"), "Total dead watchers cleared should be 3"); + // This metric is updated _after_ the dead watcher listener is invoked, so it is not always immediately visible, + // hence the wait. + waitForMetricValue("dead_watchers_cleared", 3L, 5_000); assertEquals(3L, values.get("cnt_dead_watchers_cleaner_latency")); @@ -171,4 +177,15 @@ public void testDeadWatcherMetrics() { assertEquals(20D, ((Long) values.get("p99_dead_watchers_cleaner_latency")).doubleValue(), 20); } + /** + * Waits in a loop for the given metric to have the required value. If the given timeout is reached, the test fails. + */ + private static void waitForMetricValue(String metricName, Object expected, long timeoutMs) throws InterruptedException { + long start = Time.currentElapsedTime(); + while (!Objects.equals(MetricsUtils.currentServerMetrics().get(metricName), expected)) { + Thread.sleep(100); + assertFalse(Time.currentElapsedTime() - start > timeoutMs, + "Metric value was not updated in " + timeoutMs + "ms!"); + } + } } From 1052ee9856ae2339661d906cc08473da098e664e Mon Sep 17 00:00:00 2001 From: PapaCharlie Date: Wed, 8 Mar 2023 09:52:21 -0800 Subject: [PATCH 2/3] Leverage an existing method and refactor the rest of the code to match Since there was an existing waitFor method in ZKTestCase, along with an existing implementation of a waitForMetric LearnerMetricsTest, this commit moves waitForMetric to ZKTestCase and refactors the metric-related usages of waitFor. --- .../java/org/apache/zookeeper/ZKTestCase.java | 42 +++++++++++++++- .../server/RequestThrottlerTest.java | 16 ++---- .../FileTxnSnapLogMetricsTest.java | 3 +- .../server/quorum/LearnerMetricsTest.java | 15 ------ .../server/watch/WatcherCleanerTest.java | 49 ++++++------------- 5 files changed, 62 insertions(+), 63 deletions(-) diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java index 8d9430e1cf6..b29deedb0bc 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java @@ -22,7 +22,12 @@ import static org.junit.jupiter.api.Assertions.fail; import java.io.File; import java.time.Instant; +import org.apache.zookeeper.metrics.MetricsUtils; import org.apache.zookeeper.util.ServiceUtils; +import org.hamcrest.CustomMatcher; +import org.hamcrest.Description; +import org.hamcrest.Matcher; +import org.hamcrest.StringDescription; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; @@ -40,6 +45,7 @@ public class ZKTestCase { protected static final File testBaseDir = new File(System.getProperty("build.test.dir", "build")); private static final Logger LOG = LoggerFactory.getLogger(ZKTestCase.class); + public static final int DEFAULT_METRIC_TIMEOUT = 30; static { // Disable System.exit in tests. @@ -103,7 +109,7 @@ public interface WaitForCondition { * @param timeout timeout in seconds * @throws InterruptedException */ - public void waitFor(String msg, WaitForCondition condition, int timeout) throws InterruptedException { + public static void waitFor(String msg, WaitForCondition condition, int timeout) throws InterruptedException { final Instant deadline = Instant.now().plusSeconds(timeout); while (Instant.now().isBefore(deadline)) { if (condition.evaluate()) { @@ -114,4 +120,36 @@ public void waitFor(String msg, WaitForCondition condition, int timeout) throws fail(msg); } -} + public static void waitForMetric(String metricKey, Matcher matcher) throws InterruptedException { + waitForMetric(metricKey, matcher, DEFAULT_METRIC_TIMEOUT); + } + + public static void waitForMetric(String metricKey, Matcher matcher, int timeoutInSeconds) throws InterruptedException { + String errorMessage = String.format("metric \"%s\" failed to match after %d seconds", + metricKey, timeoutInSeconds); + waitFor(errorMessage, () -> { + @SuppressWarnings("unchecked") + T actual = (T) MetricsUtils.currentServerMetrics().get(metricKey); + if (!matcher.matches(actual)) { + Description description = new StringDescription(); + matcher.describeMismatch(actual, description); + LOG.info("match failed for metric {}: {}", metricKey, description); + return false; + } + return true; + }, timeoutInSeconds); + } + + /** + * Functionally identical to {@link org.hamcrest.Matchers#closeTo} except that it accepts all numerical types + * instead of failing if the value is not a {@link Double}. + */ + public static Matcher closeTo(double operand, double error) { + return new CustomMatcher(String.format("A number within %s of %s", error, operand)) { + @Override + public boolean matches(Object actual) { + return Math.abs(operand - ((Number) actual).doubleValue()) <= error; + } + }; + } +} \ No newline at end of file diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/RequestThrottlerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/RequestThrottlerTest.java index 15259207599..088f80c4857 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/RequestThrottlerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/RequestThrottlerTest.java @@ -19,6 +19,8 @@ package org.apache.zookeeper.server; import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.File; @@ -75,10 +77,6 @@ public class RequestThrottlerTest extends ZKTestCase { ZooKeeper zk = null; int connectionLossCount = 0; - private long getCounterMetric(String name) { - return (long) MetricsUtils.currentServerMetrics().get(name); - } - @BeforeEach public void setup() throws Exception { // start a server and create a client @@ -222,11 +220,8 @@ public void testRequestThrottler() throws Exception { submitted.await(5, TimeUnit.SECONDS); // but only two requests can get into the pipeline because of the throttler - WaitForCondition requestQueued = () -> getCounterMetric("prep_processor_request_queued") == 2; - waitFor("request not queued", requestQueued, 5); - - WaitForCondition throttleWait = () -> getCounterMetric("request_throttle_wait_count") >= 1; - waitFor("no throttle wait", throttleWait, 5); + waitForMetric("prep_processor_request_queued", is(2L)); + waitForMetric("request_throttle_wait_count", greaterThanOrEqualTo(1L)); // let the requests go through the pipeline and the throttler will be waken up to allow more requests // to enter the pipeline @@ -387,8 +382,7 @@ public void testGlobalOutstandingRequestThrottlingWithRequestThrottlerDisabled() // be GLOBAL_OUTSTANDING_LIMIT + 2. // // But due to leak of consistent view of number of outstanding requests, the number could be larger. - WaitForCondition requestQueued = () -> getCounterMetric("prep_processor_request_queued") >= Integer.parseInt(GLOBAL_OUTSTANDING_LIMIT) + 2; - waitFor("no enough requests queued", requestQueued, 5); + waitForMetric("prep_processor_request_queued", greaterThanOrEqualTo(Long.parseLong(GLOBAL_OUTSTANDING_LIMIT) + 2)); resumeProcess.countDown(); } catch (Exception e) { diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnSnapLogMetricsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnSnapLogMetricsTest.java index 65648fefce4..1a569e4d47e 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnSnapLogMetricsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/persistence/FileTxnSnapLogMetricsTest.java @@ -79,8 +79,7 @@ public void testFileTxnSnapLogMetrics() throws Exception { } // It is possible that above writes will trigger more than one snapshot due to randomization. - WaitForCondition newSnapshot = () -> (long) MetricsUtils.currentServerMetrics().get("cnt_snapshottime") >= 2L; - waitFor("no snapshot in 10s", newSnapshot, 10); + waitForMetric("cnt_snapshottime", greaterThanOrEqualTo(2L), 10); // Pauses snapshot and logs more txns. cnxnFactory.getZooKeeperServer().getTxnLogFactory().snapLog.close(); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java index 5df14600a15..aa3ba3622ad 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/quorum/LearnerMetricsTest.java @@ -26,10 +26,8 @@ import org.apache.zookeeper.PortAssignment; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.ZooKeeper; -import org.apache.zookeeper.metrics.MetricsUtils; import org.apache.zookeeper.server.ServerMetrics; import org.apache.zookeeper.test.ClientBase; -import org.hamcrest.Matcher; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeAll; @@ -38,7 +36,6 @@ public class LearnerMetricsTest extends QuorumPeerTestBase { - private static final int TIMEOUT_SECONDS = 30; private static final int SERVER_COUNT = 4; // 1 observer, 3 participants private final QuorumPeerTestBase.MainThread[] mt = new QuorumPeerTestBase.MainThread[SERVER_COUNT]; private ZooKeeper zk_client; @@ -113,18 +110,6 @@ public void testLearnerMetricsTest(boolean asyncSending) throws Exception { waitForMetric("min_commit_propagation_latency", greaterThanOrEqualTo(0L)); } - private void waitForMetric(final String metricKey, final Matcher matcher) throws InterruptedException { - final String errorMessage = String.format("unable to match on metric: %s", metricKey); - waitFor(errorMessage, () -> { - long actual = (long) MetricsUtils.currentServerMetrics().get(metricKey); - if (!matcher.matches(actual)) { - LOG.info("match failed on {}, actual value: {}", metricKey, actual); - return false; - } - return true; - }, TIMEOUT_SECONDS); - } - @AfterEach public void tearDown() throws Exception { zk_client.close(); diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java index c88a0fdf95c..3320f1a247c 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/watch/WatcherCleanerTest.java @@ -17,14 +17,13 @@ package org.apache.zookeeper.server.watch; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.number.OrderingComparison.greaterThan; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.is; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.HashSet; import java.util.Map; -import java.util.Objects; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -32,7 +31,6 @@ import org.apache.zookeeper.common.Time; import org.apache.zookeeper.metrics.MetricsUtils; import org.apache.zookeeper.server.ServerMetrics; -import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -141,9 +139,7 @@ public void testMaxInProcessingDeadWatchers() { assertTrue(listener.wait(5000)); } - // There used to be a race condition surrounding this test which was reproducible by running the test multiple - // times. This test is kept as repeated to flag if the race condition reappears. - @RepeatedTest(5) + @Test public void testDeadWatcherMetrics() throws InterruptedException { ServerMetrics.getMetrics().resetAll(); MyDeadWatcherListener listener = new MyDeadWatcherListener(); @@ -160,32 +156,19 @@ public void testDeadWatcherMetrics() throws InterruptedException { assertTrue(listener.wait(5000)); Map values = MetricsUtils.currentServerMetrics(); - assertThat("Adding dead watcher should be stalled twice", (Long) values.get("add_dead_watcher_stall_time"), greaterThan(0L)); - assertEquals(3L, values.get("dead_watchers_queued"), "Total dead watchers added to the queue should be 3"); - // This metric is updated _after_ the dead watcher listener is invoked, so it is not always immediately visible, - // hence the wait. - waitForMetricValue("dead_watchers_cleared", 3L, 5_000); - - assertEquals(3L, values.get("cnt_dead_watchers_cleaner_latency")); - - //Each latency should be a little over 20 ms, allow 20 ms deviation - assertEquals(20D, (Double) values.get("avg_dead_watchers_cleaner_latency"), 20); - assertEquals(20D, ((Long) values.get("min_dead_watchers_cleaner_latency")).doubleValue(), 20); - assertEquals(20D, ((Long) values.get("max_dead_watchers_cleaner_latency")).doubleValue(), 20); - assertEquals(20D, ((Long) values.get("p50_dead_watchers_cleaner_latency")).doubleValue(), 20); - assertEquals(20D, ((Long) values.get("p95_dead_watchers_cleaner_latency")).doubleValue(), 20); - assertEquals(20D, ((Long) values.get("p99_dead_watchers_cleaner_latency")).doubleValue(), 20); + // Adding dead watcher should be stalled twice + waitForMetric("add_dead_watcher_stall_time", greaterThan(0L)); + waitForMetric("dead_watchers_queued", is(3L)); + waitForMetric("dead_watchers_cleared", is(3L)); + waitForMetric("cnt_dead_watchers_cleaner_latency", is(3L)); + + //Each latency should be a little over 20 ms, allow 5 ms deviation + waitForMetric("avg_dead_watchers_cleaner_latency", closeTo(20, 5)); + waitForMetric("min_dead_watchers_cleaner_latency", closeTo(20, 5)); + waitForMetric("max_dead_watchers_cleaner_latency", closeTo(20, 5)); + waitForMetric("p50_dead_watchers_cleaner_latency", closeTo(20, 5)); + waitForMetric("p95_dead_watchers_cleaner_latency", closeTo(20, 5)); + waitForMetric("p99_dead_watchers_cleaner_latency", closeTo(20, 5)); } - /** - * Waits in a loop for the given metric to have the required value. If the given timeout is reached, the test fails. - */ - private static void waitForMetricValue(String metricName, Object expected, long timeoutMs) throws InterruptedException { - long start = Time.currentElapsedTime(); - while (!Objects.equals(MetricsUtils.currentServerMetrics().get(metricName), expected)) { - Thread.sleep(100); - assertFalse(Time.currentElapsedTime() - start > timeoutMs, - "Metric value was not updated in " + timeoutMs + "ms!"); - } - } } From c0eb6fc258c50e8a6e6588a122c292b833ba2092 Mon Sep 17 00:00:00 2001 From: PapaCharlie Date: Wed, 24 May 2023 13:15:11 -0400 Subject: [PATCH 3/3] Update log message to be less alarming when polling for metric value This message will log every 100ms as long as the metric does not match the expected value, and should be clear that it does not indicate a test failure (at least not yet) --- .../src/test/java/org/apache/zookeeper/ZKTestCase.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java index b29deedb0bc..5e62ce26832 100644 --- a/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java +++ b/zookeeper-server/src/test/java/org/apache/zookeeper/ZKTestCase.java @@ -46,6 +46,7 @@ public class ZKTestCase { protected static final File testBaseDir = new File(System.getProperty("build.test.dir", "build")); private static final Logger LOG = LoggerFactory.getLogger(ZKTestCase.class); public static final int DEFAULT_METRIC_TIMEOUT = 30; + public static final int METRIC_POLL_INTERVAL_MS = 100; static { // Disable System.exit in tests. @@ -115,7 +116,7 @@ public static void waitFor(String msg, WaitForCondition condition, int timeout) if (condition.evaluate()) { return; } - Thread.sleep(100); + Thread.sleep(METRIC_POLL_INTERVAL_MS); } fail(msg); } @@ -133,7 +134,8 @@ public static void waitForMetric(String metricKey, Matcher matcher, int t if (!matcher.matches(actual)) { Description description = new StringDescription(); matcher.describeMismatch(actual, description); - LOG.info("match failed for metric {}: {}", metricKey, description); + LOG.info("metric {} does not yet match expected value: {}. Checking again in {}ms", + metricKey, description, METRIC_POLL_INTERVAL_MS); return false; } return true;