CASSANDRA-21191: [CEP-59] Implementation of In-Band Connection Draining (Graceful Disconnect)

CHANGES.txt

@@ -6,6 +6,7 @@
  * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394)
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
+ * [CEP-59] Implementation of In-Band Connection Draining (Graceful Disconnect) (CASSANDRA-21191)
 Merged from 6.0:
  * Make cqlsh prompt to reset to no keyspace set by USE after dropping that keyspace (CASSANDRA-21548)
  * Implement CMS rediscovery and recovery protocol (CASSANDRA-20476)

conf/cassandra.yaml

@@ -1104,6 +1104,19 @@ 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: doc/modules/cassandra/pages/managing/operating/graceful_disconnect.adoc
+# Defaults to false.
+# graceful_disconnect_enabled: false
+
+# Time given to clients to stop sending new requests after the GRACEFUL_DISCONNECT event is emitted.
+# graceful_disconnect_grace_period: 5s
+
 # The address or interface to bind the native transport server to.
 #
 # Set rpc_address OR rpc_interface, not both.

conf/cassandra_latest.yaml

@@ -1087,6 +1087,20 @@ 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: doc/modules/cassandra/pages/managing/operating/graceful_disconnect.adoc
+
+graceful_disconnect_enabled: true
+
+# Time given to clients to stop sending new requests after the GRACEFUL_DISCONNECT event is emitted.
+# graceful_disconnect_grace_period: 5s
+
 # The address or interface to bind the native transport server to.
 #
 # Set rpc_address OR rpc_interface, not both.

doc/modules/cassandra/pages/managing/operating/graceful_disconnect.adoc

@@ -0,0 +1,91 @@
+= Graceful Disconnect — In-Band Connection Draining for Cassandra Node Shutdown
+
+== Vocabulary
+
+in-band:: same connection
+In-flight requests:: requests that have been sent but not yet completed
+
+== Introduction
+
+When a Cassandra node has to be taken offline client drivers have no reliable, in-band signal that the node is going away.
+
+* Drivers keep sending requests to a shutting-down node until they hit a socket close or timeout.
+* In-flight requests are abandoned, producing `ReadTimeoutException` errors on the client side.
+* Retry storms emerge as clients simultaneously rediscover the topology.
+
+== Solution
+
+Introduce an in-band signal — `GRACEFUL_DISCONNECT` — so that the server can notify clients (that have subscribed) connection before closing it. This gives drivers time to:
+
+. Stop sending new requests on that connection/node.
+. Let all in-flight requests complete.
+. Close that socket connection (this notifies server that there are no pending queries on client side).
+. Try reconnecting with exponential backoff.
+
+== New Configurations Introduced
+
+`graceful_disconnect_enabled`:: A configuration that enables the server to perform graceful disconnect.
+`graceful_disconnect_grace_period`:: A configuration that forces shutdown of any active driver connection to the node after the grace period expires.
+
+[cols="1,1,3,1", options="header"]
+|===
+| Parameter | Type | Description | Default
+| `graceful_disconnect_enabled` | Boolean | A configuration that enables server to perform graceful disconnect. | false
+| `graceful_disconnect_grace_period` | Duration (ms) | A configuration that determines after how much time to force close a socket connection, if there is any pending connection from driver side. | 5s
+|===
+
+== Client Compatibility & Upgrade Strategy
+
+=== Legacy Drivers — No Change Required
+
+Drivers that do not support graceful disconnect are not affected by any value of `graceful_disconnect_enabled` or `graceful_disconnect_grace_period`.
+
+=== Required Driver Versions
+
+To benefit from graceful draining, a compatible driver must be used.
+
+[cols="1,1", options="header"]
+|===
+| Driver | Status
+| Java | https://issues.apache.org/jira/browse/CASSJAVA-124[In progress]
+| Python | https://issues.apache.org/jira/browse/CASSPYTHON-16[In progress]
+| Node.js | https://issues.apache.org/jira/browse/CASSNODEJS-5[In progress]
+| Go | https://issues.apache.org/jira/browse/CASSGO-117[In progress]
+| C++ | https://issues.apache.org/jira/browse/CASSCPP-7[In progress]
+|===
+
+=== Mixed-Fleet Rollout
+
+During a rolling upgrade where some nodes support the feature and others do not:
+
+* Nodes *with* `graceful_disconnect_enabled: true` advertise the capability in `SUPPORTED` and drivers subscribe.
+* Nodes *without* the feature omit the key from `SUPPORTED`. Compatible drivers silently fall back to the legacy TCP teardown path for those nodes.
+
+No special coordination is needed. Drivers handle both peers transparently within the same session.
+
+== Operational Visibility
+
+=== Metrics
+
+[cols="1,3", options="header"]
+|===
+| Metric | Description
+| `ConnectionsDraining` | Current count of connections in the Draining state (subscribed and awaiting in-flight completion). Should rise at drain start and fall to zero before shutdown completes.
+| `ForcedDisconnects` | Cumulative count of connections force-closed after `graceful_disconnect_grace_period` expired without the driver closing cleanly. Persistent non-zero values here indicate driver-side issues or an undersized grace period.
+|===
+
+== Failure Modes & Edge Cases
+
+=== What if the driver does not support `GRACEFUL_DISCONNECT`?
+
+If the driver never sent `REGISTER` for this event type (legacy driver, or a compatible driver connected to a legacy node), the server applies *standard TCP teardown* — the same behavior as before this feature existed. There is no error, no retry storm specific to this feature, and no action required from operators.
+
+This is the designed fallback and not an error condition.
+
+=== What will happen if two different driver, one supporting graceful disconnect while other not, are talking to a node supporting graceful disconnect?
+
+Driver supporting graceful disconnect will disconnect gracefully, while the one not supporting graceful disconnect will behave as a legacy driver.
+
+=== What if the server crashes instead of draining cleanly?
+
+`GRACEFUL_DISCONNECT` is *not* a crash-safety mechanism. If the JVM is killed with `SIGKILL`, the node loses power, or an OOM kill occurs, no `GRACEFUL_DISCONNECT` event is emitted. Drivers fall back to socket-close detection and gossip `DOWN` events — the current behavior.

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

