CNDB-14624: do not fail user read when speculative retry handling throws

Author: jakubzytka

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

@@ -94,7 +94,7 @@
 @SuppressWarnings({ "WeakerAccess", "FieldMayBeFinal", "NonAtomicOperationOnVolatileField", "SameParameterValue" })
 public class OutboundConnection
 {
-    static final Logger logger = LoggerFactory.getLogger(OutboundConnection.class);
+    private static final Logger logger = LoggerFactory.getLogger(OutboundConnection.class);
     private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 30L, TimeUnit.SECONDS);
 
     private static final AtomicLongFieldUpdater<OutboundConnection> submittedUpdater = AtomicLongFieldUpdater.newUpdater(OutboundConnection.class, "submittedCount");
@@ -107,6 +107,8 @@ public class OutboundConnection
     private final EventLoop eventLoop;
     private final Delivery delivery;
 
+    private final String id;
+
     private final OutboundMessageCallbacks callbacks;
     private final OutboundDebugCallbacks debug;
     @VisibleForTesting
@@ -313,6 +315,7 @@ void cancel()
                         ? new LargeMessageDelivery(template.socketFactory.synchronousWorkExecutor)
                         : new EventLoopDelivery();
         setDisconnected();
+        this.id = computeId();
     }
 
     /**
@@ -485,7 +488,7 @@ private boolean onExpired(Message<?> message)
      */
     private void onFailedSerialize(Message<?> message, int messagingVersion, int bytesWrittenToNetwork, Throwable t)
     {
-        logger.warn("{} dropping message of type {} due to error", id(), message.verb(), t);
+        noSpamLogger.warn("{} dropping message of type {} due to error", id(), message.verb(), t);
         JVMStabilityInspector.inspectThrowable(t);
         releaseCapacity(1, canonicalSize(message));
         errorCount += 1;
@@ -949,7 +952,7 @@ public void run()
             try
             {
                 priorThreadName = Thread.currentThread().getName();
-                threadName = "Messaging-OUT-" + template.from() + "->" + template.to + '-' + type;
+                threadName = "Messaging-OUT-" + template.from + "->" + template.to + '-' + type;
                 Thread.currentThread().setName(threadName);
 
                 super.run();
@@ -1089,21 +1092,23 @@ class Initiate
             void onFailure(Throwable cause)
             {
                 if (cause instanceof ConnectException)
-                    noSpamLogger.info("{} failed to connect", id(), cause);
+                    logger.info("{} failed to connect", id(), cause);
                 else
-                    noSpamLogger.error("{} failed to connect", id(), cause);
+                    logger.error("{} failed to connect", id(), cause);
 
                 JVMStabilityInspector.inspectThrowable(cause);
 
                 if (hasPending())
                 {
+                    logger.info("{} scheduling reconnect to {}", id(), settings.connectToId());
                     Promise<Result<MessagingSuccess>> result = new AsyncPromise<>(eventLoop);
                     state = new Connecting(state.disconnected(), result, eventLoop.schedule(() -> attempt(result), max(100, retryRateMillis), MILLISECONDS));
                     retryRateMillis = min(1000, retryRateMillis * 2);
                 }
                 else
                 {
                     // this Initiate will be discarded
+                    logger.info("{} no more pending messages; not retrying connection to {}", id(), settings.connectToId());
                     state = Disconnected.dormant(state.disconnected().maintenance);
                 }
             }
@@ -1143,7 +1148,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                                 }
                                 catch (Throwable t)
                                 {
-                                    logger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
+                                    noSpamLogger.error("Unexpected exception in {}.exceptionCaught", this.getClass().getSimpleName(), t);
                                 }
                             }
                         });
@@ -1191,6 +1196,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
             private void attempt(Promise<Result<MessagingSuccess>> result)
             {
                 ++connectionAttempts;
+                logger.info("{} attempting to connect to {}; attempt {}", id(), template.connectToId(), connectionAttempts);
 
                 /*
                  * Re-evaluate messagingVersion before re-attempting the connection in case
@@ -1601,6 +1607,11 @@ private String id(boolean includeReal)
     }
 
     private String id()
+    {
+        return id;
+    }
+
+    private String computeId()
     {
         State state = this.state;
         Channel channel = null;

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

@@ -172,8 +172,8 @@ public boolean withEncryption()
 
     public String toString()
     {
-        return String.format("peer: (%s, %s), framing: %s, encryption: %s",
-                             to, connectTo, framing, SocketFactory.encryptionOptionsSummary(encryption));
+        return String.format("from: %s peer: (%s, %s), framing: %s, encryption: %s",
+                             from, to, connectTo, framing, SocketFactory.encryptionOptionsSummary(encryption));
     }
 
     public OutboundConnectionSettings withAuthenticator(IInternodeAuthenticator authenticator)

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

@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.net;
+
+public class UnknownEndpointException extends RuntimeException
+{
+    UnknownEndpointException(String message)
+    {
+        super(message);
+    }
+}

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -117,6 +117,7 @@
 import org.apache.cassandra.net.MessageFlag;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.net.RequestCallback;
+import org.apache.cassandra.net.UnknownEndpointException;
 import org.apache.cassandra.net.Verb;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
@@ -2239,7 +2240,18 @@ private static PartitionIterator fetchRows(List<SinglePartitionReadCommand> comm
         // set of replicas we sent messages to, speculatively send an additional messages to an un-contacted replica
         for (int i=0; i<cmdCount; i++)
         {
-            reads[i].maybeTryAdditionalReplicas();
+            try
+            {
+                reads[i].maybeTryAdditionalReplicas();
+            }
+            catch (UnknownEndpointException ex)
+            {
+                logger.debug("Failed to send speculative read retries; The target replica is not known: {}", ex.getMessage());
+            }
+            catch (Exception ex)
+            {
+                logger.warn("Caught exception during speculative read retry; This is unexpected, but we're ignoring it because spec retry is just a best-effort attempt;", ex);
+            }
         }
 
         // wait for enough responses to meet the consistency level. If there's a digest mismatch, begin the read