From 2d82d86e284633455e8b41e724eee5dd4324a17a Mon Sep 17 00:00:00 2001 From: Maxim Muzafarov Date: Sun, 12 Jul 2026 19:51:15 +0200 Subject: [PATCH] isolate backpressure for management and regular connections --- .../cassandra/transport/Dispatcher.java | 3 +- .../transport/PipelineConfigurator.java | 2 +- .../transport/QueueBackpressure.java | 35 ++++++++--- .../MessageManagementDispatcherTest.java | 61 +++++++++++++++++++ .../transport/QueueBackpressureTest.java | 21 +++++++ 5 files changed, 112 insertions(+), 10 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 762b653caa..f16d2b7b03 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -540,7 +540,8 @@ public class Dispatcher implements CQLMessageHandler.MessageConsumer 0; - QueueBackpressure DEFAULT = new QueueBackpressure() - { - private final AtomicReference state = new AtomicReference<>(noBackpressure(() -> DatabaseDescriptor.getNativeTransportMinBackoffOnQueueOverload(TimeUnit.NANOSECONDS), - () -> DatabaseDescriptor.getNativeTransportMaxBackoffOnQueueOverload(TimeUnit.NANOSECONDS))); + /** Shared incident state for connections served by {@link Dispatcher#requestExecutor}. */ + QueueBackpressure DEFAULT = newDefault(); - public long markAndGetDelay(TimeUnit timeUnit) + /** + * Shared incident state for management connections ({@link Dispatcher#managementExecutor}). Kept separate + * from {@link #DEFAULT} because incident severity describes a single executor's queue: an overload streak + * on the client transport must not inflate delays applied to management connections, and vice versa. + */ + QueueBackpressure MANAGEMENT_DEFAULT = newDefault(); + + static QueueBackpressure newDefault() + { + return newDefault(() -> DatabaseDescriptor.getNativeTransportMinBackoffOnQueueOverload(TimeUnit.NANOSECONDS), + () -> DatabaseDescriptor.getNativeTransportMaxBackoffOnQueueOverload(TimeUnit.NANOSECONDS)); + } + + @VisibleForTesting + static QueueBackpressure newDefault(LongSupplier minDelayNanos, LongSupplier maxDelayNanos) + { + return new QueueBackpressure() { - return state.updateAndGet(Incident::mark).delay(timeUnit); - } - }; + private final AtomicReference state = new AtomicReference<>(noBackpressure(minDelayNanos, maxDelayNanos)); + + public long markAndGetDelay(TimeUnit timeUnit) + { + return state.updateAndGet(Incident::mark).delay(timeUnit); + } + }; + } long markAndGetDelay(TimeUnit timeUnit); diff --git a/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java b/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java index 5856d8cf2b..1afb700dca 100644 --- a/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java +++ b/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java @@ -431,6 +431,67 @@ public class MessageManagementDispatcherTest managementDispatcher::isDone); } + /** + * Queue-time backpressure must be keyed to the executor serving the connection: a backlog on the + * management pool should trigger backpressure for management connections only, and must stay + * invisible to regular client connections (and vice versa). + */ + @Test + public void testHasQueueCapacityTracksOwnExecutor() throws Exception + { + Dispatcher managementDispatcher = new ManagementTestDispatcher(true); + double thresholdBefore = DatabaseDescriptor.getRawConfig().native_transport_queue_max_item_age_threshold; + DatabaseDescriptor.getRawConfig().native_transport_queue_max_item_age_threshold = 1e-9; + + int poolSize = Dispatcher.managementExecutor.getMaximumPoolSize(); + CountDownLatch saturated = new CountDownLatch(poolSize); + CountDownLatch release = new CountDownLatch(1); + CountDownLatch processed = new CountDownLatch(1); + try + { + assertTrue("Both drains should report capacity while idle", managementDispatcher.hasQueueCapacity()); + assertTrue(dispatch.hasQueueCapacity()); + + for (int i = 0; i < poolSize; i++) + { + Dispatcher.managementExecutor.submit(() -> { + saturated.countDown(); + Uninterruptibles.awaitUninterruptibly(release); + }); + } + assertTrue(saturated.await(10, TimeUnit.SECONDS)); + + Dispatcher queuedProcessor = new ManagementTestDispatcher(true) + { + @Override + void processRequest(Channel channel, + Message.Request request, + FlushItemConverter forFlusher, + ClientResourceLimits.Overload backpressure, + RequestTime requestTime) + { + processed.countDown(); + } + }; + Message.Request request = createManagementRequest(Message.Type.STARTUP); + queuedProcessor.dispatch(request.connection().channel(), request, (channel, req, response) -> null, + ClientResourceLimits.Overload.NONE); + + awaitTrue("A queued management request should trigger management backpressure", + () -> !managementDispatcher.hasQueueCapacity()); + assertTrue("A management backlog should be invisible to the regular drain", + dispatch.hasQueueCapacity()); + } + finally + { + release.countDown(); + DatabaseDescriptor.getRawConfig().native_transport_queue_max_item_age_threshold = thresholdBefore; + } + assertTrue(processed.await(10, TimeUnit.SECONDS)); + awaitTrue("The management pool should drain once its tasks finish", + managementDispatcher::isDone); + } + private static void awaitTrue(String message, Callable condition) throws Exception { long timeout = Clock.Global.currentTimeMillis(); diff --git a/test/unit/org/apache/cassandra/transport/QueueBackpressureTest.java b/test/unit/org/apache/cassandra/transport/QueueBackpressureTest.java index b76739f1a5..b3fca67bfd 100644 --- a/test/unit/org/apache/cassandra/transport/QueueBackpressureTest.java +++ b/test/unit/org/apache/cassandra/transport/QueueBackpressureTest.java @@ -57,4 +57,25 @@ public class QueueBackpressureTest backpressure = backpressure.mark(backpressure.appliedAt() + TimeUnit.MILLISECONDS.toNanos(1001)); Assert.assertEquals(backpressure.minDelayNanos(), backpressure.delay(TimeUnit.NANOSECONDS)); } + + /** + * The client and management transports use separate {@link QueueBackpressure} instances, so an overload + * incident on one must not inflate the delay applied by the other. + */ + @Test + public void testIncidentStateIsPerInstance() + { + long minDelayNanos = TimeUnit.MILLISECONDS.toNanos(10); + long maxDelayNanos = TimeUnit.MILLISECONDS.toNanos(100); + QueueBackpressure client = QueueBackpressure.newDefault(() -> minDelayNanos, () -> maxDelayNanos); + QueueBackpressure management = QueueBackpressure.newDefault(() -> minDelayNanos, () -> maxDelayNanos); + + long clientDelay = 0; + for (int i = 0; i < 50; i++) + clientDelay = client.markAndGetDelay(TimeUnit.NANOSECONDS); + Assert.assertTrue("Client incident should have ramped past the minimum delay", clientDelay > minDelayNanos); + + Assert.assertEquals("A first management overload must start its own incident at the minimum delay", + minDelayNanos, management.markAndGetDelay(TimeUnit.NANOSECONDS)); + } }