CASSANDRA-21654: Coordinator node experiences read failures on ungraceful shutdown of replica

conf/cassandra.yaml

@@ -1116,6 +1116,16 @@ uuid_sstable_identifiers_enabled: false
 # Lowest acceptable value is 10 ms.
 # Min unit: ms
 read_request_timeout: 5000ms
+
+# What the coordinator does when a read message is dropped because the send queue for a
+# replica is full. This happens while a replica is unreachable and gossip has not yet marked
+# it down.
+# If true, the coordinator sends the read to another candidate replica, as a speculative
+# retry. The table must permit speculative retry, so a table with speculative_retry: NONE
+# keeps the behaviour below.
+# If false, the coordinator fails the read.
+# read_fallback_on_overloaded_connection: false
+
 # How long the coordinator should wait for seq or index scans to complete.
 # Lowest acceptable value is 10 ms.
 # Min unit: ms

doc/modules/cassandra/pages/operating/metrics.adoc

@@ -223,6 +223,10 @@ ongoing incremental repair
 |SpeculativeRetries |Counter |Number of times speculative retries were
 sent for this table.
 
+|OverloadSpeculativeRetries |Counter |Number of speculative retries that
+a read message dropped on an overloaded connection triggered. A subset of
+SpeculativeRetries.
+
 |SpeculativeFailedRetries |Counter |Number of speculative retries that
 failed to prevent a timeout
 

src/java/org/apache/cassandra/config/Config.java

@@ -132,6 +132,8 @@ public static Set<String> splitCommaDelimited(String src)
     @Replaces(oldName = "read_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound read_request_timeout = new DurationSpec.LongMillisecondsBound("5000ms");
 
+    public volatile boolean read_fallback_on_overloaded_connection = false;
+
     @Replaces(oldName = "range_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound range_request_timeout = new DurationSpec.LongMillisecondsBound("10000ms");
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1836,6 +1836,16 @@ public static void setReadRpcTimeout(long timeOutInMillis)
         conf.read_request_timeout = new DurationSpec.LongMillisecondsBound(timeOutInMillis);
     }
 
+    public static boolean getReadFallbackOnOverloadedConnection()
+    {
+        return conf.read_fallback_on_overloaded_connection;
+    }
+
+    public static void setReadFallbackOnOverloadedConnection(boolean enabled)
+    {
+        conf.read_fallback_on_overloaded_connection = enabled;
+    }
+
     public static long getRangeRpcTimeout(TimeUnit unit)
     {
         return conf.range_request_timeout.to(unit);

src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java

@@ -101,6 +101,8 @@ public class KeyspaceMetrics
     public final LatencyMetrics idealCLWriteLatency;
     /** Speculative retries **/
     public final Counter speculativeRetries;
+    /** The subset of speculative retries that a dropped read message triggered **/
+    public final Counter overloadSpeculativeRetries;
     /** Speculative retry occured but still timed out **/
     public final Counter speculativeFailedRetries;
     /** Needed to speculate, but didn't have enough replicas **/
@@ -238,6 +240,7 @@ public KeyspaceMetrics(final Keyspace ks)
         idealCLWriteLatency = createLatencyMetrics("IdealCLWrite");
 
         speculativeRetries = createKeyspaceCounter("SpeculativeRetries", metric -> metric.speculativeRetries.getCount());
+        overloadSpeculativeRetries = createKeyspaceCounter("OverloadSpeculativeRetries", metric -> metric.overloadSpeculativeRetries.getCount());
         speculativeFailedRetries = createKeyspaceCounter("SpeculativeFailedRetries", metric -> metric.speculativeFailedRetries.getCount());
         speculativeInsufficientReplicas = createKeyspaceCounter("SpeculativeInsufficientReplicas", metric -> metric.speculativeInsufficientReplicas.getCount());
         additionalWrites = createKeyspaceCounter("AdditionalWrites", metric -> metric.additionalWrites.getCount());

src/java/org/apache/cassandra/metrics/TableMetrics.java

@@ -221,6 +221,7 @@ public class TableMetrics
     private final MetricNameFactory aliasFactory;
 
     public final Counter speculativeRetries;
+    public final Counter overloadSpeculativeRetries;
     public final Counter speculativeFailedRetries;
     public final Counter speculativeInsufficientReplicas;
     public final Gauge<Long> speculativeSampleLatencyNanos;
@@ -823,6 +824,7 @@ public Long getValue()
             }
         });
         speculativeRetries = createTableCounter("SpeculativeRetries");
+        overloadSpeculativeRetries = createTableCounter("OverloadSpeculativeRetries");
         speculativeFailedRetries = createTableCounter("SpeculativeFailedRetries");
         speculativeInsufficientReplicas = createTableCounter("SpeculativeInsufficientReplicas");
         speculativeSampleLatencyNanos = createTableGauge("SpeculativeSampleLatencyNanos", () -> MICROSECONDS.toNanos(cfs.sampleReadLatencyMicros));

src/java/org/apache/cassandra/net/OutboundConnection.java

@@ -339,8 +339,10 @@ public void enqueue(Message message) throws ClosedChannelException
                 // this is an optimisation only; messages will be expired on ~100ms cycle, and by Delivery when it runs
                 if (queue.maybePruneExpired() && SUCCESS == acquireCapacity(canonicalSize))
                     break;
+                onOverloaded(message, INSUFFICIENT_ENDPOINT);
+                return;
             case INSUFFICIENT_GLOBAL:
-                onOverloaded(message);
+                onOverloaded(message, INSUFFICIENT_GLOBAL);
                 return;
         }
 
@@ -452,7 +454,7 @@ private void releaseCapacity(long count, long bytes)
         }
     }
 