@@ -143,6 +143,10 @@ public static Set<String> splitCommaDelimited(String src)
     /** Triggers automatic allocation of tokens if set, based on the provided replica count for a datacenter */
     public Integer allocate_tokens_for_local_replication_factor = null;
 
+    public boolean graceful_disconnect_enabled = false;
+
+    public volatile DurationSpec.LongMillisecondsBound graceful_disconnect_grace_period = new DurationSpec.LongMillisecondsBound(5000);
+
     @Replaces(oldName = "native_transport_idle_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public DurationSpec.LongMillisecondsBound native_transport_idle_timeout = new DurationSpec.LongMillisecondsBound("0ms");
 

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

@@ -2685,6 +2685,24 @@ 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)
+    {
+        if (gracefulDisconnectGracePeriod > 0)
+            conf.graceful_disconnect_grace_period = new DurationSpec.LongMillisecondsBound(gracefulDisconnectGracePeriod);
+        else
+            throw new IllegalArgumentException(String.format("{} <= 0, not allowed, non positive values not allowed", gracefulDisconnectGracePeriod));
+    }
+
+    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/metrics/ClientMetrics.java

@@ -71,6 +71,10 @@ public final class ClientMetrics
     @VisibleForTesting
     Gauge<Integer> connectedNativeClients;
 
+    public AtomicInteger connectionsDraining = new AtomicInteger();
+
+    public Meter forcedDisconnects;
+
     @VisibleForTesting
     Gauge<Integer> encryptedConnectedNativeClients;
 
@@ -173,6 +177,21 @@ public void markProtocolException()
         protocolException.mark();
     }
 
