Skip to content
Open
Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
a226207
WIP, driver can register GRACEFUL_DISCONNECT event but cannot receive…
SiyaoIsHiding Feb 26, 2026
45c5f95
WIP
SiyaoIsHiding Feb 27, 2026
8c5139e
works. Got Connection reset by peers
SiyaoIsHiding Mar 12, 2026
0a95963
docker compose
SiyaoIsHiding Mar 18, 2026
1de60ae
channel.close();
SiyaoIsHiding Mar 20, 2026
3927a33
java 11
SiyaoIsHiding Apr 2, 2026
32fcb1b
CASSJAVA-124: Add GRACEFUL_DISCONNECT support (CEP-59)
Shanzita May 28, 2026
85e7c25
CASSJAVA-124: Revert logback-test.xml to trunk
Shanzita Aug 26, 2026
442561d
CASSJAVA-124: Track graceful-disconnect capability per connection
Shanzita Aug 26, 2026
23835de
CASSJAVA-124: Remove unneeded config stubs from event-processing tests
Shanzita Aug 26, 2026
618a7f8
CASSJAVA-124: Initialize GRACEFUL_DISCONNECTS metrics in all backends
Shanzita Aug 26, 2026
c146d19
CASSJAVA-124: Real integration test for graceful disconnect
Shanzita Aug 26, 2026
402d1bf
CASSJAVA-124: Install native-protocol snapshot from the cep-59 branch
Shanzita Aug 26, 2026
6de6523
CASSJAVA-124: Install snapshot dependencies in CI before building
Shanzita Aug 26, 2026
4885827
CASSJAVA-124: Install native-protocol snapshot from the PR #61 branch
Shanzita Aug 26, 2026
6b9ac8f
CASSJAVA-124: Harden snapshot install for CI
Shanzita Aug 26, 2026
3234760
CASSJAVA-124: Revert CI and snapshot-install changes
Shanzita Aug 28, 2026
2ae468b
CASSJAVA-124: Simplify GRACEFUL_DISCONNECT registration
Shanzita Aug 28, 2026
86d0fad
CASSJAVA-124: Move GracefulDisconnectEvent to the metadata package
Shanzita Aug 28, 2026
9e1437e
CASSJAVA-124: Ensure the IT load thread terminates deterministically
Shanzita Aug 28, 2026
ef912d2
CASSJAVA-124: Update registration tests for the simplified handling
Shanzita Aug 28, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion bom/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@
<dependency>
<groupId>com.datastax.oss</groupId>
<artifactId>native-protocol</artifactId>
<version>1.5.2</version>
<version>1.5.3-SNAPSHOT</version>
Comment thread
Shanzita marked this conversation as resolved.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to remember to change it to 1.5.3 after the release of the native protocol

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agreed — I'll bump this to 1.5.3 as soon as native-protocol releases (tracking it in my native-protocol PR, datastax/native-protocol#61).

Related finding while fixing CI: the build had never actually resolved this snapshot — ci/run-tests.sh wasn't running install-snapshots.sh at all, so every CI run failed at dependency resolution. That's fixed now (402d1bf, 4885827, 6b9ac8f) and CI installs the snapshot from the PR #61 branch. One heads-up: I initially pointed it at your fork's cep-59 branch (which the PR description referenced), but that copy has Frame.forResponse stubbed out with UnsupportedOperationException, which failed the graph unit tests — you may want to update or remove that branch so nothing else picks it up.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Don't worry about CI or install-snapshots.sh. You can revert these changes

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reverted the CI changes in 3234760.