-    private void onOverloaded(Message<?> message)
+    private void onOverloaded(Message<?> message, Outcome outcome)
     {
         overloadedCountUpdater.incrementAndGet(this);
 
@@ -463,7 +465,7 @@ private void onOverloaded(Message<?> message)
                           this, FBUtilities.prettyPrintMemory(canonicalSize),
                           readablePendingBytes, readableReserveEndpointUsing, readableReserveGlobalUsing);
 
-        callbacks.onOverloaded(message, template.to);
+        callbacks.onOverloaded(message, template.to, outcome);
     }
 
     /**

src/java/org/apache/cassandra/net/OutboundMessageCallbacks.java

@@ -18,11 +18,17 @@
 package org.apache.cassandra.net;
 
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.ResourceLimits.Outcome;
 
 interface OutboundMessageCallbacks
 {
-    /** A message was not enqueued to the link because too many messages are already waiting to send */
-    void onOverloaded(Message<?> message, InetAddressAndPort peer);
+    /**
+     * A message was not enqueued to the link because too many messages are already waiting to send.
+     *
+     * {@code outcome} is {@link Outcome#INSUFFICIENT_ENDPOINT} when only this peer is out of capacity,
+     * and {@link Outcome#INSUFFICIENT_GLOBAL} when the node-wide reserve is exhausted.
+     */
+    void onOverloaded(Message<?> message, InetAddressAndPort peer, Outcome outcome);
 
     /** A message was not serialized to a frame because it had expired */
     void onExpired(Message<?> message, InetAddressAndPort peer);

src/java/org/apache/cassandra/net/RequestCallback.java

@@ -19,6 +19,7 @@
 
 import org.apache.cassandra.exceptions.RequestFailureReason;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.ResourceLimits.Outcome;
 
 /**
  * implementors of {@link RequestCallback} need to make sure that any public methods
@@ -40,6 +41,31 @@ default void onFailure(InetAddressAndPort from, RequestFailureReason failureReas
     {
     }
 
+    /**
+     * Called when the outbound connection to the peer was overloaded i.e., the request was dropped
+     * before it was sent.
+     *
+     * This method runs on the internal response stage, unless the callback asks for it to run inline
+     * with {@link #invokeOnOverloadedInline()}.
+     *
+     * @param outcome which capacity limit the request was refused by
+     */
+    default void onOverloaded(InetAddressAndPort from, Outcome outcome)
+    {
+        onFailure(from, RequestFailureReason.TIMEOUT);
+    }
+
+    /**
+     * Dictates whether the overloaded callback should run on the INTERNAL_RESPONSE threadpool or on the
+     * calling thread.
+     *
+     * @return true if {@link #onOverloaded} must run on the thread that dropped the request
+     */
+    default boolean invokeOnOverloadedInline()
+    {
+        return false;
+    }
+
     /**
      * Returns true if the callback handles failure reporting - in which case the remove host will be asked to
      * report failures to us in the event of a problem processing the request.

src/java/org/apache/cassandra/net/RequestCallbacks.java

@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.net;
 
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -36,6 +37,7 @@
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.metrics.InternodeOutboundMetrics;
+import org.apache.cassandra.net.ResourceLimits.Outcome;
 import org.apache.cassandra.service.AbstractWriteResponseHandler;
 
 import static java.lang.String.format;
@@ -128,6 +130,12 @@ private void removeAndExpire(long id, InetAddressAndPort peer)
         if (null != ci) onExpired(ci);
     }
 
+    private void removeAndOverload(long id, InetAddressAndPort peer, Outcome outcome)
+    {
+        CallbackInfo ci = remove(id, peer);
+        if (null != ci) onOverloaded(ci, outcome);
+    }
+
     private void expire()
     {
         long start = preciseTime.now();
@@ -164,6 +172,18 @@ private void onExpired(CallbackInfo info)
             INTERNAL_RESPONSE.submit(() -> info.callback.onFailure(info.peer, RequestFailureReason.TIMEOUT));
     }
 
+
+    private void onOverloaded(CallbackInfo info, Outcome outcome)
+    {
+        if (!info.invokeOnFailure())
+            return;
+
+        if (info.callback.invokeOnOverloadedInline())
+            info.callback.onOverloaded(info.peer, outcome);
+        else
+            INTERNAL_RESPONSE.submit(() -> info.callback.onOverloaded(info.peer, outcome));
+    }
+
     void shutdownNow(boolean expireCallbacks)
     {
         executor.shutdownNow();
@@ -277,9 +297,9 @@ public String toString()
     }
 
     @Override
-    public void onOverloaded(Message<?> message, InetAddressAndPort peer)
+    public void onOverloaded(Message<?> message, InetAddressAndPort peer, Outcome outcome)
     {
-        removeAndExpire(message, peer);
+        removeAndOverload(message, peer, outcome);
     }
 
     @Override
@@ -310,6 +330,16 @@ private void removeAndExpire(Message message, InetAddressAndPort peer)
             forwardTo.forEach(this::removeAndExpire);
     }
 
+    private void removeAndOverload(Message message, InetAddressAndPort peer, Outcome outcome)
+    {
+        removeAndOverload(message.id(), peer, outcome);
+
+        /* in case of a write sent to a different DC, also fail all forwarding targets */
+        ForwardingInfo forwardTo = message.forwardTo();
+        if (null != forwardTo)
+            forwardTo.forEach((id, target) -> removeAndOverload(id, target, outcome));
+    }
+
     public static long defaultExpirationInterval()
     {
         return DatabaseDescriptor.getMinRpcTimeout(NANOSECONDS) / 2;