From 636e8b99226703d643cc4b30e5c30a64ce830434 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9s=20de=20la=20Pe=C3=B1a?= Date: Thu, 24 Jun 2021 14:00:46 +0100 Subject: [PATCH] Fix flaky ClientResourceLimitsTest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit patch by Andrés de la Peña, Berenguer Blasi and Caleb Rackliffe; reviewed by Andrés de la Peña, Berenguer Blasi and Caleb Rackliffe for CASSANDRA-16758 Co-authored-by: Andrés de la Peña Co-authored-by: Berenguer Blasi Co-authored-by: Caleb Rackliffe --- .../org/apache/cassandra/cql3/CQLTester.java | 28 +++++++++++---- .../transport/ClientResourceLimitsTest.java | 35 ++++++++++++------- 2 files changed, 44 insertions(+), 19 deletions(-) diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index 2f88f7c5dc..59968fe51b 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -406,6 +406,12 @@ public abstract class CQLTester return allArgs; } + protected static void requireNetworkWithoutDriver() + { + startServices(); + startServer(server -> {}); + } + // lazy initialization for all tests that require Java Driver protected static void requireNetwork() throws ConfigurationException { @@ -418,11 +424,16 @@ public abstract class CQLTester if (server != null) return; + startServices(); + initializeNetwork(decorator, null); + } + + private static void startServices() + { SystemKeyspace.finishStartup(); VirtualKeyspaceRegistry.instance.register(VirtualSchemaKeyspace.instance); StorageService.instance.initServer(); SchemaLoader.startGossiper(); - initializeNetwork(decorator, null); } protected static void reinitializeNetwork() @@ -451,11 +462,7 @@ public abstract class CQLTester private static void initializeNetwork(Consumer decorator, Consumer clusterConfigurator) { - Server.Builder serverBuilder = new Server.Builder().withHost(nativeAddr).withPort(nativePort); - decorator.accept(serverBuilder); - server = serverBuilder.build(); - ClientMetrics.instance.init(Collections.singleton(server)); - server.start(); + startServer(decorator); for (ProtocolVersion version : PROTOCOL_VERSIONS) { @@ -491,6 +498,15 @@ public abstract class CQLTester } } + private static void startServer(Consumer decorator) + { + Server.Builder serverBuilder = new Server.Builder().withHost(nativeAddr).withPort(nativePort); + decorator.accept(serverBuilder); + server = serverBuilder.build(); + ClientMetrics.instance.init(Collections.singleton(server)); + server.start(); + } + protected void dropPerTestKeyspace() throws Throwable { execute(String.format("DROP KEYSPACE IF EXISTS %s", KEYSPACE_PER_TEST)); diff --git a/test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java b/test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java index c52f43e272..9cc900aa50 100644 --- a/test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java +++ b/test/unit/org/apache/cassandra/transport/ClientResourceLimitsTest.java @@ -44,8 +44,10 @@ import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.transport.messages.QueryMessage; import org.apache.cassandra.utils.FBUtilities; +import org.awaitility.Awaitility; import static org.apache.cassandra.Util.spinAssertEquals; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; @@ -73,7 +75,9 @@ public class ClientResourceLimitsTest extends CQLTester DatabaseDescriptor.setNativeTransportReceiveQueueCapacityInBytes(1); DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytesPerIp(LOW_LIMIT); DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytes(LOW_LIMIT); - requireNetwork(); + + // The driver control connections would send queries that might interfere with the tests. + requireNetworkWithoutDriver(); } @AfterClass @@ -139,18 +143,18 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testQueryExecutionWithThrowOnOverload() throws Throwable + public void testQueryExecutionWithThrowOnOverload() { testQueryExecution(true); } @Test - public void testQueryExecutionWithoutThrowOnOverload() throws Throwable + public void testQueryExecutionWithoutThrowOnOverload() { testQueryExecution(false); } - private void testQueryExecution(boolean throwOnOverload) throws Throwable + private void testQueryExecution(boolean throwOnOverload) { try (SimpleClient client = client(throwOnOverload)) { @@ -239,7 +243,7 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testOverloadedExceptionWhenGlobalLimitExceeded() throws Throwable + public void testOverloadedExceptionWhenGlobalLimitExceeded() { // Bump the per-endpoint limit to make sure we exhaust the global ClientResourceLimits.setEndpointLimit(HIGH_LIMIT); @@ -247,7 +251,7 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testOverloadedExceptionWhenEndpointLimitExceeded() throws Throwable + public void testOverloadedExceptionWhenEndpointLimitExceeded() { // Make sure we can only exceed the per-endpoint limit ClientResourceLimits.setGlobalLimit(HIGH_LIMIT); @@ -255,7 +259,7 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testOverloadedExceptionWhenGlobalLimitByMultiFrameMessage() throws Throwable + public void testOverloadedExceptionWhenGlobalLimitByMultiFrameMessage() { // Bump the per-endpoint limit to make sure we exhaust the global ClientResourceLimits.setEndpointLimit(HIGH_LIMIT); @@ -263,7 +267,7 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testOverloadedExceptionWhenEndpointLimitByMultiFrameMessage() throws Throwable + public void testOverloadedExceptionWhenEndpointLimitByMultiFrameMessage() { // Make sure we can only exceed the per-endpoint limit ClientResourceLimits.setGlobalLimit(HIGH_LIMIT); @@ -300,7 +304,7 @@ public class ClientResourceLimitsTest extends CQLTester return new QueryMessage(query.toString(), V5_DEFAULT_OPTIONS); } - @SuppressWarnings("unchecked") + @SuppressWarnings({"rawtypes", "unchecked"}) private Gauge getPausedConnectionsGauge() { String metricName = "org.apache.cassandra.metrics.Client.PausedConnections"; @@ -316,6 +320,12 @@ public class ClientResourceLimitsTest extends CQLTester { try (SimpleClient client = client(true)) { + // wait for the completion of the intial messages created by the client connection + Awaitility.await() + .pollDelay(1, TimeUnit.SECONDS) + .atMost(30, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, ClientResourceLimits.getCurrentGlobalUsage())); + CyclicBarrier barrier = new CyclicBarrier(2); String table = createTableName(); @@ -345,8 +355,6 @@ public class ClientResourceLimitsTest extends CQLTester final QueryMessage queryMessage = new QueryMessage(String.format("SELECT * FROM %s.%s", table, table), V5_DEFAULT_OPTIONS); - - Assert.assertEquals(0L, ClientResourceLimits.getCurrentGlobalUsage()); try { Thread tester = new Thread(() -> client.execute(queryMessage)); @@ -355,7 +363,8 @@ public class ClientResourceLimitsTest extends CQLTester // block until query in progress barrier.await(30, TimeUnit.SECONDS); assertTrue(ClientResourceLimits.getCurrentGlobalUsage() > 0); - } finally + } + finally { // notify query thread that metric has been checked. This will also throw TimeoutException if both // the query threads barriers are not reached @@ -365,7 +374,7 @@ public class ClientResourceLimitsTest extends CQLTester } @Test - public void testChangingLimitsAtRuntime() throws Throwable + public void testChangingLimitsAtRuntime() { SimpleClient client = client(true); try