CASSANDRA-19483 Optionally skip exception logging on invalid legacy protocol magic ex…

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

@@ -820,6 +820,7 @@ public static void setClientMode(boolean clientMode)
 
     public volatile SubnetGroups client_error_reporting_exclusions = new SubnetGroups();
     public volatile SubnetGroups internode_error_reporting_exclusions = new SubnetGroups();
+    public volatile boolean invalid_legacy_protocol_magic_no_spam_enabled = false;
 
     public volatile int keyspaces_warn_threshold = -1;
     public volatile int keyspaces_fail_threshold = -1;

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

@@ -4339,6 +4339,11 @@ public static SubnetGroups getInternodeErrorReportingExclusions()
         return conf.internode_error_reporting_exclusions;
     }
 
+    public static boolean getInvalidLegacyProtocolMagicNoSpamEnabled()
+    {
+        return conf.invalid_legacy_protocol_magic_no_spam_enabled;
+    }
+
     public static boolean getReadThresholdsEnabled()
     {
         return conf.read_thresholds_enabled;

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

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -56,6 +57,7 @@
 import org.apache.cassandra.streaming.StreamDeserializingTask;
 import org.apache.cassandra.streaming.StreamingChannel;
 import org.apache.cassandra.streaming.async.NettyStreamingChannel;
+import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.memory.BufferPools;
 
 import static java.lang.Math.*;
@@ -69,6 +71,8 @@ public class InboundConnectionInitiator
 {
     private static final Logger logger = LoggerFactory.getLogger(InboundConnectionInitiator.class);
 
+    private static final NoSpamLogger noSpam5m = NoSpamLogger.getLogger(logger, 5, TimeUnit.MINUTES);
+
     private static class Initializer extends ChannelInitializer<SocketChannel>
     {
         private static final String PIPELINE_INTERNODE_ERROR_EXCLUSIONS = "Internode Error Exclusions";
@@ -396,6 +400,8 @@ private void exceptionCaught(Channel channel, Throwable cause)
 
             if (reportingExclusion)
                 logger.debug("Excluding internode exception for {}; address contained in internode_error_reporting_exclusions", remoteAddress, cause);
+            else if (cause != null && cause.getCause() != null && cause.getCause() instanceof Message.InvalidLegacyProtocolMagic && DatabaseDescriptor.getInvalidLegacyProtocolMagicNoSpamEnabled())
+                noSpam5m.warn("Failed to properly handshake with peer " + ((InetSocketAddress) channel.remoteAddress()).getHostName() + ". Closing the channel. Invalid legacy protocol magic. {}", cause.getMessage());
             else
                 logger.error("Failed to properly handshake with peer {}. Closing the channel.", remoteAddress, cause);
 

test/distributed/org/apache/cassandra/distributed/test/InternodeErrorExclusionTest.java

@@ -53,16 +53,38 @@ public void ignoreExcludedInternodeErrors() throws IOException, TimeoutException
                                                        .set("internode_error_reporting_exclusions", ImmutableMap.of("subnets", Arrays.asList("127.0.0.1"))))
                                       .start())
         {
-            try (SimpleClient client = SimpleClient.builder("127.0.0.1", 7012).build())
-            {
-                client.connect(true);
-                Assert.fail("Connection should fail");
-            }
-            catch (Exception e)
-            {
-                // expected
-            }
+
+            causeException();
             assertThat(cluster.get(1).logs().watchFor("address contained in internode_error_reporting_exclusions").getResult()).hasSize(1);
         }
     }
+
+    @Test
+    public void testNoSpammingInvalidLegacyProtocolMagicException() throws Throwable
+    {
+        try (Cluster cluster = Cluster.build(1)
+                                      .withConfig(c -> c
+                                                       .with(Feature.NETWORK)
+                                                       .set("invalid_legacy_protocol_magic_no_spam_enabled", true))
+                                      .start())
+        {
+            causeException();
+            causeException();
+            // we used no spam logger so the second message will not be emitted (the size is still 1).
+            assertThat(cluster.get(1).logs().watchFor("Failed to properly handshake with peer localhost. Closing the channel. Invalid legacy protocol magic.").getResult()).hasSize(1);
+        }
+    }
+
+    private void causeException()
+    {
+        try (SimpleClient client = SimpleClient.builder("127.0.0.1", 7012).build())
+        {
+            client.connect(true);
+            Assert.fail("Connection should fail");
+        }
+        catch (Exception e)
+        {
+            // expected
+        }
+    }
 }