mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.11' into trunk
This commit is contained in:
Benedict Elliott Smith
commit
8a04204c10
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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 <paging_state> returned in the v4 protocol is not compatible with the v3
|
||||
protocol. In other words, a <paging_state> 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).
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<Integer> 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<ConnectedClient> allConnectedClients()
|
||||
{
|
||||
List<ConnectedClient> 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;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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. <em>MUST</em> 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<Concurrent> 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 }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Server> getServers()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -72,7 +72,7 @@ public class Frame
|
|||
|
||||
public static Frame create(Message.Type type, int streamId, ProtocolVersion version, EnumSet<Header.Flag> 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<Flag> flags;
|
||||
public final int streamId;
|
||||
public final Message.Type type;
|
||||
public final long bodySizeInBytes;
|
||||
|
||||
private Header(ProtocolVersion version, EnumSet<Flag> flags, int streamId, Message.Type type)
|
||||
private Header(ProtocolVersion version, EnumSet<Flag> 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
|
||||
|
|
|
|||
|
|
@ -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<Request>
|
||||
{
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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<Runnable> 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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<InetAddress, EndpointPayloadTracker> 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<Channel>
|
||||
{
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<String, String> 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)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -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<StartupMessage> codec = new Message.Codec<StartupMessage>()
|
||||
{
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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());
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue