Feature/cep 59/configs setup

conf/cassandra.yaml

@@ -1083,6 +1083,29 @@ native_transport_allow_older_protocols: true
 # native_transport_rate_limiting_enabled: false
 # native_transport_max_requests_per_second: 1000000
 
+# When enabled, nodes will signal connected clients before shutting down,
+# allowing in-flight requests to complete without client-visible timeouts.
+# This applies to intentional shutdowns (nodetool drain, rolling restarts,
+# controlled JVM shutdown). Clients must subscribe to the GRACEFUL_DISCONNECT
+# event via REGISTER to benefit from this behavior.
+#
+# Requires driver support for the GRACEFUL_DISCONNECT event type.
+# See: https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=406619103
+
+# Enable or disable graceful disconnect. When false, shutdown behavior is
+# unchanged from previous versions.
+# graceful_disconnect_enabled: false
+
+# Time given to clients to stop sending new requests after the
+# GRACEFUL_DISCONNECT event is emitted. Must not exceed graceful_disconnect_max_drain.
+# graceful_disconnect_grace_period: 5000
+
+# Hard timeout for draining connections. Once this limit is reached,
+# remaining connections are force-closed regardless of in-flight requests.
+# Must be greater than 0 when graceful_disconnect_enabled is true.
+# graceful_disconnect_max_drain: 30000
+
+
 # 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

@@ -167,6 +167,12 @@ public static Set<String> splitCommaDelimited(String src)
 
     public volatile DurationSpec.LongMillisecondsBound accord_preaccept_timeout = new DurationSpec.LongMillisecondsBound("1s");
 
+    public boolean graceful_disconnect_enabled = false;
+
+    public volatile DurationSpec.LongMillisecondsBound graceful_disconnect_grace_period = new  DurationSpec.LongMillisecondsBound("5s");
+
+    public volatile DurationSpec.LongMillisecondsBound graceful_disconnect_max_drain = new  DurationSpec.LongMillisecondsBound("30s");
+
     @Replaces(oldName = "truncate_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound truncate_request_timeout = new DurationSpec.LongMillisecondsBound("60000ms");
 

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

@@ -685,6 +685,17 @@ else if (DiskAccessMode.direct == conf.compaction_read_disk_access_mode)
             throw new ConfigurationException("phi_convict_threshold must be between 5 and 16, but was " + conf.phi_convict_threshold, false);
         }
 
+        if (conf.graceful_disconnect_enabled && conf.graceful_disconnect_max_drain.toMilliseconds() <= 0)
+        {
+            throw new ConfigurationException("graceful_disconnect_max_drain must be greater than 0 when graceful_disconnect_enabled is set to true.", false);
+        }
+
+        if (conf.graceful_disconnect_enabled &&
+            conf.graceful_disconnect_grace_period.toMilliseconds() > conf.graceful_disconnect_max_drain.toMilliseconds())
+        {
+            throw new ConfigurationException("graceful_disconnect_grace_period cannot exceed graceful_disconnect_max_drain.", false);
+        }
+
         /* Thread per pool */
         if (conf.concurrent_reads < 2)
         {
@@ -2598,6 +2609,31 @@ public static void setRpcTimeout(long timeOutInMillis)
         conf.request_timeout = new DurationSpec.LongMillisecondsBound(timeOutInMillis);
     }
 
+    public static long getGracefulDisconnectGracePeriod()
+    {
+        return conf.graceful_disconnect_grace_period.toMilliseconds();
+    }
+
+    public static void setGracefulDisconnectGracePeriod(long gracefulDisconnectGracePeriod)
+    {
+        conf.graceful_disconnect_grace_period = new DurationSpec.LongMillisecondsBound(gracefulDisconnectGracePeriod);
+    }
+
+    public static long getGracefulDisconnectMaxDrain()
+    {
+        return conf.graceful_disconnect_max_drain.toMilliseconds();
+    }
+
+    public static void setGracefulDisconnectMaxDrain(long gracefulDisconnectMaxDrain)
+    {
+        conf.graceful_disconnect_max_drain = new DurationSpec.LongMillisecondsBound(gracefulDisconnectMaxDrain);
+    }
+
+    public static boolean getGracefulDisconnectEnabled()
+    {
+        return conf.graceful_disconnect_enabled;
+    }
+
     public static long getReadRpcTimeout(TimeUnit unit)
     {
         return conf.read_request_timeout.to(unit);

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

@@ -1197,6 +1197,48 @@ public long getRpcTimeout()
         return DatabaseDescriptor.getRpcTimeout(MILLISECONDS);
     }
 
