natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge dfad626691 into 10557d7ffe

This commit is contained in:
Rishabh Saraswat 2026-07-31 14:32:31 +08:00 committed by GitHub
parent 10557d7ffe dfad626691
commit 3af90b1196
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
21 changed files with 931 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
conf/cassandra.yaml
View File

@ -1102,6 +1102,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.

14
conf/cassandra_latest.yaml
View File

@ -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.

91
doc/modules/cassandra/pages/managing/operating/graceful_disconnect.adoc Normal file
View File

@ -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.

4
src/java/org/apache/cassandra/config/Config.java
View File

@ -143,6 +143,10 @@ public class Config
/** 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");

15
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -2685,6 +2685,21 @@ public class DatabaseDescriptor
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 boolean getGracefulDisconnectEnabled()
{
return conf.graceful_disconnect_enabled;
}
public static long getReadRpcTimeout(TimeUnit unit)
{
return conf.read_request_timeout.to(unit);

23
src/java/org/apache/cassandra/metrics/ClientMetrics.java
View File

@ -71,6 +71,10 @@ public final class ClientMetrics
@VisibleForTesting
Gauge<Integer> connectedNativeClients;
public AtomicInteger connectionsDraining;
public Meter forcedDisconnects;
@VisibleForTesting
Gauge<Integer> encryptedConnectedNativeClients;
@ -173,6 +177,21 @@ public final class ClientMetrics
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 final class ClientMetrics
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)

6
src/java/org/apache/cassandra/service/NativeTransportService.java
View File

@ -37,6 +37,7 @@ import org.apache.cassandra.utils.NativeLibrary;
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 @@ public class NativeTransportService
return server;
}
public ChannelGroup getChannelsSubscribedToGracefulDisconnect()
{
return server.getChannelsSubscribedToGracefulDisconnect();
}
public void clearConnectionHistory()
{
server.clearConnectionHistory();

150
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -40,13 +40,17 @@ import java.util.TreeMap;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@ -158,6 +162,7 @@ import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.Replicas;
import org.apache.cassandra.locator.SnitchAdapter;
import org.apache.cassandra.locator.SystemReplicas;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.metrics.Sampler;
import org.apache.cassandra.metrics.SamplingManager;
import org.apache.cassandra.metrics.StorageMetrics;
@ -219,7 +224,9 @@ import org.apache.cassandra.tcm.transformations.Register;
import org.apache.cassandra.tcm.transformations.Startup;
import org.apache.cassandra.tcm.transformations.Unregister;
import org.apache.cassandra.transport.ClientResourceLimits;
import org.apache.cassandra.transport.Event;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.transport.messages.EventMessage;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -239,6 +246,9 @@ import org.apache.cassandra.utils.progress.ProgressListener;
import org.apache.cassandra.utils.progress.jmx.JMXBroadcastExecutor;
import org.apache.cassandra.utils.progress.jmx.JMXProgressSupport;
import io.netty.channel.group.ChannelGroup;
import io.netty.util.AttributeKey;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
@ -292,9 +302,13 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
private static final Logger logger = LoggerFactory.getLogger(StorageService.class);
private static ReentrantLock drainLock = new ReentrantLock();
public static final int INDEFINITE = -1;
public static final int RING_DELAY_MILLIS = getRingDelay(); // delay after which we assume ring has stablized
static final AttributeKey<Consumer<EventMessage>> EVENT_DISPATCHER = AttributeKey.valueOf("EVTDISP");
{
PathUtils.setDeletionListener(path -> {
if (isDaemonSetupCompleted())
@ -617,7 +631,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
throw new IllegalStateException("No configured daemon");
}
daemon.stopNativeTransport(force);
gracefulDisconnect(() -> {
daemon.stopNativeTransport(force);
});
}
public boolean isNativeTransportRunning()
@ -715,7 +731,9 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
{
if (daemon == null)
throw new IllegalStateException("No configured daemon");
daemon.deactivate();
gracefulDisconnect(() -> {
daemon.deactivate();
});
}
// for testing only
@ -1199,6 +1217,27 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
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);
@ -3875,9 +3914,112 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
/**
* Shuts node off to writes, empties memtables and the commit log.
*/
public synchronized void drain() throws IOException, InterruptedException, ExecutionException
public void drain() throws IOException, InterruptedException, ExecutionException, TimeoutException
{
drain(false);
if (!drainLock.tryLock())
{
throw new IllegalStateException("Drain already in progress");
}
try
{
CountDownLatch drainComplete = new CountDownLatch(1);
AtomicReference<Exception> failure = new AtomicReference<>();
gracefulDisconnect(() -> {
try
{
drain(false);
}
catch (IOException | InterruptedException | ExecutionException e)
{
failure.set(e);
}
finally
{
drainComplete.countDown();
}
});
if (!drainComplete.await(DatabaseDescriptor.getGracefulDisconnectGracePeriod(), MILLISECONDS))
throw new TimeoutException("Timed out waiting for drain to complete after graceful disconnect");
Exception e = failure.get();
if (e instanceof IOException) throw (IOException) e;
if (e instanceof InterruptedException) throw (InterruptedException) e;
if (e instanceof ExecutionException) throw (ExecutionException) e;
if (e != null) throw new RuntimeException(e);
}
finally
{
drainLock.unlock();
}
}
private void gracefulDisconnect(Runnable defaultAction)
{
NativeTransportService nts = CassandraDaemon.instance.nativeTransportService();
if (nts == null || nts.getServer() == null)
{
defaultAction.run();
return;
}
nts.getServer().stopAcceptingNewConnections();
ChannelGroup channelGroup = nts.getServer().getChannelsSubscribedToGracefulDisconnect();
if (channelGroup == null || channelGroup.isEmpty())
{
defaultAction.run();
return;
}
gracefulDisconnect(defaultAction, channelGroup);
}
public void gracefulDisconnect(Runnable defaultAction, ChannelGroup channelGroup)
{
if (channelGroup == null || channelGroup.isEmpty())
{
defaultAction.run();
return;
}
AtomicBoolean actionStarted = new AtomicBoolean(false);
AtomicInteger connectedChannels = new AtomicInteger(channelGroup.size());
Runnable runOnceAction = () -> {
if (actionStarted.compareAndSet(false, true))
defaultAction.run();
};
Runnable onTimeout = () -> {
if (actionStarted.get())
return;
int remaining = connectedChannels.get();
if (remaining > 0)
{
ClientMetrics.instance.markForcedDisconnect(remaining);
channelGroup.close();
}
else
{
runOnceAction.run();
}
};
ScheduledFuture<?> timeoutTask = ScheduledExecutors.nonPeriodicTasks
.schedule(onTimeout, DatabaseDescriptor.getGracefulDisconnectGracePeriod(), MILLISECONDS);
EventMessage gracefulDisconnectEventMessage = new EventMessage(new Event.GracefulDisconnect());
channelGroup.forEach(channel -> {
Consumer<EventMessage> dispatcher = channel.attr(EVENT_DISPATCHER).get();
if (dispatcher != null)
dispatcher.accept(gracefulDisconnectEventMessage);
channel.closeFuture().addListener(future -> {
ClientMetrics.instance.decrementConnectionsDraining();
if (connectedChannels.decrementAndGet() == 0)
{
timeoutTask.cancel(false);
runOnceAction.run();
}
});
ClientMetrics.instance.incrementConnectionsDraining();
});
if (connectedChannels.get() == 0)
{
timeoutTask.cancel(false);
runOnceAction.run();
}
}
protected synchronized void drain(boolean isFinalShutdown) throws IOException, InterruptedException, ExecutionException