+    public void incrementConnectionsDraining()
+    {
+        connectionsDraining.incrementAndGet();
+    }
+
+    public void decrementConnectionsDraining()
+    {
+        connectionsDraining.decrementAndGet();
+    }
+
+    public void markForcedDisconnect(int forceDisconnectedClients)
+    {
+        forcedDisconnects.mark(forceDisconnectedClients);
+    }
+
     public void markSSLHandshakeException()
     {
         sslHandshakeException.mark();
@@ -197,6 +216,10 @@ public synchronized void init(Server servers)
         registerGauge("ClientsByProtocolVersion", "clientsByProtocolVersion", this::recentClientStats);
         registerGauge("RequestsSize", ClientResourceLimits::getCurrentGlobalUsage);
 
+        connectionsDraining = new AtomicInteger();
+        registerGauge("ConnectionsDraining", connectionsDraining::get);
+        forcedDisconnects = registerMeter("ForcedDisconnects");
+
         CassandraReservoir ipUsageReservoir = ClientResourceLimits.ipUsageReservoir();
         Metrics.register(factory.createMetricName("RequestsSizeByIpDistribution"),
                          new OverrideHistogram(ipUsageReservoir)

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

@@ -37,6 +37,7 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.epoll.Epoll;
 import io.netty.channel.epoll.EpollEventLoopGroup;
+import io.netty.channel.group.ChannelGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.util.Version;
 
@@ -163,6 +164,11 @@ Server getServer()
         return server;
     }
 
+    public ChannelGroup getChannelsSubscribedToGracefulDisconnect()
+    {
+        return server.getChannelsSubscribedToGracefulDisconnect();
+    }
+
     public void clearConnectionHistory()
     {
         server.clearConnectionHistory();

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

@@ -1199,6 +1199,27 @@ public long getRpcTimeout()
         return DatabaseDescriptor.getRpcTimeout(MILLISECONDS);
     }
 
+
+    @Override
+    public boolean getGracefulDisconnectEnabled()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectEnabled();
+    }
+
+    @Override
+    public void setGracefulDisconnectGracePeriod(long value)
+    {
+        if (value <= 0 && DatabaseDescriptor.getGracefulDisconnectEnabled())
+            throw new IllegalArgumentException("Graceful disconnect grace period must be positive when graceful disconnect is enabled. Got " + value);
+        DatabaseDescriptor.setGracefulDisconnectGracePeriod(value);
+    }
+
+    @Override
+    public long getGracefulDisconnectGracePeriod()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectGracePeriod();
+    }
+
     public void setReadRpcTimeout(long value)
     {
         DatabaseDescriptor.setReadRpcTimeout(value);

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

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

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

@@ -34,7 +34,8 @@ public void execute(NodeProbe probe)
         try
         {
             probe.drain();
-        } catch (IOException | InterruptedException | ExecutionException e)
+        }
+        catch (IOException | InterruptedException | ExecutionException e)
         {
             throw new RuntimeException("Error occurred during flushing", e);
         }

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

@@ -36,7 +36,8 @@ public enum Type
         TOPOLOGY_CHANGE(ProtocolVersion.V3),
         STATUS_CHANGE(ProtocolVersion.V3),
         SCHEMA_CHANGE(ProtocolVersion.V3),
-        TRACE_COMPLETE(ProtocolVersion.V4);
+        TRACE_COMPLETE(ProtocolVersion.V4),
+        GRACEFUL_DISCONNECT(ProtocolVersion.V5);
 
         public final ProtocolVersion minimumVersion;
 
@@ -66,6 +67,8 @@ public static Event deserialize(ByteBuf cb, ProtocolVersion version)
                 return StatusChange.deserializeEvent(cb, version);
             case SCHEMA_CHANGE:
                 return SchemaChange.deserializeEvent(cb, version);
+            case GRACEFUL_DISCONNECT:
+                return GracefulDisconnect.deserializeEvent(cb, version);
         }
         throw new AssertionError();
     }
@@ -442,4 +445,26 @@ public boolean equals(Object other)
                 && Objects.equal(argTypes, scc.argTypes);
         }
     }
+
+    public static class GracefulDisconnect extends Event
+    {
+        public GracefulDisconnect()
+        {
+            super(Type.GRACEFUL_DISCONNECT);
+        }
+
+        @Override
+        protected int eventSerializedSize(ProtocolVersion version)
+        {
+            return 0;
+        }
+
+        @Override
+        protected void serializeEvent(ByteBuf dest, ProtocolVersion version) {}
+
+        public static GracefulDisconnect deserializeEvent(ByteBuf cb, ProtocolVersion version)
+        {
+            return new GracefulDisconnect();
+        }
+    }
 }