diff --git a/CHANGES.txt b/CHANGES.txt index cbdd91f177..04dcefd853 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -370,6 +370,7 @@ * Fixed nodetool cfstats printing index name twice (CASSANDRA-14903) * Add flag to disable SASI indexes, and warnings on creation (CASSANDRA-14866) Merged from 3.0: + * Prevent client requests from blocking on executor task queue (CASSANDRA-15013) * Toughen up column drop/recreate type validations (CASSANDRA-15204) * LegacyLayout should handle paging states that cross a collection column (CASSANDRA-15201) * Prevent RuntimeException when username or password is empty/null (CASSANDRA-15198) diff --git a/doc/native_protocol_v4.spec b/doc/native_protocol_v4.spec index 5e1e01d2dd..5670241655 100644 --- a/doc/native_protocol_v4.spec +++ b/doc/native_protocol_v4.spec @@ -275,6 +275,9 @@ Table of Contents mode. This mode will make all Thrift and Compact Tables to be exposed as if they were CQL Tables. This is optional; if not specified, the option will not be used. + - "THROW_ON_OVERLOAD": In case of server overloaded with too many requests, by default the server puts + back pressure on the client connection. Instead, the server can send an OverloadedException error message back to + the client if this option is set to true. 4.1.2. AUTH_RESPONSE @@ -1185,3 +1188,4 @@ Table of Contents * The returned in the v4 protocol is not compatible with the v3 protocol. In other words, a returned by a node using protocol v4 should not be used to query a node using protocol v3 (and vice-versa). + * Added THROW_ON_OVERLOAD startup option (Section 4.1.1). diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 6b487fea01..34a5ce8745 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -185,6 +185,8 @@ public class Config public boolean native_transport_flush_in_batches_legacy = false; public volatile boolean native_transport_allow_older_protocols = true; public int native_transport_frame_block_size_in_kb = 32; + public volatile long native_transport_max_concurrent_requests_in_bytes_per_ip = -1L; + public volatile long native_transport_max_concurrent_requests_in_bytes = -1L; /** diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index bb92716116..0166c5fd09 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -508,6 +508,16 @@ public class DatabaseDescriptor conf.hints_directory = storagedirFor("hints"); } + if (conf.native_transport_max_concurrent_requests_in_bytes <= 0) + { + conf.native_transport_max_concurrent_requests_in_bytes = Runtime.getRuntime().maxMemory() / 10; + } + + if (conf.native_transport_max_concurrent_requests_in_bytes_per_ip <= 0) + { + conf.native_transport_max_concurrent_requests_in_bytes_per_ip = Runtime.getRuntime().maxMemory() / 40; + } + if (conf.cdc_raw_directory == null) { conf.cdc_raw_directory = storagedirFor("cdc_raw"); @@ -2035,6 +2045,26 @@ public class DatabaseDescriptor conf.commitlog_sync_group_window_in_ms = windowMillis; } + public static long getNativeTransportMaxConcurrentRequestsInBytesPerIp() + { + return conf.native_transport_max_concurrent_requests_in_bytes_per_ip; + } + + public static void setNativeTransportMaxConcurrentRequestsInBytesPerIp(long maxConcurrentRequestsInBytes) + { + conf.native_transport_max_concurrent_requests_in_bytes_per_ip = maxConcurrentRequestsInBytes; + } + + public static long getNativeTransportMaxConcurrentRequestsInBytes() + { + return conf.native_transport_max_concurrent_requests_in_bytes; + } + + public static void setNativeTransportMaxConcurrentRequestsInBytes(long maxConcurrentRequestsInBytes) + { + conf.native_transport_max_concurrent_requests_in_bytes = maxConcurrentRequestsInBytes; + } + public static int getCommitLogSyncPeriod() { return conf.commitlog_sync_period_in_ms; diff --git a/src/java/org/apache/cassandra/metrics/ClientMetrics.java b/src/java/org/apache/cassandra/metrics/ClientMetrics.java index a80033ab0e..7599096ae9 100644 --- a/src/java/org/apache/cassandra/metrics/ClientMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ClientMetrics.java @@ -19,6 +19,7 @@ package org.apache.cassandra.metrics; import java.util.*; +import java.util.concurrent.atomic.AtomicInteger; import com.codahale.metrics.Gauge; import com.codahale.metrics.Meter; @@ -40,6 +41,10 @@ public final class ClientMetrics private Meter authSuccess; private Meter authFailure; + private AtomicInteger pausedConnections; + private Gauge pausedConnectionsGauge; + private Meter requestDiscarded; + private ClientMetrics() { } @@ -54,6 +59,11 @@ public final class ClientMetrics authFailure.mark(); } + public void pauseConnection() { pausedConnections.incrementAndGet(); } + public void unpauseConnection() { pausedConnections.decrementAndGet(); } + + public void markRequestDiscarded() { requestDiscarded.mark(); } + public List allConnectedClients() { List clients = new ArrayList<>(); @@ -79,6 +89,10 @@ public final class ClientMetrics authSuccess = registerMeter("AuthSuccess"); authFailure = registerMeter("AuthFailure"); + pausedConnections = new AtomicInteger(); + pausedConnectionsGauge = registerGauge("PausedConnections", pausedConnections::get); + requestDiscarded = registerMeter("RequestDiscarded"); + initialized = true; } diff --git a/src/java/org/apache/cassandra/net/ResourceLimits.java b/src/java/org/apache/cassandra/net/ResourceLimits.java index 83e9b432d2..f8d24d778b 100644 --- a/src/java/org/apache/cassandra/net/ResourceLimits.java +++ b/src/java/org/apache/cassandra/net/ResourceLimits.java @@ -19,7 +19,7 @@ package org.apache.cassandra.net; import java.util.concurrent.atomic.AtomicLongFieldUpdater; -abstract class ResourceLimits +public abstract class ResourceLimits { /** * Represents permits to utilise a resource and ways to allocate and release them. @@ -28,7 +28,7 @@ abstract class ResourceLimits * 1. {@link Concurrent}, for shared limits, which is thread-safe; * 2. {@link Basic}, for limits that are not shared between threads, is not thread-safe. */ - interface Limit + public interface Limit { /** * @return total amount of permits represented by this {@link Limit} - the capacity @@ -54,9 +54,18 @@ abstract class ResourceLimits boolean tryAllocate(long amount); /** - * @param amount return the amount of permits back to this limit + * Allocates an amount independent of permits available from this limit. MUST eventually + * be released back with {@link #release(long)}. + * */ - void release(long amount); + void allocate(long amount); + + /** + * @param amount return the amount of permits back to this limit + * @return {@code ABOVE_LIMIT} if there aren't enough permits available even after the release, or + * {@code BELOW_LIMIT} if there are enough permits available after the releaese. + */ + Outcome release(long amount); } /** @@ -70,7 +79,7 @@ abstract class ResourceLimits private static final AtomicLongFieldUpdater usingUpdater = AtomicLongFieldUpdater.newUpdater(Concurrent.class, "using"); - Concurrent(long limit) + public Concurrent(long limit) { this.limit = limit; } @@ -106,11 +115,22 @@ abstract class ResourceLimits return true; } - public void release(long amount) + public void allocate(long amount) + { + long current, next; + do + { + current = using; + next = current + amount; + } while (!usingUpdater.compareAndSet(this, current, next)); + } + + public Outcome release(long amount) { assert amount >= 0; long using = usingUpdater.addAndGet(this, -amount); assert using >= 0; + return using >= limit ? Outcome.ABOVE_LIMIT : Outcome.BELOW_LIMIT; } } @@ -151,10 +171,16 @@ abstract class ResourceLimits return true; } - public void release(long amount) + public void allocate(long amount) + { + using += amount; + } + + public Outcome release(long amount) { assert amount >= 0 && amount <= using; using -= amount; + return using >= limit ? Outcome.ABOVE_LIMIT : Outcome.BELOW_LIMIT; } } @@ -162,23 +188,33 @@ abstract class ResourceLimits * A convenience class that groups a per-endpoint limit with the global one * to allow allocating/releasing permits from/to both limits as one logical operation. */ - static class EndpointAndGlobal + public static class EndpointAndGlobal { final Limit endpoint; final Limit global; - EndpointAndGlobal(Limit endpoint, Limit global) + public EndpointAndGlobal(Limit endpoint, Limit global) { this.endpoint = endpoint; this.global = global; } + public Limit endpoint() + { + return endpoint; + } + + public Limit global() + { + return global; + } + /** * @return {@code INSUFFICIENT_GLOBAL} if there weren't enough permits in the global limit, or * {@code INSUFFICIENT_ENDPOINT} if there weren't enough permits in the per-endpoint limit, or * {@code SUCCESS} if there were enough permits to take from both. */ - Outcome tryAllocate(long amount) + public Outcome tryAllocate(long amount) { if (!global.tryAllocate(amount)) return Outcome.INSUFFICIENT_GLOBAL; @@ -190,12 +226,20 @@ abstract class ResourceLimits return Outcome.INSUFFICIENT_ENDPOINT; } - void release(long amount) + public void allocate(long amount) { - endpoint.release(amount); - global.release(amount); + global.allocate(amount); + endpoint.allocate(amount); + } + + public Outcome release(long amount) + { + Outcome endpointReleaseOutcome = endpoint.release(amount); + Outcome globalReleaseOutcome = global.release(amount); + return (endpointReleaseOutcome == Outcome.ABOVE_LIMIT || globalReleaseOutcome == Outcome.ABOVE_LIMIT) + ? Outcome.ABOVE_LIMIT : Outcome.BELOW_LIMIT; } } - enum Outcome { SUCCESS, INSUFFICIENT_ENDPOINT, INSUFFICIENT_GLOBAL } + public enum Outcome { SUCCESS, INSUFFICIENT_ENDPOINT, INSUFFICIENT_GLOBAL, BELOW_LIMIT, ABOVE_LIMIT } } diff --git a/src/java/org/apache/cassandra/service/NativeTransportService.java b/src/java/org/apache/cassandra/service/NativeTransportService.java index 79caafcc7f..66b50007be 100644 --- a/src/java/org/apache/cassandra/service/NativeTransportService.java +++ b/src/java/org/apache/cassandra/service/NativeTransportService.java @@ -32,10 +32,9 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.epoll.Epoll; import io.netty.channel.epoll.EpollEventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; -import io.netty.util.concurrent.EventExecutor; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.metrics.ClientMetrics; -import org.apache.cassandra.transport.RequestThreadPoolExecutor; +import org.apache.cassandra.transport.Message; import org.apache.cassandra.transport.Server; import org.apache.cassandra.utils.NativeLibrary; @@ -51,7 +50,6 @@ public class NativeTransportService private boolean initialized = false; private EventLoopGroup workerGroup; - private EventExecutor eventExecutorGroup; /** * Creates netty thread pools and event loops. @@ -62,9 +60,6 @@ public class NativeTransportService if (initialized) return; - // prepare netty resources - eventExecutorGroup = new RequestThreadPoolExecutor(); - if (useEpoll()) { workerGroup = new EpollEventLoopGroup(); @@ -81,7 +76,6 @@ public class NativeTransportService InetAddress nativeAddr = DatabaseDescriptor.getRpcAddress(); org.apache.cassandra.transport.Server.Builder builder = new org.apache.cassandra.transport.Server.Builder() - .withEventExecutor(eventExecutorGroup) .withEventLoopGroup(workerGroup) .withHost(nativeAddr); @@ -141,8 +135,7 @@ public class NativeTransportService // shutdown executors used by netty for native transport server workerGroup.shutdownGracefully(3, 5, TimeUnit.SECONDS).awaitUninterruptibly(); - // shutdownGracefully not implemented yet in RequestThreadPoolExecutor - eventExecutorGroup.shutdown(); + Message.Dispatcher.shutdown(); } /** @@ -174,12 +167,6 @@ public class NativeTransportService return workerGroup; } - @VisibleForTesting - EventExecutor getEventExecutor() - { - return eventExecutorGroup; - } - @VisibleForTesting Collection getServers() { diff --git a/src/java/org/apache/cassandra/transport/Connection.java b/src/java/org/apache/cassandra/transport/Connection.java index 908e7e9648..b7f5b170ff 100644 --- a/src/java/org/apache/cassandra/transport/Connection.java +++ b/src/java/org/apache/cassandra/transport/Connection.java @@ -30,6 +30,7 @@ public class Connection private final Tracker tracker; private volatile FrameBodyTransformer transformer; + private boolean throwOnOverload; public Connection(Channel channel, ProtocolVersion version, Tracker tracker) { @@ -50,6 +51,16 @@ public class Connection return transformer; } + public void setThrowOnOverload(boolean throwOnOverload) + { + this.throwOnOverload = throwOnOverload; + } + + public boolean isThrowOnOverload() + { + return throwOnOverload; + } + public Tracker getTracker() { return tracker; diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java index d3c810b2b9..8163d7ab1f 100644 --- a/src/java/org/apache/cassandra/transport/Frame.java +++ b/src/java/org/apache/cassandra/transport/Frame.java @@ -72,7 +72,7 @@ public class Frame public static Frame create(Message.Type type, int streamId, ProtocolVersion version, EnumSet flags, ByteBuf body) { - Header header = new Header(version, flags, streamId, type); + Header header = new Header(version, flags, streamId, type, body.readableBytes()); return new Frame(header, body); } @@ -87,13 +87,15 @@ public class Frame public final EnumSet flags; public final int streamId; public final Message.Type type; + public final long bodySizeInBytes; - private Header(ProtocolVersion version, EnumSet flags, int streamId, Message.Type type) + private Header(ProtocolVersion version, EnumSet flags, int streamId, Message.Type type, long bodySizeInBytes) { this.version = version; this.flags = flags; this.streamId = streamId; this.type = type; + this.bodySizeInBytes = bodySizeInBytes; } public enum Flag @@ -227,7 +229,7 @@ public class Frame idx += bodyLength; buffer.readerIndex(idx); - return new Frame(new Header(version, decodedFlags, streamId, type), body); + return new Frame(new Header(version, decodedFlags, streamId, type, bodyLength), body); } @Override diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 0571478c67..99c0127701 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -42,6 +42,11 @@ import com.google.common.collect.ImmutableSet; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.apache.cassandra.concurrent.LocalAwareExecutorService; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.exceptions.OverloadedException; +import org.apache.cassandra.metrics.ClientMetrics; +import org.apache.cassandra.net.ResourceLimits; import org.apache.cassandra.service.ClientWarn; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tracing.Tracing; @@ -50,6 +55,8 @@ import org.apache.cassandra.service.QueryState; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.UUIDGen; +import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED; + /** * A message from the CQL binary protocol. */ @@ -452,19 +459,42 @@ public abstract class Message } } - @ChannelHandler.Sharable public static class Dispatcher extends SimpleChannelInboundHandler { + private static final LocalAwareExecutorService requestExecutor = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(), + Integer.MAX_VALUE, + "transport", + "Native-Transport-Requests"); + + /** + * Current count of *request* bytes that are live on the channel. + * + * Note: should only be accessed while on the netty event loop. + */ + private long channelPayloadBytesInFlight; + + private final Server.EndpointPayloadTracker endpointPayloadTracker; + + private boolean paused; + private static class FlushItem { final ChannelHandlerContext ctx; final Object response; final Frame sourceFrame; - private FlushItem(ChannelHandlerContext ctx, Object response, Frame sourceFrame) + final Dispatcher dispatcher; + + private FlushItem(ChannelHandlerContext ctx, Object response, Frame sourceFrame, Dispatcher dispatcher) { this.ctx = ctx; this.sourceFrame = sourceFrame; this.response = response; + this.dispatcher = dispatcher; + } + + public void release() + { + dispatcher.releaseItem(this); } } @@ -520,7 +550,7 @@ public abstract class Message for (ChannelHandlerContext channel : channels) channel.flush(); for (FlushItem item : flushed) - item.sourceFrame.release(); + item.release(); channels.clear(); flushed.clear(); @@ -572,7 +602,7 @@ public abstract class Message for (ChannelHandlerContext channel : channels) channel.flush(); for (FlushItem item : flushed) - item.sourceFrame.release(); + item.release(); channels.clear(); flushed.clear(); @@ -584,16 +614,98 @@ public abstract class Message private final boolean useLegacyFlusher; - public Dispatcher(boolean useLegacyFlusher) + public Dispatcher(boolean useLegacyFlusher, Server.EndpointPayloadTracker endpointPayloadTracker) { super(false); this.useLegacyFlusher = useLegacyFlusher; + this.endpointPayloadTracker = endpointPayloadTracker; } @Override public void channelRead0(ChannelHandlerContext ctx, Request request) { + // if we decide to handle this message, process it outside of the netty event loop + if (shouldHandleRequest(ctx, request)) + requestExecutor.submit(() -> processRequest(ctx, request)); + } + /** This check for inflight payload to potentially discard the request should have been ideally in one of the + * first handlers in the pipeline (Frame::decode()). However, incase of any exception thrown between that + * handler (where inflight payload is incremented) and this handler (Dispatcher::channelRead0) (where inflight + * payload in decremented), inflight payload becomes erroneous. ExceptionHandler is not sufficient for this + * purpose since it does not have the frame associated with the exception. + * + * Note: this method should execute on the netty event loop. + */ + private boolean shouldHandleRequest(ChannelHandlerContext ctx, Request request) + { + long frameSize = request.getSourceFrame().header.bodySizeInBytes; + + ResourceLimits.EndpointAndGlobal endpointAndGlobalPayloadsInFlight = endpointPayloadTracker.endpointAndGlobalPayloadsInFlight; + + // check for overloaded state by trying to allocate framesize to inflight payload trackers + if (endpointAndGlobalPayloadsInFlight.tryAllocate(frameSize) != ResourceLimits.Outcome.SUCCESS) + { + if (request.connection.isThrowOnOverload()) + { + // discard the request and throw an exception + ClientMetrics.instance.markRequestDiscarded(); + logger.trace("Discarded request of size: {}. InflightChannelRequestPayload: {}, InflightEndpointRequestPayload: {}, InflightOverallRequestPayload: {}, Request: {}", + frameSize, + channelPayloadBytesInFlight, + endpointAndGlobalPayloadsInFlight.endpoint().using(), + endpointAndGlobalPayloadsInFlight.global().using(), + request); + throw ErrorMessage.wrap(new OverloadedException("Server is in overloaded state. Cannot accept more requests at this point"), + request.getSourceFrame().header.streamId); + } + else + { + // set backpressure on the channel, and handle the request + endpointAndGlobalPayloadsInFlight.allocate(frameSize); + ctx.channel().config().setAutoRead(false); + ClientMetrics.instance.pauseConnection(); + paused = true; + } + } + + channelPayloadBytesInFlight += frameSize; + return true; + } + + /** + * Note: this method will be used in the {@link Flusher#run()}, which executes on the netty event loop + * ({@link Dispatcher#flusherLookup}). Thus, we assume the semantics and visibility of variables + * of being on the event loop. + */ + private void releaseItem(FlushItem item) + { + long itemSize = item.sourceFrame.header.bodySizeInBytes; + item.sourceFrame.release(); + + // since the request has been processed, decrement inflight payload at channel, endpoint and global levels + channelPayloadBytesInFlight -= itemSize; + ResourceLimits.Outcome endpointGlobalReleaseOutcome = endpointPayloadTracker.endpointAndGlobalPayloadsInFlight.release(itemSize); + + // now check to see if we need to reenable the channel's autoRead. + // If the current payload side is zero, we must reenable autoread as + // 1) we allow no other thread/channel to do it, and + // 2) there's no other events following this one (becuase we're at zero bytes in flight), + // so no successive to trigger the other clause in this if-block + ChannelConfig config = item.ctx.channel().config(); + if (paused && (channelPayloadBytesInFlight == 0 || endpointGlobalReleaseOutcome == ResourceLimits.Outcome.BELOW_LIMIT)) + { + paused = false; + ClientMetrics.instance.unpauseConnection(); + config.setAutoRead(true); + } + } + + /** + * Note: this method is not expected to execute on the netty event loop. + */ + void processRequest(ChannelHandlerContext ctx, Request request) + { final Response response; final ServerConnection connection; long queryStartNanoTime = System.nanoTime(); @@ -619,7 +731,7 @@ public abstract class Message { JVMStabilityInspector.inspectThrowable(t); UnexpectedChannelExceptionHandler handler = new UnexpectedChannelExceptionHandler(ctx.channel(), true); - flush(new FlushItem(ctx, ErrorMessage.fromException(t, handler).setStreamId(request.getStreamId()), request.getSourceFrame())); + flush(new FlushItem(ctx, ErrorMessage.fromException(t, handler).setStreamId(request.getStreamId()), request.getSourceFrame(), this)); return; } finally @@ -628,7 +740,19 @@ public abstract class Message } logger.trace("Responding: {}, v={}", response, connection.getVersion()); - flush(new FlushItem(ctx, response, request.getSourceFrame())); + flush(new FlushItem(ctx, response, request.getSourceFrame(), this)); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) + { + endpointPayloadTracker.release(); + if (paused) + { + paused = false; + ClientMetrics.instance.unpauseConnection(); + } + ctx.fireChannelInactive(); } private void flush(FlushItem item) @@ -646,6 +770,14 @@ public abstract class Message flusher.queued.add(item); flusher.start(); } + + public static void shutdown() + { + if (requestExecutor != null) + { + requestExecutor.shutdown(); + } + } } @ChannelHandler.Sharable diff --git a/src/java/org/apache/cassandra/transport/RequestThreadPoolExecutor.java b/src/java/org/apache/cassandra/transport/RequestThreadPoolExecutor.java deleted file mode 100644 index 75dd05ddf8..0000000000 --- a/src/java/org/apache/cassandra/transport/RequestThreadPoolExecutor.java +++ /dev/null @@ -1,96 +0,0 @@ -/* - * 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.transport; - -import java.util.List; -import java.util.concurrent.TimeUnit; - -import io.netty.util.concurrent.AbstractEventExecutor; -import io.netty.util.concurrent.EventExecutorGroup; -import io.netty.util.concurrent.Future; -import org.apache.cassandra.concurrent.LocalAwareExecutorService; -import org.apache.cassandra.config.DatabaseDescriptor; - -import static org.apache.cassandra.concurrent.SharedExecutorPool.SHARED; - -public class RequestThreadPoolExecutor extends AbstractEventExecutor -{ - private final static int MAX_QUEUED_REQUESTS = Integer.getInteger("cassandra.max_queued_native_transport_requests", 128); - private final static String THREAD_FACTORY_ID = "Native-Transport-Requests"; - private final LocalAwareExecutorService wrapped = SHARED.newExecutor(DatabaseDescriptor.getNativeTransportMaxThreads(), - MAX_QUEUED_REQUESTS, - "transport", - THREAD_FACTORY_ID); - - public boolean isShuttingDown() - { - return wrapped.isShutdown(); - } - - public Future shutdownGracefully(long l, long l2, TimeUnit timeUnit) - { - throw new IllegalStateException(); - } - - public Future terminationFuture() - { - throw new IllegalStateException(); - } - - @Override - public void shutdown() - { - wrapped.shutdown(); - } - - @Override - public List shutdownNow() - { - return wrapped.shutdownNow(); - } - - public boolean isShutdown() - { - return wrapped.isShutdown(); - } - - public boolean isTerminated() - { - return wrapped.isTerminated(); - } - - public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException - { - return wrapped.awaitTermination(timeout, unit); - } - - public EventExecutorGroup parent() - { - return null; - } - - public boolean inEventLoop(Thread thread) - { - return false; - } - - public void execute(Runnable command) - { - wrapped.execute(command); - } -} diff --git a/src/java/org/apache/cassandra/transport/Server.java b/src/java/org/apache/cassandra/transport/Server.java index f16aa881a0..c4690f1578 100644 --- a/src/java/org/apache/cassandra/transport/Server.java +++ b/src/java/org/apache/cassandra/transport/Server.java @@ -24,7 +24,9 @@ import java.net.UnknownHostException; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -55,6 +57,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.EncryptionOptions; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.ResourceLimits; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaChangeListener; import org.apache.cassandra.security.SSLFactory; @@ -87,7 +90,6 @@ public class Server implements CassandraDaemon.Server private final AtomicBoolean isRunning = new AtomicBoolean(false); private EventLoopGroup workerGroup; - private EventExecutor eventExecutorGroup; private Server (Builder builder) { @@ -104,8 +106,6 @@ public class Server implements CassandraDaemon.Server else workerGroup = new NioEventLoopGroup(); } - if (builder.eventExecutorGroup != null) - eventExecutorGroup = builder.eventExecutorGroup; EventNotifier notifier = new EventNotifier(this); StorageService.instance.register(notifier); Schema.instance.registerListener(notifier); @@ -234,12 +234,6 @@ public class Server implements CassandraDaemon.Server return this; } - public Builder withEventExecutor(EventExecutor eventExecutor) - { - this.eventExecutorGroup = eventExecutor; - return this; - } - public Builder withHost(InetAddress host) { this.hostAddr = host; @@ -341,6 +335,49 @@ public class Server implements CassandraDaemon.Server } + // global inflight payload across all channels across all endpoints + private static final ResourceLimits.Concurrent globalRequestPayloadInFlight = new ResourceLimits.Concurrent(DatabaseDescriptor.getNativeTransportMaxConcurrentRequestsInBytes()); + + public static class EndpointPayloadTracker + { + // inflight payload per endpoint across corresponding channels + private static final ConcurrentMap requestPayloadInFlightPerEndpoint = new ConcurrentHashMap<>(); + + private final AtomicInteger refCount = new AtomicInteger(0); + private final InetAddress endpoint; + + final ResourceLimits.EndpointAndGlobal endpointAndGlobalPayloadsInFlight = new ResourceLimits.EndpointAndGlobal(new ResourceLimits.Concurrent(DatabaseDescriptor.getNativeTransportMaxConcurrentRequestsInBytesPerIp()), + globalRequestPayloadInFlight); + + private EndpointPayloadTracker(InetAddress endpoint) + { + this.endpoint = endpoint; + } + + public static EndpointPayloadTracker get(InetAddress endpoint) + { + while (true) + { + EndpointPayloadTracker result = requestPayloadInFlightPerEndpoint.computeIfAbsent(endpoint, EndpointPayloadTracker::new); + if (result.acquire()) + return result; + + requestPayloadInFlightPerEndpoint.remove(endpoint, result); + } + } + + private boolean acquire() + { + return 0 < refCount.updateAndGet(i -> i < 0 ? i : i + 1); + } + + public void release() + { + if (-1 == refCount.updateAndGet(i -> i == 1 ? -1 : i - 1)) + requestPayloadInFlightPerEndpoint.remove(endpoint, this); + } + } + private static class Initializer extends ChannelInitializer { // Stateless handlers @@ -350,7 +387,6 @@ public class Server implements CassandraDaemon.Server private static final Frame.OutboundBodyTransformer outboundFrameTransformer = new Frame.OutboundBodyTransformer(); private static final Frame.Encoder frameEncoder = new Frame.Encoder(); private static final Message.ExceptionHandler exceptionHandler = new Message.ExceptionHandler(); - private static final Message.Dispatcher dispatcher = new Message.Dispatcher(DatabaseDescriptor.useNativeTransportLegacyFlusher()); private static final ConnectionLimitHandler connectionLimitHandler = new ConnectionLimitHandler(); private final Server server; @@ -397,6 +433,9 @@ public class Server implements CassandraDaemon.Server pipeline.addLast("messageDecoder", messageDecoder); pipeline.addLast("messageEncoder", messageEncoder); + pipeline.addLast("executor", new Message.Dispatcher(DatabaseDescriptor.useNativeTransportLegacyFlusher(), + EndpointPayloadTracker.get(((InetSocketAddress) channel.remoteAddress()).getAddress()))); + // The exceptionHandler will take care of handling exceptionCaught(...) events while still running // on the same EventLoop as all previous added handlers in the pipeline. This is important as the used // eventExecutorGroup may not enforce strict ordering for channel events. @@ -404,11 +443,6 @@ public class Server implements CassandraDaemon.Server // correctly handled before the handler itself is removed. // See https://issues.apache.org/jira/browse/CASSANDRA-13649 pipeline.addLast("exceptionHandler", exceptionHandler); - - if (server.eventExecutorGroup != null) - pipeline.addLast(server.eventExecutorGroup, "executor", dispatcher); - else - pipeline.addLast("executor", dispatcher); } } diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index deba207d6a..6340b69cb5 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -123,11 +123,19 @@ public class SimpleClient implements Closeable } public SimpleClient connect(boolean useCompression, boolean useChecksums) throws IOException + { + return connect(useCompression, useChecksums, false); + } + + public SimpleClient connect(boolean useCompression, boolean useChecksums, boolean throwOnOverload) throws IOException { establishConnection(); Map options = new HashMap<>(); options.put(StartupMessage.CQL_VERSION, "3.0.0"); + if (throwOnOverload) + options.put(StartupMessage.THROW_ON_OVERLOAD, "1"); + connection.setThrowOnOverload(throwOnOverload); if (useChecksums) { diff --git a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java index ef846c1f9a..ee2b34e8ae 100644 --- a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java @@ -46,6 +46,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 CHECKSUM = "CONTENT_CHECKSUM"; + public static final String THROW_ON_OVERLOAD = "THROW_ON_OVERLOAD"; public static final Message.Codec codec = new Message.Codec() { @@ -104,6 +105,8 @@ public class StartupMessage extends Message.Request connection.setTransformer(CompressingTransformer.getTransformer(compressor)); } + connection.setThrowOnOverload("1".equals(options.get(THROW_ON_OVERLOAD))); + ClientState clientState = state.getClientState(); String driverName = options.get(DRIVER_NAME); if (null != driverName) diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java index 1eb56bcbb6..9c4f22e51e 100644 --- a/test/unit/org/apache/cassandra/cql3/CQLTester.java +++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java @@ -52,6 +52,7 @@ import org.apache.cassandra.config.EncryptionOptions; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.metrics.ClientMetrics; import org.apache.cassandra.schema.*; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.functions.FunctionName; @@ -404,6 +405,7 @@ public abstract class CQLTester SchemaLoader.startGossiper(); server = new Server.Builder().withHost(nativeAddr).withPort(nativePort).build(); + ClientMetrics.instance.init(Collections.singleton(server)); server.start(); for (ProtocolVersion version : PROTOCOL_VERSIONS) @@ -891,9 +893,14 @@ public abstract class CQLTester return sessions.get(protocolVersion); } + protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression, boolean checksums, boolean isOverloadedException) throws IOException + { + return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions()).connect(compression, checksums, isOverloadedException); + } + protected SimpleClient newSimpleClient(ProtocolVersion version, boolean compression, boolean checksums) throws IOException { - return new SimpleClient(nativeAddr.getHostAddress(), nativePort, version, version.isBeta(), new EncryptionOptions()).connect(compression, checksums); + return newSimpleClient(version, compression, checksums, false); } protected String formatQuery(String query) diff --git a/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java b/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java index 80b3596c80..86b73ab580 100644 --- a/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java +++ b/test/unit/org/apache/cassandra/service/NativeTransportServiceTest.java @@ -85,8 +85,7 @@ public class NativeTransportServiceTest { withService((NativeTransportService service) -> { BooleanSupplier allTerminated = () -> - service.getWorkerGroup().isShutdown() && service.getWorkerGroup().isTerminated() && - service.getEventExecutor().isShutdown() && service.getEventExecutor().isTerminated(); + service.getWorkerGroup().isShutdown() && service.getWorkerGroup().isTerminated(); assertFalse(allTerminated.getAsBoolean()); service.destroy(); assertTrue(allTerminated.getAsBoolean()); diff --git a/test/unit/org/apache/cassandra/transport/InflightRequestPayloadTrackerTest.java b/test/unit/org/apache/cassandra/transport/InflightRequestPayloadTrackerTest.java new file mode 100644 index 0000000000..c9a9a02e81 --- /dev/null +++ b/test/unit/org/apache/cassandra/transport/InflightRequestPayloadTrackerTest.java @@ -0,0 +1,258 @@ +/* + * 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.transport; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.cassandra.OrderedJUnit4ClassRunner; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.EncryptionOptions; +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.exceptions.OverloadedException; +import org.apache.cassandra.transport.messages.QueryMessage; + +@RunWith(OrderedJUnit4ClassRunner.class) +public class InflightRequestPayloadTrackerTest extends CQLTester +{ + @BeforeClass + public static void setUp() + { + DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytesPerIp(600); + DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytes(600); + requireNetwork(); + } + + @AfterClass + public static void tearDown() + { + DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytesPerIp(3000000000L); + DatabaseDescriptor.setNativeTransportMaxConcurrentRequestsInBytes(5000000000L); + } + + @After + public void dropCreatedTable() + { + try + { + QueryProcessor.executeOnceInternal("DROP TABLE " + KEYSPACE + ".atable"); + } + catch (Throwable t) + { + // ignore + } + } + + @Test + public void testQueryExecutionWithThrowOnOverload() throws Throwable + { + SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), + nativePort, + ProtocolVersion.V5, + true, + new EncryptionOptions()); + + try + { + client.connect(false, false, true); + QueryOptions queryOptions = QueryOptions.create( + QueryOptions.DEFAULT.getConsistency(), + QueryOptions.DEFAULT.getValues(), + QueryOptions.DEFAULT.skipMetadata(), + QueryOptions.DEFAULT.getPageSize(), + QueryOptions.DEFAULT.getPagingState(), + QueryOptions.DEFAULT.getSerialConsistency(), + ProtocolVersion.V5, + KEYSPACE); + + QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk1 int PRIMARY KEY, v text)", + queryOptions); + client.execute(queryMessage); + } + finally + { + client.close(); + } + } + + @Test + public void testQueryExecutionWithoutThrowOnOverload() throws Throwable + { + SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), + nativePort, + ProtocolVersion.V5, + true, + new EncryptionOptions()); + + try + { + client.connect(false, false, false); + QueryOptions queryOptions = QueryOptions.create( + QueryOptions.DEFAULT.getConsistency(), + QueryOptions.DEFAULT.getValues(), + QueryOptions.DEFAULT.skipMetadata(), + QueryOptions.DEFAULT.getPageSize(), + QueryOptions.DEFAULT.getPagingState(), + QueryOptions.DEFAULT.getSerialConsistency(), + ProtocolVersion.V5, + KEYSPACE); + + QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", + queryOptions); + client.execute(queryMessage); + queryMessage = new QueryMessage("SELECT * FROM atable", + queryOptions); + client.execute(queryMessage); + } + finally + { + client.close(); + } + } + + @Test + public void testQueryExecutionWithoutThrowOnOverloadAndInflightLimitedExceeded() throws Throwable + { + SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), + nativePort, + ProtocolVersion.V5, + true, + new EncryptionOptions()); + + try + { + client.connect(false, false, false); + QueryOptions queryOptions = QueryOptions.create( + QueryOptions.DEFAULT.getConsistency(), + QueryOptions.DEFAULT.getValues(), + QueryOptions.DEFAULT.skipMetadata(), + QueryOptions.DEFAULT.getPageSize(), + QueryOptions.DEFAULT.getPagingState(), + QueryOptions.DEFAULT.getSerialConsistency(), + ProtocolVersion.V5, + KEYSPACE); + + QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", + queryOptions); + client.execute(queryMessage); + + queryMessage = new QueryMessage("INSERT INTO atable (pk, v) VALUES (1, 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')", + queryOptions); + client.execute(queryMessage); + } + finally + { + client.close(); + } + } + + @Test + public void testOverloadedExceptionForEndpointInflightLimit() throws Throwable + { + SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), + nativePort, + ProtocolVersion.V5, + true, + new EncryptionOptions()); + + try + { + client.connect(false, false, true); + QueryOptions queryOptions = QueryOptions.create( + QueryOptions.DEFAULT.getConsistency(), + QueryOptions.DEFAULT.getValues(), + QueryOptions.DEFAULT.skipMetadata(), + QueryOptions.DEFAULT.getPageSize(), + QueryOptions.DEFAULT.getPagingState(), + QueryOptions.DEFAULT.getSerialConsistency(), + ProtocolVersion.V5, + KEYSPACE); + + QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", + queryOptions); + client.execute(queryMessage); + + queryMessage = new QueryMessage("INSERT INTO atable (pk, v) VALUES (1, 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')", + queryOptions); + try + { + client.execute(queryMessage); + Assert.fail(); + } + catch (RuntimeException e) + { + Assert.assertTrue(e.getCause() instanceof OverloadedException); + } + } + finally + { + client.close(); + } + } + + @Test + public void testOverloadedExceptionForOverallInflightLimit() throws Throwable + { + SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), + nativePort, + ProtocolVersion.V5, + true, + new EncryptionOptions()); + + try + { + client.connect(false, false, true); + QueryOptions queryOptions = QueryOptions.create( + QueryOptions.DEFAULT.getConsistency(), + QueryOptions.DEFAULT.getValues(), + QueryOptions.DEFAULT.skipMetadata(), + QueryOptions.DEFAULT.getPageSize(), + QueryOptions.DEFAULT.getPagingState(), + QueryOptions.DEFAULT.getSerialConsistency(), + ProtocolVersion.V5, + KEYSPACE); + + QueryMessage queryMessage = new QueryMessage("CREATE TABLE atable (pk int PRIMARY KEY, v text)", + queryOptions); + client.execute(queryMessage); + + queryMessage = new QueryMessage("INSERT INTO atable (pk, v) VALUES (1, 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa')", + queryOptions); + try + { + client.execute(queryMessage); + Assert.fail(); + } + catch (RuntimeException e) + { + Assert.assertTrue(e.getCause() instanceof OverloadedException); + } + } + finally + { + client.close(); + } + } +} \ No newline at end of file