+    @Override
+    public void setGracefulDisconnectGracePeriod(long value)
+    {
+        if (value > DatabaseDescriptor.getGracefulDisconnectMaxDrain())
+            throw new IllegalArgumentException("graceful_disconnect_grace_period cannot exceed graceful_disconnect_max_drain.");
+
+        DatabaseDescriptor.setGracefulDisconnectGracePeriod(value);
+        logger.info("set graceful disconnect grace period to {} ms", value);
+    }
+
+    @Override
+    public long getGracefulDisconnectGracePeriod()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectGracePeriod();
+    }
+
+    @Override
+    public void setGracefulDisconnectMaxDrain(long value)
+    {
+        if (value <= 0)
+            throw new IllegalArgumentException("graceful_disconnect_max_drain must be greater than 0 when graceful_disconnect_enabled is set to true.");
+
+        if (value < DatabaseDescriptor.getGracefulDisconnectGracePeriod())
+            throw new IllegalArgumentException("graceful_disconnect_max_drain cannot be less than graceful_disconnect_grace_period.");
+
+        DatabaseDescriptor.setGracefulDisconnectMaxDrain(value);
+        logger.info("set graceful disconnect max drain to {} ms", value);
+    }
+
+
+    @Override
+    public long getGracefulDisconnectMaxDrain()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectMaxDrain();
+    }
+
+    @Override
+    public boolean getGracefulDisconnectEnabled()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectEnabled();
+    }
+
     public void setReadRpcTimeout(long value)
     {
         DatabaseDescriptor.setReadRpcTimeout(value);

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

@@ -773,6 +773,14 @@ default int upgradeSSTables(String keyspaceName, boolean excludeCurrentVersion,
     public void setRpcTimeout(long value);
     public long getRpcTimeout();
 
+    public void setGracefulDisconnectGracePeriod(long value);
+    public long getGracefulDisconnectGracePeriod();
+
+    public void setGracefulDisconnectMaxDrain(long value);
+    public long getGracefulDisconnectMaxDrain();
+
+    public boolean getGracefulDisconnectEnabled();
+
     public void setReadRpcTimeout(long value);
     public long getReadRpcTimeout();
 

test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java

@@ -791,6 +791,34 @@ public void testRowIndexSizeWarnEnabledAbortDisabled()
         DatabaseDescriptor.applyThresholdsValidations(conf);
     }
 
+    @Test
+    public void testGracefulDisconnectEnabled()
+    {
+        Config config = new Config();
+        boolean originalValue = config.graceful_disconnect_enabled;
+        Assert.assertFalse("Default value of graceful_disconnect_enabled must be false", originalValue);
+    }
+
+    @Test
+    public void testGracefulDisconnectGracePeriod()
+    {
+        long originalValue = DatabaseDescriptor.getGracefulDisconnectGracePeriod();
+        Assert.assertEquals("Default value of graceful_disconnect_grace_period must be 5000", 5000, originalValue);
+        DatabaseDescriptor.setGracefulDisconnectGracePeriod(3000);
+        Assert.assertEquals("graceful_disconnect_grace_period should be updated to 3000", 3000, DatabaseDescriptor.getGracefulDisconnectGracePeriod());
+        DatabaseDescriptor.setGracefulDisconnectGracePeriod(originalValue);
+    }
+
+    @Test
+    public void testGracefulDisconnectMaxDrain()
+    {
+        long originalValue = DatabaseDescriptor.getGracefulDisconnectMaxDrain();
+        Assert.assertEquals("Default value of graceful_disconnect_max_drain must be 30000", 30000, originalValue);
+        DatabaseDescriptor.setGracefulDisconnectMaxDrain(45000);
+        Assert.assertEquals("graceful_disconnect_max_drain should be updated to 45000", 45000, DatabaseDescriptor.getGracefulDisconnectMaxDrain());
+        DatabaseDescriptor.setGracefulDisconnectMaxDrain(originalValue);
+    }
+
     @Test
     public void testRowIndexSizeAbortEnabledWarnDisabled()
     {

test/unit/org/apache/cassandra/service/StorageServiceTest.java

@@ -201,6 +201,63 @@ public void testColumnIndexSizeInKiB()
         }
     }
 
+    @Test
+    public void testGracefulDisconnectMaxDrain()
+    {
+        StorageService storageService = StorageService.instance;
+        long originalMaxDrain = storageService.getGracefulDisconnectMaxDrain();
+        long originalGracePeriod = storageService.getGracefulDisconnectGracePeriod();
+        try
+        {
+            storageService.setGracefulDisconnectGracePeriod(1000);
+
+            storageService.setGracefulDisconnectMaxDrain(10000);
+            assertEquals(10000, storageService.getGracefulDisconnectMaxDrain());
+            try
+            {
+                storageService.setGracefulDisconnectMaxDrain(0);
+                fail("Should have received an IllegalArgumentException for max_drain of 0");
+            }
+            catch (IllegalArgumentException ignored)
+            {
+            }
+            assertEquals(10000, storageService.getGracefulDisconnectMaxDrain());
+        }
+        finally
+        {
+            storageService.setGracefulDisconnectMaxDrain(originalMaxDrain);
+            storageService.setGracefulDisconnectGracePeriod(originalGracePeriod);
+        }
+    }
+
+    @Test
+    public void testGracefulDisconnectGracePeriod()
+    {
+        StorageService storageService = StorageService.instance;
+        long originalMaxDrain = storageService.getGracefulDisconnectMaxDrain();
+        long originalGracePeriod = storageService.getGracefulDisconnectGracePeriod();
+        try
+        {
+            storageService.setGracefulDisconnectMaxDrain(20000);
+
+            storageService.setGracefulDisconnectGracePeriod(5000);
+            assertEquals(5000, storageService.getGracefulDisconnectGracePeriod());
+
+            try
+            {
+                storageService.setGracefulDisconnectGracePeriod(30000);
+                fail("Should have received an IllegalArgumentException when grace_period exceeds max_drain");
+            }
+            catch (IllegalArgumentException ignored) {}
+            assertEquals(5000, storageService.getGracefulDisconnectGracePeriod());
+        }
+        finally
+        {
+            storageService.setGracefulDisconnectMaxDrain(originalMaxDrain);
+            storageService.setGracefulDisconnectGracePeriod(originalGracePeriod);
+        }
+    }
+
     @Test
     public void testColumnIndexCacheSizeInKiB()
     {