isolate backpressure for management and regular connections

src/java/org/apache/cassandra/transport/Dispatcher.java

@@ -540,7 +540,8 @@ public class Dispatcher implements CQLMessageHandler.MessageConsumer<Message.Req
         if (threshold <= 0)
             return true;
 
-        return requestExecutor.oldestTaskQueueTime() < (DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS) * threshold);
+        LocalAwareExecutorPlus executor = isManagementDispatcher ? managementExecutor : requestExecutor;
+        return executor.oldestTaskQueueTime() < (DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS) * threshold);
     }
 
     /**

src/java/org/apache/cassandra/transport/PipelineConfigurator.java

@@ -129,7 +129,7 @@ public class PipelineConfigurator
         this.keepAlive           = keepAlive;
         this.tlsEncryptionPolicy = encryptionPolicy;
         this.dispatcher          = dispatcher;
-        this.queueBackpressure   = QueueBackpressure.DEFAULT;
+        this.queueBackpressure   = isManagementConnection ? QueueBackpressure.MANAGEMENT_DEFAULT : QueueBackpressure.DEFAULT;
         this.isManagementConnection = isManagementConnection;
     }
 

src/java/org/apache/cassandra/transport/QueueBackpressure.java

@@ -48,16 +48,35 @@ public interface QueueBackpressure
 {
     QueueBackpressure NO_OP = timeUnit -> 0;
 
-    QueueBackpressure DEFAULT = new QueueBackpressure()
-    {
-        private final AtomicReference<Incident> 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<Incident> state = new AtomicReference<>(noBackpressure(minDelayNanos, maxDelayNanos));
+
+            public long markAndGetDelay(TimeUnit timeUnit)
+            {
+                return state.updateAndGet(Incident::mark).delay(timeUnit);
+            }
+        };
+    }
 
     long markAndGetDelay(TimeUnit timeUnit);
 

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<Boolean> condition) throws Exception
     {
         long timeout = Clock.Global.currentTimeMillis();

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));
+    }
 }