</dependency>
</dependencies>
</dependencyManagement>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1041,7 +1041,15 @@ public enum DefaultDriverOption implements DriverOption {
*
* <p>Value-Type: boolean
*/
ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses");
ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"),
/**
* Whether to register for GRACEFUL_DISCONNECT events from the server (CEP-59). When enabled and
* the server advertises support, the driver will gracefully drain connections when a node shuts
* down.
*
* <p>Value-type: boolean
*/
GRACEFUL_DISCONNECT_ENABLED("advanced.connection.graceful-disconnect-enabled");

private final String path;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -272,6 +272,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
map.put(TypedDriverOption.CONNECTION_MAX_REQUESTS, 1024);
map.put(TypedDriverOption.CONNECTION_MAX_ORPHAN_REQUESTS, 256);
map.put(TypedDriverOption.CONNECTION_WARN_INIT_ERROR, true);
map.put(TypedDriverOption.GRACEFUL_DISCONNECT_ENABLED, true);
map.put(TypedDriverOption.RECONNECT_ON_INIT, false);
map.put(TypedDriverOption.RECONNECTION_POLICY_CLASS, "ExponentialReconnectionPolicy");
map.put(TypedDriverOption.RECONNECTION_BASE_DELAY, Duration.ofSeconds(1));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -914,6 +914,9 @@ public String toString() {
new TypedDriverOption<>(
DefaultDriverOption.ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES, GenericType.BOOLEAN);

public static final TypedDriverOption<Boolean> GRACEFUL_DISCONNECT_ENABLED =
new TypedDriverOption<>(DefaultDriverOption.GRACEFUL_DISCONNECT_ENABLED, GenericType.BOOLEAN);

/**
* Ordered preference list of remote dcs optionally supplied for automatic failover and included
* in query plan. This feature is enabled only when max-nodes-per-remote-dc is greater than 0.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,7 @@ public enum DefaultNodeMetric implements NodeMetric {
SPECULATIVE_EXECUTIONS("speculative-executions"),
CONNECTION_INIT_ERRORS("errors.connection.init"),
AUTHENTICATION_ERRORS("errors.connection.auth"),
GRACEFUL_DISCONNECTS("pool.graceful-disconnects"),

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done; the counter is now initialized in all three backends (DropwizardNodeMetricUpdater, MicrometerNodeMetricUpdater, MicroProfileNodeMetricUpdater) in 618a7f8, and incremented when a GRACEFUL_DISCONNECT event is received on one of the node's pooled connections (ChannelPool query-connection callback, 442561d). It's documented in reference.conf, covered by the zero-value assertions in the three metrics ITs and by ChannelPoolGracefulDisconnectTest, and the new GracefulDisconnectIT exercises the session-level counter end to end against a real drain. I also verified both counters increment during manual drain runs on 2- and 3-node ccm clusters.

;

private static final Map<String, DefaultNodeMetric> BY_PATH = sortByPath();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ public enum DefaultSessionMetric implements SessionMetric {
THROTTLING_QUEUE_SIZE("throttling.queue-size"),
THROTTLING_ERRORS("throttling.errors"),
CQL_PREPARED_CACHE_SIZE("cql-prepared-cache-size"),
GRACEFUL_DISCONNECTS("graceful-disconnects"),
Comment thread
Shanzita marked this conversation as resolved.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need integration tests and manual testing for metrics

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

and the actual implementation of incrementing the metric

@Shanzita Shanzita Aug 26, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All three parts are done now:

  • Incrementing (442561d): the session counter increments wherever a GRACEFUL_DISCONNECT event is received — the pool's query-connection callback and the control connection. The node counter (pool.graceful-disconnects) increments for events on that node's pooled connections.
  • Initialization (618a7f8): both counters are initialized in all three backends (Dropwizard, Micrometer, MicroProfile) and documented in reference.conf; the three metrics ITs assert they exist as zero-valued counters, and ControlConnectionEventsTest / ChannelPoolGracefulDisconnectTest verify the increments at the unit level.
  • Integration + manual testing: the new GracefulDisconnectIT (c146d19) asserts this counter goes above zero during a real nodetool drain under load. I also verified both counters manually against a CASSANDRA-21191 server build on 2-node and 3-node ccm clusters — the drain runs finished with the event observed, counters incremented, and 0 disruptive exceptions.

;

private static final Map<String, DefaultSessionMetric> BY_PATH = sortByPath();
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -106,6 +106,12 @@ public class ChannelFactory {
*/
@VisibleForTesting volatile String productType;

@VisibleForTesting volatile boolean serverSupportsGracefulDisconnect;

public boolean isGracefulDisconnectSupported() {
return serverSupportsGracefulDisconnect;
}

public ChannelFactory(InternalDriverContext context) {
this.logPrefix = context.getSessionName();
this.context = context;
Expand Down Expand Up @@ -232,6 +238,12 @@ private void connect(
ConsistencyLevel.LOCAL_QUORUM.name()));
}
}
if (!serverSupportsGracefulDisconnect && supportedOptions != null) {
List<String> gdValues = supportedOptions.get(GracefulDisconnectEvent.EVENT_TYPE);
if (gdValues != null && gdValues.contains("true")) {
serverSupportsGracefulDisconnect = true;
}

@SiyaoIsHiding SiyaoIsHiding Aug 24, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It should be tracked per connection

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reworked in 442561d exactly along these lines — capability is now tracked per connection:

  • The global serverSupportsGracefulDisconnect flag is removed from ChannelFactory entirely.
  • Any channel that intends to register for GRACEFUL_DISCONNECT always runs the OPTIONS step, and ProtocolInitHandler filters the REGISTER against that channel's own SUPPORTED response — so in a mixed-version cluster each node negotiates independently, and a reconnect re-negotiates. When the feature is disabled in config there's no extra round-trip.
  • ChannelPool now requests the event based on config alone and relies on the per-channel filtering, so the first-contact-point ordering issue is gone.
  • If a server advertises the capability but rejects the REGISTER, the driver retries once without the event type instead of failing channel init.

Covered by ProtocolInitHandlerGracefulDisconnectTest (advertised / not advertised / advertised-as-false / rejection-retry cases) and verified in a 3-node manual drain run.

}
resultFuture.complete(driverChannel);
} else {
Throwable error = connectFuture.cause();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
/*
* 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 com.datastax.oss.driver.internal.core.channel;

import com.datastax.oss.driver.api.core.metadata.Node;
import net.jcip.annotations.Immutable;

/**
* This event indicates that the server is shutting down gracefully and the driver should:
*
* <ul>
* <li>Stop sending new requests on the affected connection
* <li>Allow in-flight requests to complete
* <li>Begin reconnection attempts with exponential backoff
* </ul>
*
* <p>This is part of CEP-59: Graceful Disconnect – In-Band Connection Draining for Node Shutdown.
*/
@Immutable
public class GracefulDisconnectEvent {

/** The event type string as defined in the native protocol. */
public static final String EVENT_TYPE = "GRACEFUL_DISCONNECT";

/** The node that sent the graceful disconnect event. */
public final Node node;

/** The channel that received the graceful disconnect event. */
public final DriverChannel channel;

public GracefulDisconnectEvent(Node node, DriverChannel channel) {
this.node = node;
this.channel = channel;
}

@Override
public String toString() {
return "GracefulDisconnectEvent{node=" + node + ", channel=" + channel + '}';
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import com.datastax.oss.protocol.internal.Frame;
import com.datastax.oss.protocol.internal.Message;
import com.datastax.oss.protocol.internal.request.Query;
import com.datastax.oss.protocol.internal.response.Event;
import com.datastax.oss.protocol.internal.response.result.SetKeyspace;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFuture;
Expand Down Expand Up @@ -218,6 +219,15 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception

if (streamId < 0) {
Message event = responseFrame.message;
if (event instanceof Event
&& GracefulDisconnectEvent.EVENT_TYPE.equals(((Event) event).type)) {
LOG.debug("[{}] Received GRACEFUL_DISCONNECT, initiating graceful drain", logPrefix);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you confirm, on receiving a graceful disconnect event, will line 226 and line 232 both logs Received event graceful disconnect? If so, line 226 is redundant.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed, both lines logged it (the drain-specific debug plus the generic received-event log). Removed the redundant one in 86d0fad; the generic log and the logs inside startGracefulShutdown cover it.

startGracefulShutdown(ctx);
if (eventCallback != null) {
eventCallback.onEvent(event);
}
return;
}
if (eventCallback == null) {
LOG.debug("[{}] Received event {} but no callback was registered", logPrefix, event);
} else {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,9 @@
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import net.jcip.annotations.NotThreadSafe;
import org.slf4j.Logger;
Expand Down Expand Up @@ -183,12 +185,36 @@ Message getRequest() {
case AUTH_RESPONSE:
return request = new AuthResponse(authResponseToken);
case REGISTER:
return request = new Register(options.eventTypes);
return request = new Register(filterSupportedEventTypes());
default:
throw new AssertionError("unhandled step: " + step);
}
}

/**
* Filters the requested event types to only include those supported by the server.
*
* <p>Specifically, GRACEFUL_DISCONNECT is only included if the server advertises support for it
* in the SUPPORTED message response.
*/
private List<String> filterSupportedEventTypes() {
List<String> filteredEventTypes = new ArrayList<>(options.eventTypes);

// Check if GRACEFUL_DISCONNECT is in the requested event types
if (filteredEventTypes.contains(GracefulDisconnectEvent.EVENT_TYPE)) {
// Get the supported options from the channel attribute (set during OPTIONS step)
Map<String, List<String>> supportedOptions = channel.attr(DriverChannel.OPTIONS_KEY).get();

// Only include GRACEFUL_DISCONNECT if the server supports it
if (supportedOptions == null
|| !supportedOptions.containsKey(GracefulDisconnectEvent.EVENT_TYPE)) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Supported options should be tracked per connection

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 442561d together with the ChannelFactory rework (see my reply on that thread for the full design): the factory fallback is gone, channels that request GRACEFUL_DISCONNECT always run the OPTIONS step, and filterSupportedEventTypes consults only this channel's own SUPPORTED response — so later channels and reconnects each negotiate for themselves.

One correction on the EVENTS point: the CEP-59 server advertises a top-level GRACEFUL_DISCONNECT key in SUPPORTED, not an entry under EVENTS — see OptionsMessage.execute() and InitialConnectionHandler in apache/cassandra#4953 (the pre-STARTUP path even sends GRACEFUL_DISCONNECT: ["false"] when the feature is disabled, which the driver's parsing handles). So the key check itself was correct; the real issues were the caching and the skipped OPTIONS, both fixed now.

filteredEventTypes.remove(GracefulDisconnectEvent.EVENT_TYPE);
}
}

return filteredEventTypes;
}

@Override
void send() {
stepNumber++;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import com.datastax.oss.driver.internal.core.channel.DriverChannel;
import com.datastax.oss.driver.internal.core.channel.DriverChannelOptions;
import com.datastax.oss.driver.internal.core.channel.EventCallback;
import com.datastax.oss.driver.internal.core.channel.GracefulDisconnectEvent;
import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
import com.datastax.oss.driver.internal.core.metadata.DefaultTopologyMonitor;
import com.datastax.oss.driver.internal.core.metadata.DistanceEvent;
Expand Down Expand Up @@ -178,8 +179,8 @@ public void onEvent(Message eventMessage) {
if (!(eventMessage instanceof Event)) {
LOG.warn("[{}] Unsupported event class: {}", logPrefix, eventMessage.getClass().getName());
} else {
LOG.debug("[{}] Processing incoming event {}", logPrefix, eventMessage);
Event event = (Event) eventMessage;
LOG.debug("[{}] Processing incoming event {}", logPrefix, eventMessage);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pls revert the reversion of these two lines

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reverted in 86d0fad.

switch (event.type) {
case ProtocolConstants.EventType.TOPOLOGY_CHANGE:
processTopologyChange(event);
Expand All @@ -190,6 +191,9 @@ public void onEvent(Message eventMessage) {
case ProtocolConstants.EventType.SCHEMA_CHANGE:
processSchemaChange(event);
break;
case GracefulDisconnectEvent.EVENT_TYPE:
processGracefulDisconnect();
break;
default:
LOG.warn("[{}] Unsupported event type: {}", logPrefix, event.type);
}
Expand Down Expand Up @@ -242,6 +246,28 @@ private void processSchemaChange(Event event) {
});
}

private void processGracefulDisconnect() {
LOG.info(
"[{}] Received GRACEFUL_DISCONNECT event on control connection, "
+ "the server is shutting down gracefully",
logPrefix);
// Fire an internal event to notify other components (particularly the ChannelPool)
DriverChannel currentChannel = channel;
if (currentChannel != null) {
context
.getMetadataManager()
.getMetadata()
.findNode(currentChannel.getEndPoint())
.ifPresent(
node ->
context.getEventBus().fire(new GracefulDisconnectEvent(node, currentChannel)));
}
// The control connection will handle reconnection automatically when the channel closes.
// The ChannelPool will close all its channels when it receives the GracefulDisconnectEvent,
// which will cause the NodeStateManager to set the node to DOWN state and trigger the
// LoadBalancingPolicy to remove it from the live set.
}

private class SingleThreaded {
private final InternalDriverContext context;
private final DriverConfig config;
Expand Down Expand Up @@ -292,7 +318,13 @@ private void init(
}
initWasCalled = true;
try {
ImmutableList<String> eventTypes = buildEventTypes(listenToClusterEvents);
boolean gracefulDisconnectEnabled =
context
.getConfig()
.getDefaultProfile()
.getBoolean(DefaultDriverOption.GRACEFUL_DISCONNECT_ENABLED, true);
ImmutableList<String> eventTypes =
buildEventTypes(listenToClusterEvents, gracefulDisconnectEnabled);
LOG.debug("[{}] Initializing with event types {}", logPrefix, eventTypes);
channelOptions =
DriverChannelOptions.builder()
Expand Down Expand Up @@ -606,14 +638,18 @@ private boolean isAuthFailure(Throwable error) {
return true;
}

private static ImmutableList<String> buildEventTypes(boolean listenClusterEvents) {
private static ImmutableList<String> buildEventTypes(
boolean listenClusterEvents, boolean gracefulDisconnectEnabled) {
ImmutableList.Builder<String> builder = ImmutableList.builder();
builder.add(ProtocolConstants.EventType.SCHEMA_CHANGE);
if (listenClusterEvents) {
builder
.add(ProtocolConstants.EventType.STATUS_CHANGE)
.add(ProtocolConstants.EventType.TOPOLOGY_CHANGE);
}
if (gracefulDisconnectEnabled) {
builder.add(GracefulDisconnectEvent.EVENT_TYPE);
}
return builder.build();
}
}
Loading