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.

doc/native_protocol_v5.spec

@@ -698,6 +698,12 @@ Table of Contents
   for events on all connections, as this would only result in receiving
   multiple times the same event messages, wasting bandwidth.
 
+  Note: Unlike cluster-wide events (such as TOPOLOGY_CHANGE, STATUS_CHANGE,
+  and SCHEMA_CHANGE) which are distributed to any registered connection, the
+  "GRACEFUL_DISCONNECT" event is strictly connection-specific. To benefit
+  from graceful connection draining, clients must register for the "GRACEFUL_DISCONNECT"
+  event type on all connections they establish.
+
 
 4.2. Responses
 
@@ -757,6 +763,11 @@ Table of Contents
       version description. For example: 3/v3, 4/v4, 5/v5-beta. If a version is
       in beta, it will have the word "beta" in its description.
 
+      - "GRACEFUL_DISCONNECT": if graceful disconnect is supported and enabled
+      by the server, this key will be present with the list of supported values
+      set to ["true"]. If the key is absent, the server does not support or has
+      disabled the graceful disconnect feature.
+
 
 4.2.5. RESULT
 
@@ -987,6 +998,8 @@ Table of Contents
       consists of a [string] and an [inet], corresponding respectively to the
       type of change ("NEW_NODE" or "REMOVED_NODE") followed by the address of
       the new/removed node.
+
+
     - "STATUS_CHANGE": events related to change of node status. Currently,
       up/down events are sent. The body of the message (after the event type)
       consists of a [string] and an [inet], corresponding respectively to the
@@ -1013,6 +1026,12 @@ Table of Contents
             - [string] keyspace containing the user defined function / aggregate
             - [string] the function/aggregate name
             - [string list] one string for each argument type (as CQL type)
+    - "GRACEFUL_DISCONNECT": events related to an impending clean shutdown of
+      the node. This event signals to the client that the server is shutting down
+      and will close the connection after the configured grace period. The body of
+      the message (after the event type) is empty. Upon receiving this event,
+      the client must stop sending new queries on this connection, let in-flight
+      queries finish, and cleanly close the socket.
 
   All EVENT messages have a streamId of -1 (Section 2.4.1.3).
 

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,23 @@ 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)
+            throw new IllegalArgumentException(String.format("graceful_disconnect_grace_period must be positive, got %d", gracefulDisconnectGracePeriod));
+        conf.graceful_disconnect_grace_period = new DurationSpec.LongMillisecondsBound(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,16 @@ public void markProtocolException()
         protocolException.mark();
     }
 
+    public void decrementConnectionsDraining()
+    {
+        connectionsDraining.decrementAndGet();
+    }
+
+    public void markForcedDisconnect(int forceDisconnectedClients)
+    {
+        forcedDisconnects.mark(forceDisconnectedClients);
+    }
+
     public void markSSLHandshakeException()
     {
         sslHandshakeException.mark();
@@ -197,6 +211,9 @@ public synchronized void init(Server servers)
         registerGauge("ClientsByProtocolVersion", "clientsByProtocolVersion", this::recentClientStats);
         registerGauge("RequestsSize", ClientResourceLimits::getCurrentGlobalUsage);
 
+        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

@@ -163,6 +163,12 @@ Server getServer()
         return server;
     }
 
+    @VisibleForTesting
+    public int getChannelsSubscribedToGracefulDisconnectCount()
+    {
+        return server.getChannelsSubscribedToGracefulDisconnect().size();
+    }
+
     public void clearConnectionHistory()
     {
         server.clearConnectionHistory();

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

@@ -1199,6 +1199,25 @@ public long getRpcTimeout()
         return DatabaseDescriptor.getRpcTimeout(MILLISECONDS);
     }
 
+
+    @Override
+    public boolean getGracefulDisconnectEnabled()
+    {
+        return DatabaseDescriptor.getGracefulDisconnectEnabled();
+    }
+
+    @Override
+    public void setGracefulDisconnectGracePeriod(long 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/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.V4);
 
         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();
+        }
+    }
 }

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

@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.net.AsyncChannelPromise;
 import org.apache.cassandra.transport.ClientResourceLimits.Overload;
@@ -82,6 +83,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> li
                     cqlVersions.add(QueryProcessor.CQL_VERSION.toString());
 
                     List<String> compressions = new ArrayList<>();
+                    final List<String> gracefulDisconnect = List.of("true");
                     if (Compressor.SnappyCompressor.instance != null)
                         compressions.add("snappy");
                     // LZ4 is always available since worst case scenario it default to a pure JAVA implem.
@@ -91,6 +93,8 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> li
                     supportedOptions.put(StartupMessage.CQL_VERSION, cqlVersions);
                     supportedOptions.put(StartupMessage.COMPRESSION, compressions);
                     supportedOptions.put(StartupMessage.PROTOCOL_VERSIONS, ProtocolVersion.supportedVersions());
+                    if (DatabaseDescriptor.getGracefulDisconnectEnabled())
+                        supportedOptions.put(StartupMessage.GRACEFUL_DISCONNECT, gracefulDisconnect);
                     SupportedMessage supported = new SupportedMessage(supportedOptions);
                     outbound = supported.encode(inbound.header.version, inbound.header.streamId);
                     ctx.writeAndFlush(outbound);