Disable old native protocol versions on demand

patch by Dinesh Joshi; reviewed by jasobrown for CASSANDRA-14659

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * Disable old native protocol versions on demand (CASANDRA-14659)
  * Allow specifying now-in-seconds in native protocol (CASSANDRA-14664)
  * Improve BTree build performance by avoiding data copy (CASSANDRA-9989)
  * Make monotonic read / read repair configurable (CASSANDRA-14635)

conf/cassandra.yaml

@@ -664,6 +664,10 @@ native_transport_port: 9042
 # The default is -1, which means unlimited.
 # native_transport_max_concurrent_connections_per_ip: -1
 
+# Controls whether Cassandra honors older, yet currently supported, protocol versions.
+# The default is true, which means all supported protocols will be honored.
+native_transport_allow_older_protocols: true
+
 # The address or interface to bind the native transport server to.
 #
 # Set rpc_address OR rpc_interface, not both.

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

@@ -153,6 +153,7 @@ public class Config
     public volatile long native_transport_max_concurrent_connections = -1L;
     public volatile long native_transport_max_concurrent_connections_per_ip = -1L;
     public boolean native_transport_flush_in_batches_legacy = false;
+    public volatile boolean native_transport_allow_older_protocols = true;
 
     /**
      * Max size of values in SSTables, in MegaBytes.

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

@@ -1871,6 +1871,16 @@ public class DatabaseDescriptor
         return conf.native_transport_flush_in_batches_legacy;
     }
 
+    public static boolean getNativeTransportAllowOlderProtocols()
+    {
+        return conf.native_transport_allow_older_protocols;
+    }
+
+    public static void setNativeTransportAllowOlderProtocols(boolean isEnabled)
+    {
+        conf.native_transport_allow_older_protocols = isEnabled;
+    }
+
     public static double getCommitLogSyncGroupWindow()
     {
         return conf.commitlog_sync_group_window_in_ms;

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

@@ -34,12 +34,10 @@ import java.util.stream.StreamSupport;
 import javax.annotation.Nullable;
 import javax.management.*;
 import javax.management.openmbean.CompositeData;
-import javax.management.openmbean.CompositeDataSupport;
 import javax.management.openmbean.OpenDataException;
 import javax.management.openmbean.TabularData;
 import javax.management.openmbean.TabularDataSupport;
 
-import com.clearspring.analytics.stream.Counter;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Predicate;
@@ -83,7 +81,6 @@ import org.apache.cassandra.io.sstable.SSTableLoader;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.locator.*;
 import org.apache.cassandra.metrics.StorageMetrics;
-import org.apache.cassandra.metrics.TableMetrics.Sampler;
 import org.apache.cassandra.net.*;
 import org.apache.cassandra.repair.*;
 import org.apache.cassandra.repair.messages.RepairOption;
@@ -107,7 +104,6 @@ import org.apache.cassandra.streaming.*;
 import org.apache.cassandra.tracing.TraceKeyspace;
 import org.apache.cassandra.transport.ProtocolVersion;
 import org.apache.cassandra.utils.*;
-import org.apache.cassandra.utils.TopKSampler.SamplerResult;
 import org.apache.cassandra.utils.logging.LoggingSupportFactory;
 import org.apache.cassandra.utils.progress.ProgressEvent;
 import org.apache.cassandra.utils.progress.ProgressEventType;
@@ -410,6 +406,18 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return daemon.isNativeTransportRunning();
     }
 
+    @Override
+    public void enableNativeTransportOldProtocolVersions()
+    {
+        DatabaseDescriptor.setNativeTransportAllowOlderProtocols(true);
+    }
+
+    @Override
+    public void disableNativeTransportOldProtocolVersions()
+    {
+        DatabaseDescriptor.setNativeTransportAllowOlderProtocols(false);
+    }
+
     public void stopTransports()
     {
         if (isGossipActive())

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

@@ -507,6 +507,8 @@ public interface StorageServiceMBean extends NotificationEmitter
     public void stopNativeTransport();
     public void startNativeTransport();
     public boolean isNativeTransportRunning();
+    public void enableNativeTransportOldProtocolVersions();
+    public void disableNativeTransportOldProtocolVersions();
 
     // allows a node that have been started without joining the ring to join it
     public void joinRing() throws IOException;

src/java/org/apache/cassandra/tools/NodeProbe.java

@@ -1749,6 +1749,16 @@ public class NodeProbe implements AutoCloseable
     {
         ssProxy.enableAuditLog(loggerName, includedKeyspaces, excludedKeyspaces, includedCategories, excludedCategories, includedUsers, excludedUsers);
     }
+
+    public void enableOldProtocolVersions()
+    {
+        ssProxy.enableNativeTransportOldProtocolVersions();
+    }
+
+    public void disableOldProtocolVersions()
+    {
+        ssProxy.disableNativeTransportOldProtocolVersions();
+    }
 }
 
 class ColumnFamilyStoreMBeanIterator implements Iterator<Map.Entry<String, ColumnFamilyStoreMBean>>

src/java/org/apache/cassandra/tools/NodeTool.java

@@ -165,7 +165,9 @@ public class NodeTool
                 HandoffWindow.class,
                 ReloadSslCertificates.class,
                 EnableAuditLog.class,
-                DisableAuditLog.class
+                DisableAuditLog.class,
+                EnableOldProtocolVersions.class,
+                DisableOldProtocolVersions.class
         );
 
         Cli.CliBuilder<Runnable> builder = Cli.builder("nodetool");

src/java/org/apache/cassandra/tools/nodetool/DisableOldProtocolVersions.java

@@ -0,0 +1,33 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.airline.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+@Command(name = "disableoldprotocolversions", description = "Disable old protocol versions")
+public class DisableOldProtocolVersions extends NodeTool.NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.disableOldProtocolVersions();
+    }
+}

src/java/org/apache/cassandra/tools/nodetool/EnableOldProtocolVersions.java

@@ -0,0 +1,34 @@
+/*
+ * 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.tools.nodetool;
+
+import io.airlift.airline.Command;
+import org.apache.cassandra.tools.NodeProbe;
+import org.apache.cassandra.tools.NodeTool;
+
+
+@Command(name = "enableoldprotocolversions", description = "Enable old protocol versions")
+public class EnableOldProtocolVersions extends NodeTool.NodeToolCmd
+{
+    @Override
+    public void execute(NodeProbe probe)
+    {
+        probe.enableOldProtocolVersions();
+    }
+}

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

@@ -26,6 +26,8 @@ import java.util.Optional;
 
 import org.apache.commons.lang3.ArrayUtils;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
+
 /**
  * The native (CQL binary) protocol version.
  *
@@ -114,6 +116,9 @@ public enum ProtocolVersion implements Comparable<ProtocolVersion>
             throw new ProtocolException(invalidVersionMessage(versionNum), MAX_SUPPORTED_VERSION);
         }
 
+        if (!DatabaseDescriptor.getNativeTransportAllowOlderProtocols() && ret.isSmallerThan(CURRENT))
+            throw new ProtocolException(String.format("Rejecting Protocol Version %s < %s.", ret, ProtocolVersion.CURRENT));
+
         return ret;
     }
 

test/unit/org/apache/cassandra/transport/ProtocolVersionTest.java

@@ -18,11 +18,30 @@
 
 package org.apache.cassandra.transport;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
+
 public class ProtocolVersionTest
 {
+    @BeforeClass
+    public static void setupDatabaseDescriptor()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setUp()
+    {
+        DatabaseDescriptor.setNativeTransportAllowOlderProtocols(true);
+    }
+
     @Test
     public void testDecode()
     {
@@ -100,4 +119,28 @@ public class ProtocolVersionTest
         Assert.assertFalse(ProtocolVersion.V3.isSmallerThan(ProtocolVersion.V2));
         Assert.assertFalse(ProtocolVersion.V2.isSmallerThan(ProtocolVersion.V1));
     }
+
+    @Test
+    public void testDisableOldProtocolVersions_Succeeds()
+    {
+        DatabaseDescriptor.setNativeTransportAllowOlderProtocols(false);
+        List<ProtocolVersion> disallowedVersions = ProtocolVersion.SUPPORTED
+                                                       .stream()
+                                                       .filter(v -> v.isSmallerThan(ProtocolVersion.CURRENT))
+                                                       .collect(Collectors.toList());
+
+        for (ProtocolVersion version : disallowedVersions)
+        {
+            try
+            {
+                ProtocolVersion.decode(version.asInt());
+                Assert.fail("Expected invalid protocol exception");
+            }
+            catch (ProtocolException ex)
+            {
+            }
+        }
+
+        Assert.assertEquals(ProtocolVersion.CURRENT, ProtocolVersion.decode(ProtocolVersion.CURRENT.asInt()));
+    }
 }