You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "Shanzita (via GitHub)" <gi...@apache.org> on 2026/09/09 19:50:26 UTC
Re: [PR] CASSJAVA-124: Add GRACEFUL_DISCONNECT support (CEP-59) [cassandra-java-driver]
Shanzita commented on code in PR #2091:
URL: https://github.com/apache/cassandra-java-driver/pull/2091#discussion_r3972410195
##########
integration-tests/src/test/java/com/datastax/oss/driver/core/connection/GracefulDisconnectIT.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.core.connection;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import com.codahale.metrics.Counter;
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
+import com.datastax.oss.driver.api.core.metrics.DefaultSessionMetric;
+import com.datastax.oss.driver.api.testinfra.ccm.CustomCcmRule;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendRequirement;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendType;
+import com.datastax.oss.driver.api.testinfra.session.SessionUtils;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+/**
+ * Exercises CEP-59 graceful disconnect (CASSANDRA-21191) against a real cluster: when a node is
+ * drained, it sends a GRACEFUL_DISCONNECT event on every registered connection before closing the
+ * transport, and the driver must drain its pool to that node and fail over without surfacing any
+ * exception to the application.
+ *
+ * <p>Requires a server that implements the GRACEFUL_DISCONNECT event; on older servers the test is
+ * skipped by the version requirement below.
+ */
+public class GracefulDisconnectIT {
+
+ @ClassRule
+ public static final CustomCcmRule CCM_RULE = CustomCcmRule.builder().withNodes(2).build();
+
+ private static final String QUERY = "SELECT * FROM system.local";
+
+ @BackendRequirement(
+ type = BackendType.CASSANDRA,
+ minInclusive = "7.0",
+ description = "Graceful disconnect (CEP-59 / CASSANDRA-21191) requires server-side support")
+ @Test
+ public void should_fail_over_without_disruption_when_node_drains() throws Exception {
+ DriverConfigLoader loader =
+ SessionUtils.configLoaderBuilder()
+ .withStringList(
+ DefaultDriverOption.METRICS_SESSION_ENABLED,
+ ImmutableList.of(DefaultSessionMetric.GRACEFUL_DISCONNECTS.getPath()))
+ .build();
+
+ try (CqlSession session = SessionUtils.newSession(CCM_RULE, loader)) {
+
+ // Sanity check before the drain:
+ session.execute(QUERY);
+
+ // Steady query load for the whole duration of the test, collecting any exception that
+ // reaches the application:
+ AtomicLong successes = new AtomicLong();
+ List<Throwable> failures = new CopyOnWriteArrayList<>();
+ AtomicBoolean stopped = new AtomicBoolean();
+ Thread load =
+ new Thread(
+ () -> {
+ while (!stopped.get()) {
Review Comment:
Both done in b023ee4d — added `Thread.sleep(5)` to the load loop (with interrupt handling so the `finally` still shuts the thread down deterministically), and re-ran the test manually against a 2-node ccm cluster built from apache/cassandra#4953: passed in ~60s.
And good instinct on the no-support question — as written it **would have failed**, for two stacked reasons:
1. `@BackendRequirement` was on the method, but `CustomCcmRule` is a `@ClassRule` and only sees the class-level `Description`, so the version gate was never actually evaluated — on a pre-7.0 server ccm would start and the test would run and time out on the metric await. Moved the annotation to the class, which also means ccm doesn't start at all when the requirement isn't met.
2. Even on a CEP-59 server, `graceful_disconnect_enabled` defaults to **false** in Config.java, so a stock 7.0 cluster would never send the event. The rule now sets `.withCassandraConfiguration("graceful_disconnect_enabled", true)` explicitly, so the test is self-contained.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/channel/InFlightHandlerTest.java:
##########
@@ -644,6 +647,136 @@ private void addToPipeline() {
addToPipelineWithEventCallback(null);
}
+ @Test
+ public void should_initiate_graceful_drain_on_graceful_disconnect_event() {
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+ when(streamIds.acquire()).thenReturn(42);
+ MockResponseCallback responseCallback = new MockResponseCallback();
+ channel
+ .writeAndFlush(
+ new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback))
+ .awaitUninterruptibly();
+
+ // When
+ GracefulDisconnectEvent gracefulDisconnectEvent =
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent();
Review Comment:
Done across the PR in b023ee4d. One caveat: in `ControlConnectionEventsTest` and `ChannelPoolGracefulDisconnectTest` both the driver's internal `metadata.GracefulDisconnectEvent` and the protocol's `response.event.GracefulDisconnectEvent` appear, and Java has no import aliases — so those two files import the driver one and keep a single qualified `new` for the protocol one. Everywhere else it's short names only.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/channel/InFlightHandlerTest.java:
##########
@@ -644,6 +647,136 @@ private void addToPipeline() {
addToPipelineWithEventCallback(null);
}
+ @Test
+ public void should_initiate_graceful_drain_on_graceful_disconnect_event() {
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+ when(streamIds.acquire()).thenReturn(42);
+ MockResponseCallback responseCallback = new MockResponseCallback();
+ channel
+ .writeAndFlush(
+ new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback))
+ .awaitUninterruptibly();
+
+ // When
+ GracefulDisconnectEvent gracefulDisconnectEvent =
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent();
+ Frame eventFrame =
+ Frame.forResponse(
+ DefaultProtocolVersion.V4.getCode(),
+ -1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ gracefulDisconnectEvent);
+ writeInboundFrame(eventFrame);
+
+ // Then
+ // channel not closed yet because there is a pending request
+ assertThat(channel.closeFuture()).isNotDone();
+ // callback was still notified
+ verify(eventCallback).onEvent(gracefulDisconnectEvent);
+ // new writes should be refused
+ ChannelFuture otherWriteFuture =
+ channel.writeAndFlush(
+ new DriverChannel.RequestMessage(
+ QUERY, false, Frame.NO_PAYLOAD, new MockResponseCallback()));
+ assertThat(otherWriteFuture)
+ .isFailed(e -> assertThat(e).isInstanceOf(IllegalStateException.class));
+
+ // When the pending request completes
+ Frame requestFrame = readOutboundFrame();
+ writeInboundFrame(requestFrame, Void.INSTANCE);
+
+ // Then the channel closes
+ assertThat(channel.closeFuture()).isSuccess();
+ }
+
+ @Test
+ public void should_close_immediately_on_graceful_disconnect_if_no_pending() {
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+
+ // When
+ GracefulDisconnectEvent gracefulDisconnectEvent =
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent();
+ Frame eventFrame =
+ Frame.forResponse(
+ DefaultProtocolVersion.V4.getCode(),
+ -1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ gracefulDisconnectEvent);
+ writeInboundFrame(eventFrame);
+
+ // Then
+ assertThat(channel.closeFuture()).isSuccess();
+ verify(eventCallback).onEvent(gracefulDisconnectEvent);
+ }
+
+ @Test
+ public void should_handle_duplicate_graceful_disconnect_events() {
+ // The server-side CEP-59 implementation is still evolving; a node might emit the event more
Review Comment:
Removed.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/channel/InFlightHandlerTest.java:
##########
@@ -644,6 +647,136 @@ private void addToPipeline() {
addToPipelineWithEventCallback(null);
}
+ @Test
+ public void should_initiate_graceful_drain_on_graceful_disconnect_event() {
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+ when(streamIds.acquire()).thenReturn(42);
+ MockResponseCallback responseCallback = new MockResponseCallback();
+ channel
+ .writeAndFlush(
+ new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback))
+ .awaitUninterruptibly();
+
+ // When
+ GracefulDisconnectEvent gracefulDisconnectEvent =
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent();
+ Frame eventFrame =
+ Frame.forResponse(
+ DefaultProtocolVersion.V4.getCode(),
+ -1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ gracefulDisconnectEvent);
+ writeInboundFrame(eventFrame);
+
+ // Then
+ // channel not closed yet because there is a pending request
+ assertThat(channel.closeFuture()).isNotDone();
+ // callback was still notified
+ verify(eventCallback).onEvent(gracefulDisconnectEvent);
+ // new writes should be refused
+ ChannelFuture otherWriteFuture =
+ channel.writeAndFlush(
+ new DriverChannel.RequestMessage(
+ QUERY, false, Frame.NO_PAYLOAD, new MockResponseCallback()));
+ assertThat(otherWriteFuture)
+ .isFailed(e -> assertThat(e).isInstanceOf(IllegalStateException.class));
+
+ // When the pending request completes
+ Frame requestFrame = readOutboundFrame();
+ writeInboundFrame(requestFrame, Void.INSTANCE);
+
+ // Then the channel closes
+ assertThat(channel.closeFuture()).isSuccess();
+ }
+
+ @Test
+ public void should_close_immediately_on_graceful_disconnect_if_no_pending() {
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+
+ // When
+ GracefulDisconnectEvent gracefulDisconnectEvent =
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent();
+ Frame eventFrame =
+ Frame.forResponse(
+ DefaultProtocolVersion.V4.getCode(),
+ -1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ gracefulDisconnectEvent);
+ writeInboundFrame(eventFrame);
+
+ // Then
+ assertThat(channel.closeFuture()).isSuccess();
+ verify(eventCallback).onEvent(gracefulDisconnectEvent);
+ }
+
+ @Test
+ public void should_handle_duplicate_graceful_disconnect_events() {
+ // The server-side CEP-59 implementation is still evolving; a node might emit the event more
+ // than once (e.g. once per registered connection, or on a drain retry). The second event must
+ // not disrupt the drain already in progress.
+ // Given
+ EventCallback eventCallback = mock(EventCallback.class);
+ addToPipelineWithEventCallback(eventCallback);
+ when(streamIds.acquire()).thenReturn(42);
+ MockResponseCallback responseCallback = new MockResponseCallback();
+ channel
+ .writeAndFlush(
+ new DriverChannel.RequestMessage(QUERY, false, Frame.NO_PAYLOAD, responseCallback))
+ .awaitUninterruptibly();
+
+ // When: the same event is received twice while a request is still pending
+ for (int i = 0; i < 2; i++) {
+ Frame eventFrame =
+ Frame.forResponse(
+ DefaultProtocolVersion.V4.getCode(),
+ -1,
+ null,
+ Collections.emptyMap(),
+ Collections.emptyList(),
+ new com.datastax.oss.protocol.internal.response.event.GracefulDisconnectEvent());
+ writeInboundFrame(eventFrame);
+ }
+
+ // Then: still draining, not closed abruptly
+ assertThat(channel.closeFuture()).isNotDone();
+ verify(eventCallback, times(2)).onEvent(any());
+
+ // When the pending request completes, the drain finishes normally
+ Frame requestFrame = readOutboundFrame();
+ writeInboundFrame(requestFrame, Void.INSTANCE);
+ assertThat(channel.closeFuture()).isSuccess();
+ }
+
+ @Test
+ public void should_handle_graceful_disconnect_without_event_callback() {
+ // Given
Review Comment:
Removed as well.
##########
core/src/test/java/com/datastax/oss/driver/internal/core/channel/ProtocolInitHandlerGracefulDisconnectTest.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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 static com.datastax.oss.driver.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.datastax.oss.driver.api.core.DefaultProtocolVersion;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
+import com.datastax.oss.driver.api.core.config.DriverConfig;
+import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.metadata.EndPoint;
+import com.datastax.oss.driver.internal.core.DefaultProtocolVersionRegistry;
+import com.datastax.oss.driver.internal.core.ProtocolVersionRegistry;
+import com.datastax.oss.driver.internal.core.TestResponses;
+import com.datastax.oss.driver.internal.core.context.InternalDriverContext;
+import com.datastax.oss.driver.internal.core.metadata.TestNodeFactory;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
+import com.datastax.oss.protocol.internal.Frame;
+import com.datastax.oss.protocol.internal.ProtocolConstants;
+import com.datastax.oss.protocol.internal.request.Options;
+import com.datastax.oss.protocol.internal.request.Register;
+import com.datastax.oss.protocol.internal.request.Startup;
+import com.datastax.oss.protocol.internal.response.Ready;
+import com.datastax.oss.protocol.internal.response.Supported;
+import io.netty.channel.ChannelFuture;
+import java.net.InetSocketAddress;
+import java.time.Duration;
+import java.util.List;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+/**
+ * Coverage for GRACEFUL_DISCONNECT (CEP-59) registration during channel initialization: support is
+ * checked against each channel's own SUPPORTED response, and the event type is only included in
+ * REGISTER if the server advertises it.
+ */
+public class ProtocolInitHandlerGracefulDisconnectTest extends ChannelHandlerTestBase {
+
+ private static final long QUERY_TIMEOUT_MILLIS = 100L;
+ private static final EndPoint END_POINT = TestNodeFactory.newEndPoint(1);
+ private static final Supported SUPPORTED_WITH_GRACEFUL_DISCONNECT =
+ new Supported(
+ ImmutableMap.of(
+ ProtocolConstants.EventType.GRACEFUL_DISCONNECT,
+ ImmutableList.of("true"),
+ "CQL_VERSION",
+ ImmutableList.of("3.4.7")));
+ private static final Supported SUPPORTED_WITHOUT_GRACEFUL_DISCONNECT =
+ new Supported(ImmutableMap.of("CQL_VERSION", ImmutableList.of("3.4.7")));
+
+ @Mock private InternalDriverContext internalDriverContext;
+ @Mock private DriverConfig driverConfig;
+ @Mock private DriverExecutionProfile defaultProfile;
+
+ private final ProtocolVersionRegistry protocolVersionRegistry =
+ new DefaultProtocolVersionRegistry("test");
+ private HeartbeatHandler heartbeatHandler;
+
+ @Before
+ @Override
+ public void setup() {
+ super.setup();
+ MockitoAnnotations.initMocks(this);
+ when(internalDriverContext.getConfig()).thenReturn(driverConfig);
+ when(driverConfig.getDefaultProfile()).thenReturn(defaultProfile);
+ when(defaultProfile.getDuration(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT))
+ .thenReturn(Duration.ofMillis(QUERY_TIMEOUT_MILLIS));
+ when(defaultProfile.getDuration(DefaultDriverOption.HEARTBEAT_INTERVAL))
+ .thenReturn(Duration.ofSeconds(30));
+ when(internalDriverContext.getProtocolVersionRegistry()).thenReturn(protocolVersionRegistry);
+
+ channel
+ .pipeline()
+ .addLast(
+ ChannelFactory.INFLIGHT_HANDLER_NAME,
+ new InFlightHandler(
+ DefaultProtocolVersion.V4,
+ new StreamIdGenerator(100),
+ Integer.MAX_VALUE,
+ 100,
+ channel.newPromise(),
+ null,
+ "test"));
+
+ heartbeatHandler = new HeartbeatHandler(defaultProfile);
+ }
+
+ private ChannelFuture connectWithEvents(List<String> eventTypes) {
+ DriverChannelOptions driverChannelOptions =
+ DriverChannelOptions.builder().withEvents(eventTypes, mock(EventCallback.class)).build();
+ channel
+ .pipeline()
+ .addLast(
+ ChannelFactory.INIT_HANDLER_NAME,
+ new ProtocolInitHandler(
+ internalDriverContext,
+ DefaultProtocolVersion.V4,
+ null,
+ END_POINT,
+ driverChannelOptions,
+ heartbeatHandler,
+ true));
+ return channel.connect(new InetSocketAddress("localhost", 9042));
+ }
+
+ /** Completes the OPTIONS and STARTUP steps. */
+ private void initUntilAfterClusterName(Supported supportedResponse) {
+ Frame optionsFrame = readOutboundFrame();
+ assertThat(optionsFrame.message).isInstanceOf(Options.class);
+ writeInboundFrame(optionsFrame, supportedResponse);
+ Frame startupFrame = readOutboundFrame();
+ assertThat(startupFrame.message).isInstanceOf(Startup.class);
+ writeInboundFrame(startupFrame, new Ready());
+ writeInboundFrame(readOutboundFrame(), TestResponses.clusterNameResponse("someClusterName"));
+ }
+
+ @Test
+ public void should_register_graceful_disconnect_when_advertised_in_supported() {
+ ChannelFuture connectFuture =
+ connectWithEvents(
+ ImmutableList.of("STATUS_CHANGE", ProtocolConstants.EventType.GRACEFUL_DISCONNECT));
Review Comment:
Done — `ProtocolConstants.EventType.*` everywhere in this file now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org