7
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -640,7 +640,7 @@ public interface StorageServiceMBean extends NotificationEmitter
public String getDrainProgress();
/** makes node unavailable for writes, flushes memtables and replays commitlog. */
public void drain() throws IOException, InterruptedException, ExecutionException;
public void drain() throws IOException, InterruptedException, ExecutionException, TimeoutException;
/**
* Truncates (deletes) the given table from the provided keyspace.
@ -795,6 +795,11 @@ public interface StorageServiceMBean extends NotificationEmitter
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();

2
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -765,7 +765,7 @@ public class NodeProbe implements AutoCloseable
}
}
public void drain() throws IOException, InterruptedException, ExecutionException
public void drain() throws IOException, InterruptedException, ExecutionException, TimeoutException
{
ssProxy.drain();
}

3
src/java/org/apache/cassandra/tools/nodetool/Drain.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.tools.nodetool;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.apache.cassandra.tools.NodeProbe;
@ -34,7 +35,7 @@ public class Drain extends AbstractCommand
try
{
probe.drain();
} catch (IOException | InterruptedException | ExecutionException e)
} catch (IOException | InterruptedException | TimeoutException | ExecutionException e)
{
throw new RuntimeException("Error occurred during flushing", e);
}

27
src/java/org/apache/cassandra/transport/Event.java
View File

@ -36,7 +36,8 @@ public abstract class Event
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 abstract class Event
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 abstract class Event
&& 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();
}
}
}

2
src/java/org/apache/cassandra/transport/InitialConnectionHandler.java
View File

@ -28,6 +28,7 @@ import java.util.Map;
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;
@ -91,6 +92,7 @@ public class InitialConnectionHandler extends ByteToMessageDecoder
supportedOptions.put(StartupMessage.CQL_VERSION, cqlVersions);
supportedOptions.put(StartupMessage.COMPRESSION, compressions);
supportedOptions.put(StartupMessage.PROTOCOL_VERSIONS, ProtocolVersion.supportedVersions());
supportedOptions.put(StartupMessage.GRACEFUL_DISCONNECT, List.of(String.valueOf(DatabaseDescriptor.getGracefulDisconnectEnabled())));
SupportedMessage supported = new SupportedMessage(supportedOptions);
outbound = supported.encode(inbound.header.version, inbound.header.streamId);
ctx.writeAndFlush(outbound);

19
src/java/org/apache/cassandra/transport/Server.java
View File

@ -80,6 +80,7 @@ public class Server implements CassandraDaemon.Server
private static final boolean useEpoll = NativeTransportService.useEpoll();
private final ConnectionTracker connectionTracker;
private Channel bindChannel;
private final Connection.Factory connectionFactory = new Connection.Factory()
{
@ -126,6 +127,11 @@ public class Server implements CassandraDaemon.Server
Schema.instance.registerListener(notifier);
}
public ChannelGroup getChannelsSubscribedToGracefulDisconnect()
{
return connectionTracker.groups.get(Event.Type.GRACEFUL_DISCONNECT);
}
public void stop()
{
stop(false);
@ -149,6 +155,7 @@ public class Server implements CassandraDaemon.Server
// Configure the server.
ChannelFuture bindFuture = pipelineConfigurator.initializeChannel(workerGroup, socket, connectionFactory);
bindChannel = bindFuture.channel();
if (!bindFuture.awaitUninterruptibly().isSuccess())
throw new IllegalStateException(String.format("Failed to bind port %d on %s.", socket.getPort(), socket.getAddress().getHostAddress()),
bindFuture.cause());
@ -157,6 +164,16 @@ public class Server implements CassandraDaemon.Server
isRunning.set(true);
}
public void stopAcceptingNewConnections()
{
if (bindChannel != null && bindChannel.isOpen())
{
logger.info("Stopping native transport acceptor on {}", bindChannel.localAddress());
// syncUninterruptibly ensures we wait for the port to actually close
bindChannel.close().syncUninterruptibly();
}
}
public int countConnectedClients()
{
return connectionTracker.countConnectedClients();
@ -330,6 +347,8 @@ public class Server implements CassandraDaemon.Server
public void register(Event.Type type, Channel ch)
{
if (type == Event.Type.GRACEFUL_DISCONNECT && !DatabaseDescriptor.getGracefulDisconnectEnabled())
return;
groups.get(type).add(ch);
}

50
src/java/org/apache/cassandra/transport/SimpleClient.java
View File

@ -29,7 +29,7 @@ import java.util.Map;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.SynchronousQueue; // checkstyle: permit this import
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@ -82,7 +82,7 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.MessageToMessageEncoder;
import io.netty.handler.ssl.SslContext;
import io.netty.util.concurrent.Promise; // checkstyle: permit this import
import io.netty.util.concurrent.Promise;
import io.netty.util.concurrent.PromiseCombiner;
import io.netty.util.internal.logging.InternalLoggerFactory;
import io.netty.util.internal.logging.Slf4JLoggerFactory;
@ -100,6 +100,8 @@ public class SimpleClient implements Closeable
public static final int TIMEOUT_SECONDS = 10;
private final AtomicBoolean draining = new AtomicBoolean(false);
static
{
InternalLoggerFactory.setDefaultFactory(new Slf4JLoggerFactory());
@ -112,7 +114,7 @@ public class SimpleClient implements Closeable
private final EncryptionOptions.ClientEncryptionOptions encryptionOptions;
private final int largeMessageThreshold;
protected final ResponseHandler responseHandler = new ResponseHandler();
protected final ResponseHandler responseHandler = new ResponseHandler(this);
protected final Connection.Tracker tracker = new ConnectionTracker();
protected final ProtocolVersion version;
// We don't track connection really, so we don't need one Connection per channel
@ -238,8 +240,9 @@ public class SimpleClient implements Closeable
options.put(StartupMessage.COMPRESSION, "LZ4");
connection.setCompressor(Compressor.LZ4Compressor.instance);
}
if (version.isGreaterOrEqualTo(ProtocolVersion.V5))
options.put(StartupMessage.GRACEFUL_DISCONNECT, "GRACEFUL_DISCONNECT");
execute(new StartupMessage(options));
return this;
}
@ -324,6 +327,10 @@ public class SimpleClient implements Closeable
public Message.Response execute(Message.Request request, boolean throwOnErrorResponse)
{
if (draining.get())
{
throw new RuntimeException("Connection is draining (GRACEFUL_DISCONNECT received)");
}
try
{
request.attach(connection);
@ -639,6 +646,18 @@ public class SimpleClient implements Closeable
}
}
private void handleGracefulDisconnect()
{
draining.set(true);
ChannelFuture writeFuture = lastWriteFuture;
if (writeFuture != null)
writeFuture.addListener(f -> channel.close());
else
channel.close();
channel.closeFuture().addListener(f -> bootstrap.group().shutdownGracefully());
}
@ChannelHandler.Sharable
static class MessageBatchEncoder extends MessageToMessageEncoder<List<Message>>
{
@ -700,6 +719,14 @@ public class SimpleClient implements Closeable
@ChannelHandler.Sharable
static class ResponseHandler extends SimpleChannelInboundHandler<Message.Response>
{
private final SimpleClient client;
public ResponseHandler(SimpleClient client)
{
this.client = client;
}
public final BlockingQueue<Message.Response> responses = new SynchronousQueue<>(true);
public EventHandler eventHandler;
@ -719,8 +746,19 @@ public class SimpleClient implements Closeable
if (r instanceof EventMessage)
{
Event event = ((EventMessage) r).event;
if (event.type == Event.Type.GRACEFUL_DISCONNECT)
{
logger.info("Received GRACEFUL_DISCONNECT. Entering draining mode.");
if (eventHandler != null)
eventHandler.onEvent(event);
client.handleGracefulDisconnect();
return;
}
if (eventHandler != null)
eventHandler.onEvent(((EventMessage) r).event);
eventHandler.onEvent(event);
}
else
responses.put(r);
@ -899,4 +937,4 @@ public class SimpleClient implements Closeable
return futures.toArray(EMPTY_FUTURES_ARRAY);
}
}
}
}

3
src/java/org/apache/cassandra/transport/messages/OptionsMessage.java
View File

@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.transport.Compressor;
@ -73,6 +74,8 @@ public class OptionsMessage extends Message.Request
Map<String, List<String>> supported = new HashMap<String, List<String>>();
supported.put(StartupMessage.CQL_VERSION, cqlVersions);
supported.put(StartupMessage.COMPRESSION, compressions);
if (DatabaseDescriptor.getGracefulDisconnectEnabled())
supported.put(StartupMessage.GRACEFUL_DISCONNECT, List.of("true"));
supported.put(StartupMessage.PROTOCOL_VERSIONS, ProtocolVersion.supportedVersions());
return new SupportedMessage(supported);

1
src/java/org/apache/cassandra/transport/messages/StartupMessage.java
View File

@ -51,6 +51,7 @@ public class StartupMessage extends Message.Request
public static final String DRIVER_NAME = "DRIVER_NAME";
public static final String DRIVER_VERSION = "DRIVER_VERSION";
public static final String THROW_ON_OVERLOAD = "THROW_ON_OVERLOAD";
public static final String GRACEFUL_DISCONNECT = "GRACEFUL_DISCONNECT";
public static final Message.Codec<StartupMessage> codec = new Message.Codec<StartupMessage>()
{

464
test/distributed/org/apache/cassandra/distributed/test/GracefulDisconnectTest.java Normal file
View File

@ -0,0 +1,464 @@
/*
* 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 org.apache.cassandra.distributed.test;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.awaitility.Awaitility;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.metrics.ClientMetrics;
import org.apache.cassandra.service.CassandraDaemon;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.Event;
import org.apache.cassandra.transport.Message;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.transport.messages.OptionsMessage;
import org.apache.cassandra.transport.messages.ReadyMessage;
import org.apache.cassandra.transport.messages.RegisterMessage;
import org.apache.cassandra.transport.messages.StartupMessage;
import org.apache.cassandra.transport.messages.SupportedMessage;
import io.netty.channel.group.ChannelGroup;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
public class GracefulDisconnectTest
{
@BeforeClass
public static void setUp() throws IOException
{
DatabaseDescriptor.daemonInitialization();
}
public Cluster buildCluster(int nodeCount, boolean gracefulDisconnectEnabled) throws IOException
{
return Cluster
.build(nodeCount)
.withConfig(config ->
config
.with(Feature.NATIVE_PROTOCOL, Feature.GOSSIP)
.set("graceful_disconnect_enabled", gracefulDisconnectEnabled))
.start();
}
@Test
public void testGracefulDisconnectAdvertisedWhenEnabled() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042).build())
{
client.connect(false);
Message.Response response = client.execute(new OptionsMessage());
SupportedMessage supported = (SupportedMessage) response;
assertThat(supported.supported.containsKey(StartupMessage.GRACEFUL_DISCONNECT))
.as("GRACEFUL_DISCONNECT should be advertised in SUPPORTED when enabled")
.isTrue();
}
}
}
@Test
public void testGracefulDisconnectDoesNotAdvertisedWhenNotEnabled() throws IOException
{
try (Cluster cluster = buildCluster(1, false))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042).build())
{
client.connect(false);
SupportedMessage supported = (SupportedMessage) client.execute(new OptionsMessage());
assertThat(supported.supported.containsKey(StartupMessage.GRACEFUL_DISCONNECT))
.as("GRACEFUL_DISCONNECT should NOT be advertised in SUPPORTED when disabled")
.isFalse();
}
}
}
@Test
public void testSubscriptionViaREGISTER() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
client.connect(false);
Message.Response response = client.execute(
new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
assertThat(response).isInstanceOf(ReadyMessage.class);
int subscribedCount = cluster.get(1).callOnInstance(() ->
CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect()
.size());
assertThat(subscribedCount)
.as("One channel should be subscribed to GRACEFUL_DISCONNECT")
.isEqualTo(1);
}
}
}
@Test
public void testRegisterGracefulDisconnectRejectedOnV4() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042).protocolVersion(ProtocolVersion.V4).build())
{
client.connect(false);
assertThatThrownBy(() -> client.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT))))
.hasCauseInstanceOf(org.apache.cassandra.transport.ProtocolException.class);
int subscribedCount = cluster.get(1).callOnInstance(() ->
CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect()
.size());
assertThat(subscribedCount)
.as("V4 client should not be able to subscribe")
.isEqualTo(0);
}
}
}
@Test
public void testNonSubscribedClientDoesNotReceiveEvent() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V4)
.build())
{
client.connect(false);
int subscribedCount = cluster.get(1).callOnInstance(() ->
CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect()
.size());
assertThat(subscribedCount).isEqualTo(0);
}
}
}
@Test
public void testServerStopsAcceptingNewConnectionsOnDrain() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042).build())
{
client.connect(false);
cluster.get(1).nodetool("drain");
assertThatThrownBy(() -> {
SimpleClient newClient = SimpleClient.builder(nativeAddr.getHostString(), 9042).build();
newClient.connect(false);
}).as("Server should reject new connections after stopAcceptingNewConnections")
.isNotNull();
cluster.get(1).shutdown();
}
}
}
@Test
public void testNoEventEmittedWhenDisabled() throws IOException
{
try (Cluster cluster = buildCluster(1, false))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042).build())
{
client.connect(false);
client.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
int subscribedCount = cluster.get(1).callOnInstance(() ->
CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect()
.size());
assertThat(subscribedCount).isEqualTo(0);
boolean disabled = cluster.get(1).callOnInstance(() -> !DatabaseDescriptor.getGracefulDisconnectEnabled());
assertThat(disabled).isTrue();
}
}
}
@Test
public void testDrainProceedsImmediatelyWithNoSubscribedConnections() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V4)
.build())
{
client.connect(false);
cluster.get(1).nodetoolResult("drain").asserts().success();
cluster.get(1).shutdown();
}
}
}
@Test
public void testMultipleConnectionsCanSubscribe() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client1 = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build();
SimpleClient client2 = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
client1.connect(false);
client2.connect(false);
client1.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
client2.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
int subscribedCount = cluster.get(1).callOnInstance(() ->
CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect()
.size());
assertThat(subscribedCount).isEqualTo(2);
}
}
}
@Test
public void testGracefulDisconnectEventReceivedOnDrain() throws IOException, InterruptedException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
SimpleClient.SimpleEventHandler handler = new SimpleClient.SimpleEventHandler();
client.setEventHandler(handler);
client.connect(false);
client.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
cluster.get(1).runOnInstance(() -> {
ChannelGroup channelGroup = CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect();
StorageService.instance.gracefulDisconnect(() -> {
}, channelGroup);
});
Event event = handler.queue.poll(10, TimeUnit.SECONDS);
assertThat(event).isNotNull();
assertThat(event.type).isEqualTo(Event.Type.GRACEFUL_DISCONNECT);
}
}
}
@Test
public void testDefaultActionCalledAfterAllConnectionsClose() throws IOException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
client.connect(false);
client.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
client.close();
Awaitility.await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> {
int drainingCount = cluster.get(1).callOnInstance(() ->
ClientMetrics.instance.connectionsDraining.get());
assertThat(drainingCount).isEqualTo(0);
});
boolean actionCalled = cluster.get(1).callOnInstance(() -> {
AtomicBoolean called = new AtomicBoolean(false);
ChannelGroup channelGroup = CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect();
StorageService.instance.gracefulDisconnect(() -> called.set(true), channelGroup);
return called.get();
});
assertThat(actionCalled).isTrue();
}
}
}
@Test
public void testDefaultActionCalledAfterMaxDrainMs() throws IOException, InterruptedException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
client.connect(false);
client.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
cluster.get(1).runOnInstance(() -> {
StorageService.instance.setGracefulDisconnectGracePeriod(3000);
});
cluster.get(1).runOnInstance(() -> {
ChannelGroup channelGroup = CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect();
StorageService.instance.gracefulDisconnect(() -> {
}, channelGroup);
});
Thread.sleep(5000);
boolean finishedDraining = cluster.get(1).callOnInstance(() -> ClientMetrics.instance.connectionsDraining.get() == 0);
assertThat(finishedDraining).isTrue();
}
finally
{
cluster.get(1).runOnInstance(() -> {
StorageService.instance.setGracefulDisconnectGracePeriod(5000);
});
}
}
}
@Test
public void testMultipleConnectionsAllReceiveEvent() throws IOException, InterruptedException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient client1 = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build();
SimpleClient client2 = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build())
{
SimpleClient.SimpleEventHandler handler1 = new SimpleClient.SimpleEventHandler();
SimpleClient.SimpleEventHandler handler2 = new SimpleClient.SimpleEventHandler();
client1.setEventHandler(handler1);
client2.setEventHandler(handler2);
client1.connect(false);
client2.connect(false);
client1.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
client2.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
cluster.get(1).runOnInstance(() -> {
ChannelGroup channelGroup = CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect();
StorageService.instance.gracefulDisconnect(() -> {
}, channelGroup);
});
Event event1 = handler1.queue.poll(10, TimeUnit.SECONDS);
Event event2 = handler2.queue.poll(10, TimeUnit.SECONDS);
assertThat(event1).isNotNull();
assertThat(event2).isNotNull();
assertThat(event1.type).isEqualTo(Event.Type.GRACEFUL_DISCONNECT);
assertThat(event2.type).isEqualTo(Event.Type.GRACEFUL_DISCONNECT);
}
}
}
@Test
public void testMixedFleetOnlySubscribedReceivesEvent() throws IOException, InterruptedException
{
try (Cluster cluster = buildCluster(1, true))
{
InetSocketAddress nativeAddr = cluster.get(1).config().broadcastAddress();
try (SimpleClient subscribedClient = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V5)
.build();
SimpleClient nonSubscribedClient = SimpleClient.builder(nativeAddr.getHostString(), 9042)
.protocolVersion(ProtocolVersion.V4)
.build())
{
SimpleClient.SimpleEventHandler subscribedHandler = new SimpleClient.SimpleEventHandler();
SimpleClient.SimpleEventHandler nonSubscribedHandler = new SimpleClient.SimpleEventHandler();
subscribedClient.setEventHandler(subscribedHandler);
nonSubscribedClient.setEventHandler(nonSubscribedHandler);
subscribedClient.connect(false);
nonSubscribedClient.connect(false);
subscribedClient.execute(new RegisterMessage(Collections.singletonList(Event.Type.GRACEFUL_DISCONNECT)));
cluster.get(1).runOnInstance(() -> {
ChannelGroup channelGroup = CassandraDaemon.getInstanceForTesting()
.nativeTransportService()
.getChannelsSubscribedToGracefulDisconnect();
StorageService.instance.gracefulDisconnect(() -> {
}, channelGroup);
});
Event subscribedEvent = subscribedHandler.queue.poll(10, TimeUnit.SECONDS);
Event nonSubscribedEvent = nonSubscribedHandler.queue.poll(3, TimeUnit.SECONDS);
assertThat(subscribedEvent).isNotNull();
assertThat(subscribedEvent.type).isEqualTo(Event.Type.GRACEFUL_DISCONNECT);
assertThat(nonSubscribedEvent).isNull();
}
}
}
}

2
test/unit/org/apache/cassandra/concurrent/DebuggableScheduledThreadPoolExecutorTest.java
View File

@ -56,7 +56,7 @@ public class DebuggableScheduledThreadPoolExecutorTest
}
@Test
public void testShutdown() throws ExecutionException, InterruptedException, IOException
public void testShutdown() throws ExecutionException, InterruptedException, IOException, TimeoutException
{
ScheduledExecutorPlus testPool = executorFactory().scheduled("testpool");

23
test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java
View File

@ -817,6 +817,29 @@ public class DatabaseDescriptorTest
DatabaseDescriptor.applyThresholdsValidations(conf);
}
@Test
public void testGracefulDisconnectEnabled()
{
Config config = new Config();
Assert.assertFalse("Default value of graceful_disconnect_enabled must be false", config.graceful_disconnect_enabled);
}
@Test
public void testGracefulDisconnectGracePeriod()
{
long originalValue = DatabaseDescriptor.getGracefulDisconnectGracePeriod();
Assert.assertEquals("Default value of graceful_disconnect_grace_period must be 5000", 5000, originalValue);
try
{
DatabaseDescriptor.setGracefulDisconnectGracePeriod(3000);
Assert.assertEquals("graceful_disconnect_grace_period should be updated to 3000", 3000, DatabaseDescriptor.getGracefulDisconnectGracePeriod());
}
finally
{
DatabaseDescriptor.setGracefulDisconnectGracePeriod(originalValue);
}
}
@Test
public void testRowIndexSizeAbortEnabledWarnDisabled()
{

27
test/unit/org/apache/cassandra/service/StorageServiceTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.service;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;
import org.assertj.core.api.Assertions;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
@ -201,6 +202,32 @@ public class StorageServiceTest extends TestBaseImpl
}
}
@Test
public void testGracefulDisconnectGracePeriod()
{
StorageService storageService = StorageService.instance;
long originalGracePeriod = storageService.getGracefulDisconnectGracePeriod();
try
{
storageService.setGracefulDisconnectGracePeriod(3000);
Assertions.assertThat(3000).isEqualTo(storageService.getGracefulDisconnectGracePeriod());
Assertions.assertThatThrownBy(() -> storageService.setGracefulDisconnectGracePeriod(-1)).isInstanceOf(IllegalArgumentException.class);
Assertions.assertThat(3000).isEqualTo(storageService.getGracefulDisconnectGracePeriod());
}
finally
{
storageService.setGracefulDisconnectGracePeriod(originalGracePeriod);
}
}
@Test
public void testGracefulDisconnectEnabled()
{
Assertions.assertThat(StorageService.instance.getGracefulDisconnectEnabled()).isFalse();
}
@Test
public void testColumnIndexCacheSizeInKiB()
{