From 4f1f2d954f77ba31761b2b854a52cc40d6e202d6 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Mon, 13 Jul 2026 15:05:17 -0500 Subject: [PATCH 1/3] Ensure a Message's Response streamId is always set patch by Francisco Guerrero; reviewed by Caleb Rackliffe, Benedict Elliot Smith, Tejal Chakeres, Alexander Mitin for CASSANDRA-21508 --- .../transport/CQLMessageHandler.java | 33 +-- .../cassandra/transport/Dispatcher.java | 55 ++-- .../apache/cassandra/transport/Envelope.java | 55 +++- .../transport/ExceptionHandlers.java | 55 +++- .../apache/cassandra/transport/Flusher.java | 32 +-- .../transport/InitialConnectionHandler.java | 22 +- .../apache/cassandra/transport/Message.java | 35 +-- .../transport/PipelineConfigurator.java | 2 +- .../cassandra/transport/PreV5Handlers.java | 59 ++-- .../cassandra/transport/SimpleClient.java | 48 ++-- .../transport/messages/AuthUtil.java | 2 +- .../transport/messages/BatchMessage.java | 2 +- .../transport/messages/ErrorMessage.java | 52 ++-- .../transport/messages/EventMessage.java | 3 +- .../transport/messages/ExecuteMessage.java | 2 +- .../transport/messages/PrepareMessage.java | 12 +- .../transport/messages/QueryMessage.java | 2 +- .../cassandra/transport/DriverBurnTest.java | 4 +- .../transport/SimpleClientPerfTest.java | 4 +- .../test/StreamIdMisrouteTest.java | 254 ++++++++++++++++++ .../test/UnableToParseClientMessageTest.java | 8 +- .../transport/CQLConnectionTest.java | 176 +++++++++++- .../cassandra/transport/ErrorMessageTest.java | 12 +- .../transport/MessageDispatcherTest.java | 7 +- .../transport/ProtocolErrorTest.java | 112 +++++++- .../transport/ProtocolNegotiationTest.java | 24 +- 26 files changed, 867 insertions(+), 205 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java index 2486497c2f37..eec2d4d65f1e 100644 --- a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java +++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java @@ -106,7 +106,7 @@ public class CQLMessageHandler extends AbstractMessageHandler interface MessageConsumer { - void dispatch(Channel channel, M message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure); +

void dispatch(Channel channel, M message, Dispatcher.FlushItemConverter

toFlushItem, P param, Overload backpressure); boolean hasQueueCapacity(); } @@ -389,7 +389,7 @@ protected boolean processRequest(Envelope request, Overload backpressure) try { message = messageDecoder.decode(channel, request); - dispatcher.dispatch(channel, message, this::toFlushItem, backpressure); + dispatcher.dispatch(channel, message, CQLMessageHandler::toFlushItem, this, backpressure); // sucessfully delivered a CQL message to the execution // stage, so reset the counter of consecutive errors @@ -465,13 +465,6 @@ private void handleError(Throwable t, int streamId) errorHandler.accept(ErrorMessage.wrap(t, streamId)); } - /** - * For use in the case where the error can't be mapped to a specific stream id, - * such as a corrupted frame, or when extracting a CQL message from the frame's - * payload fails. This does not attempt to release any resources, as these errors - * should only occur before any capacity acquisition is attempted (e.g. on receipt - * of a corrupt frame, or failure to extract a CQL message from the envelope). - */ private void handleError(Throwable t) { errorHandler.accept(t); @@ -485,7 +478,8 @@ private Framed toFlushItem(Channel channel, Message.Request request, Message.Res // The Dispatcher will call this to obtain the FlushItem to enqueue with its Flusher once // a dispatched request has been processed. - Envelope responseFrame = response.encode(request.getSource().header.version); + Envelope.Header header = request.getSource().header; + Envelope responseFrame = response.encode(header.version, header.streamId); int responseSize = envelopeSize(responseFrame.header); ClientMessageSizeMetrics.bytesSent.inc(responseSize); ClientMessageSizeMetrics.bytesSentPerResponse.update(responseSize); @@ -500,9 +494,9 @@ private Framed toFlushItem(Channel channel, Message.Request request, Message.Res @Override public void cleanup(Flusher.FlushItem flushItem) { - release(flushItem.request.header); - flushItem.request.release(); - flushItem.response.release(); + release(flushItem.requestEnvelope.header); + flushItem.requestEnvelope.release(); + flushItem.responseEnvelope.release(); } private void release(Envelope.Header header) @@ -524,8 +518,9 @@ protected boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpo if (!extracted.isSuccess()) { // Hard fail on any decoding error as we can't trust the subsequent frames of - // the large message - handleError(ProtocolException.toFatalException(extracted.error())); + // the large message. The stream id is a best-effort value read before extraction + // failed, so route it back where possible rather than defaulting. + handleError(ProtocolException.toFatalException(extracted.error()), extracted.streamId()); return false; } @@ -542,7 +537,7 @@ protected boolean processFirstFrameOfLargeMessage(IntactFrame frame, Limit endpo // not make sense to continue processing subsequent frames handleError(ProtocolException.toFatalException(new OversizedAuthMessageException( MULTI_FRAME_AUTH_ERROR_MESSAGE_PREFIX + - "type = " + header.type + ", size = " + header.bodySizeInBytes))); + "type = " + header.type + ", size = " + header.bodySizeInBytes)), header.streamId); ClientMetrics.instance.markRequestDiscarded(); return false; } @@ -731,6 +726,12 @@ protected void processCorruptFrame(FrameDecoder.CorruptFrame frame) processSubsequentFrameOfLargeMessage(frame); } + // A corrupt frame's bytes can't be reliably mapped back to a specific in-flight request (the + // frame may span several streams, and the stream id bytes may themselves be part of the + // corruption). We fall back to this connection's STARTUP stream id purely so the client still + // receives a diagnostic error frame before we tear the connection down (see the class javadoc + // above and CQLConnectionTest#handleFrameCorruptionAfterNegotiation). The exception is fatal, so + // the channel is closed as soon as the error has been written. handleError(ProtocolException.toFatalException(new ProtocolException(error))); } diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 3f6468c21825..5c7879074882 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -93,10 +93,9 @@ public class Dispatcher implements CQLMessageHandler.MessageConsumer { - FlushItem toFlushItem(Channel channel, Message.Request request, Message.Response response); + FlushItem toFlushItem(P param, Channel channel, Message.Request request, Message.Response response); } public Dispatcher(boolean useLegacyFlusher) @@ -105,18 +104,17 @@ public Dispatcher(boolean useLegacyFlusher) } @Override - public void dispatch(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure) + public

void dispatch(Channel channel, Message.Request request, FlushItemConverter

forFlusher, P param, Overload backpressure) { if (!request.connection().getTracker().isRunning()) { // We can not respond with a custom, transport, or server exceptions since, given current implementation of clients, // they will defunct the connection. Without a protocol version bump that introduces an "I am going away message", // we have to stick to an existing error code. - Message.Response response = ErrorMessage.fromException(new OverloadedException("Server is shutting down")); - response.setStreamId(request.getStreamId()); + Message.Response response = ErrorMessage.fromTransportException(new OverloadedException("Server is shutting down")); response.setWarnings(ClientWarn.instance.getWarnings()); response.attach(request.connection); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); + FlushItem toFlush = forFlusher.toFlushItem(param, channel, request, response); flush(toFlush); return; } @@ -128,7 +126,7 @@ public void dispatch(Channel channel, Message.Request request, FlushItemConverte // Importantly, the authExecutor will handle the AUTHENTICATE message which may be CPU intensive. LocalAwareExecutorPlus executor = isAuthQuery ? authExecutor : requestExecutor; - executor.submit(new RequestProcessor(channel, request, forFlusher, backpressure)); + executor.submit(new RequestProcessor<>(channel, request, forFlusher, param, backpressure)); ClientMetrics.instance.markRequestDispatched(); } @@ -293,20 +291,22 @@ public long timeSpentInQueueNanos() * is the only way we can keep it not wrapped into a callable on SEPExecutor submission path. And we need this * functionality for tracking time purposes. */ - public class RequestProcessor implements DebuggableTask.RunnableDebuggableTask + public class RequestProcessor

implements DebuggableTask.RunnableDebuggableTask { private final Channel channel; private final Message.Request request; - private final FlushItemConverter forFlusher; + private final FlushItemConverter

forFlusher; + private final P flusherParam; private final Overload backpressure; private volatile long startTimeNanos; - public RequestProcessor(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure) + public RequestProcessor(Channel channel, Message.Request request, FlushItemConverter

forFlusher, P flusherParam, Overload backpressure) { this.channel = channel; this.request = request; this.forFlusher = forFlusher; + this.flusherParam = flusherParam; this.backpressure = backpressure; } @@ -314,7 +314,7 @@ public RequestProcessor(Channel channel, Message.Request request, FlushItemConve public void run() { startTimeNanos = MonotonicClock.Global.preciseTime.now(); - processRequest(channel, request, forFlusher, backpressure, new RequestTime(request.createdAtNanos, startTimeNanos)); + processRequest(channel, request, forFlusher, flusherParam, backpressure, new RequestTime(request.createdAtNanos, startTimeNanos)); } @Override @@ -374,7 +374,7 @@ private static Message.Response processRequest(ServerConnection connection, Mess if (queueTime > DatabaseDescriptor.getNativeTransportTimeout(TimeUnit.NANOSECONDS)) { ClientMetrics.instance.markTimedOutBeforeProcessing(); - return ErrorMessage.fromException(new OverloadedException("Query timed out before it could start")); + return ErrorMessage.fromTransportException(new OverloadedException("Query timed out before it could start")); } if (connection.getVersion().isGreaterOrEqualTo(ProtocolVersion.V4)) @@ -434,8 +434,6 @@ private static Message.Response processRequest(ServerConnection connection, Mess CoordinatorWriteWarnings.done(); } - response.setStreamId(request.getStreamId()); - response.setWarnings(ClientWarn.instance.getWarnings()); response.attach(connection); connection.applyStateTransition(request.type, response.type); return response; @@ -448,7 +446,10 @@ static Message.Response processRequest(Channel channel, Message.Request request, { try { - return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); + // decorateResponse must run before the finally below resets ClientWarn, so that it can read the + // warnings accumulated during processing. Decorating in the caller (after this method returns) + // would observe the reset state and drop all client warnings. + return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), request); } catch (Throwable t) { @@ -461,10 +462,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - ErrorMessage error = ErrorMessage.fromException(t, handler); - error.setStreamId(request.getStreamId()); - error.setWarnings(ClientWarn.instance.getWarnings()); - return error; + return decorateResponse(ErrorMessage.fromExceptionNoStreamId(t, handler), request); } finally { @@ -477,14 +475,21 @@ static Message.Response processRequest(Channel channel, Message.Request request, /** * Note: this method is not expected to execute on the netty event loop. */ - void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure, RequestTime requestTime) +

void processRequest(Channel channel, Message.Request request, FlushItemConverter

forFlusher, P param, Overload backpressure, RequestTime requestTime) { Message.Response response = processRequest(channel, request, backpressure, requestTime); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); + FlushItem toFlush = forFlusher.toFlushItem(param, channel, request, response); Message.logger.trace("Responding: {}, v={}", response, request.connection().getVersion()); flush(toFlush); } + private static Message.Response decorateResponse(Message.Response response, Message.Request request) + { + assert response != null; + response.setWarnings(ClientWarn.instance.getWarnings()); + return response; + } + private void flush(FlushItem item) { EventLoop loop = item.channel.eventLoop(); @@ -517,7 +522,7 @@ public static void shutdown() * for delivering events to registered clients is dependent on protocol version and the configuration * of the pipeline. For v5 and newer connections, the event message is encoded into an Envelope, * wrapped in a FlushItem and then delivered via the pipeline's flusher, in a similar way to - * a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, Overload, RequestTime)}. + * a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, P, Overload, RequestTime)}. * It's worth noting that events are not generally fired as a direct response to a client request, * so this flush item has a null request attribute. The dispatcher itself is created when the * pipeline is first configured during protocol negotiation and is attached to the channel for @@ -531,9 +536,9 @@ Consumer eventDispatcher(final Channel channel, final FrameEncoder.PayloadAllocator allocator) { return eventMessage -> flush(new FlushItem.Framed(channel, - eventMessage.encode(version), + eventMessage.encode(version, EventMessage.EVENT_MESSAGE_STREAM_ID), // -1 was set in EventMessage previously null, allocator, - f -> f.response.release())); + f -> f.responseEnvelope.release())); } } diff --git a/src/java/org/apache/cassandra/transport/Envelope.java b/src/java/org/apache/cassandra/transport/Envelope.java index e70b4fe5514a..b33b054819ab 100644 --- a/src/java/org/apache/cassandra/transport/Envelope.java +++ b/src/java/org/apache/cassandra/transport/Envelope.java @@ -146,6 +146,11 @@ public static class Header public final Message.Type type; public final long bodySizeInBytes; + public static Header dummy(int streamId, Message.Type type) + { + return new Header(ProtocolVersion.CURRENT, 0, streamId, type, 0); + } + private Header(ProtocolVersion version, int flags, int streamId, Message.Type type, long bodySizeInBytes) { this.version = version; @@ -258,7 +263,7 @@ HeaderExtractionResult extractHeader(ByteBuffer buffer) // This throws a protocol exception if the version number is unsupported, // the opcode is unknown or invalid flags are set for the version version = ProtocolVersion.decode(versionNum, DatabaseDescriptor.getNativeTransportAllowOlderProtocols()); - validateFlags(version, flags); + validateFlags(version, flags, streamId); type = Message.Type.fromOpcode(opcode, direction); return new HeaderExtractionResult.Success(new Header(version, flags, streamId, type, bodyLength)); } @@ -272,6 +277,10 @@ HeaderExtractionResult extractHeader(ByteBuffer buffer) // cause the channel to be closed. return new HeaderExtractionResult.Error(e, streamId, bodyLength); } + catch (ErrorMessage.WrappedException e) + { + return new HeaderExtractionResult.Error((ProtocolException) e.getCause(), e.getStreamId(), bodyLength); + } } public static abstract class HeaderExtractionResult @@ -370,7 +379,8 @@ Envelope decode(ByteBuf buffer) Message.Direction direction = Message.Direction.extractFromVersion(firstByte); int versionNum = firstByte & PROTOCOL_VERSION_MASK; - ProtocolVersion version; + ProtocolVersion version = null; + ProtocolException protocolException = null; try { @@ -378,19 +388,43 @@ Envelope decode(ByteBuf buffer) } catch (ProtocolException e) { - // Skip the remaining useless bytes. Otherwise the channel closing logic may try to decode again. - buffer.skipBytes(readableBytes); - throw e; + // defer throw to attempt to extract the stream id + protocolException = e; } // Wait until we have the complete header if (readableBytes < Header.LENGTH) + { + if (protocolException != null) + { + // Skip the remaining useless bytes. Otherwise the channel closing logic may try to decode again. + buffer.skipBytes(readableBytes); + // The header is incomplete, so there is no stream id to recover. Wrap with the unset + // sentinel; the channel-level exception handler sees it has no routable stream id and + // closes the connection rather than emit an unroutable error frame. + throw protocolException; + } return null; + } int flags = buffer.getByte(idx++); - validateFlags(version, flags); - int streamId = buffer.getShort(idx); + + if (protocolException != null) + { + // Protocol versions 1 and 2 use a shorter header with a single-byte stream id. Reading a + // 16-bit stream id from such a header splices the stream id byte together with the opcode + // byte and recovers a bogus id, routing the error to a stream the client never used + // (CASSANDRA-21508). A v1/v2 client that downgrades would then never see the error and time + // out, so recover the stream id using the attempted version's header layout. + int recoveredStreamId = versionNum < ProtocolVersion.V3.asInt() ? buffer.getByte(idx) : streamId; + // Skip the remaining useless bytes. Otherwise the channel closing logic may try to decode again. + buffer.skipBytes(readableBytes); + throw ErrorMessage.wrap(protocolException, recoveredStreamId); + } + + validateFlags(version, flags, streamId); + idx += 2; // This throws a protocol exceptions if the opcode is unknown @@ -436,11 +470,12 @@ Envelope decode(ByteBuf buffer) return new Envelope(new Header(version, flags, streamId, type, bodyLength), body); } - private void validateFlags(ProtocolVersion version, int flags) + private void validateFlags(ProtocolVersion version, int flags, int streamId) { if (version.isBeta() && !Header.Flag.contains(flags, Header.Flag.USE_BETA)) - throw new ProtocolException(String.format("Beta version of the protocol used (%s), but USE_BETA flag is unset", version), - version); + throw ErrorMessage.wrap(new ProtocolException(String.format("Beta version of the protocol used (%s), but USE_BETA flag is unset", version), + version), + streamId); } @Override diff --git a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java index 12bfe4190de6..2a97704db81e 100644 --- a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java +++ b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java @@ -38,6 +38,7 @@ import org.apache.cassandra.metrics.ClientMetrics; import org.apache.cassandra.net.FrameEncoder; import org.apache.cassandra.transport.messages.ErrorMessage; +import org.apache.cassandra.transport.messages.ErrorMessage.WithStreamId; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.Throwables; @@ -76,23 +77,53 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) if (ctx.channel().isOpen()) { Predicate handler = getUnexpectedExceptionHandler(ctx.channel(), false); - ErrorMessage errorMessage = ErrorMessage.fromException(cause, handler); - Envelope response = errorMessage.encode(version); - FrameEncoder.Payload payload = allocator.allocate(true, CQLMessageHandler.envelopeSize(response.header)); + // No request in scope at the channel level; a WrappedException cause carries the frame's + // stream id and overrides this fallback. + WithStreamId withStreamId = ErrorMessage.fromException(cause, handler); + ErrorMessage errorMessage = withStreamId.message; try { - response.encodeInto(payload.buffer); - response.release(); - payload.finish(); - ChannelPromise promise = ctx.newPromise(); - // On protocol exception, close the channel as soon as the message has been sent - if (isFatal(cause)) - promise.addListener(future -> ctx.close()); - ctx.writeAndFlush(payload, promise); + int streamId = withStreamId.streamId; + boolean isFatal = isFatal(cause); + if (streamId == Message.UNSET_STREAM_ID) + { + // No stream id could be recovered, so we have no request to route a response to. + // Close the connection rather than emit an unroutable frame (CASSANDRA-21508). + isFatal = true; + streamId = 0; + } + + Envelope response = errorMessage.encode(version, streamId); + FrameEncoder.Payload payload = allocator.allocate(true, CQLMessageHandler.envelopeSize(response.header)); + try + { + response.encodeInto(payload.buffer); + response.release(); + payload.finish(); + ChannelPromise promise = ctx.newPromise(); + // On a fatal error, close the channel only once the error frame has been written, + // so the client receives the diagnostic before the connection is torn down. Closing + // synchronously here can abort the in-flight flush and drop the frame when the socket + // can't drain it immediately (TCP backpressure, TLS buffering, or a large frame). + // Matches PreV5Handlers.ExceptionHandler and InitialConnectionHandler. + // + // Trade-off of deferring the close (CASSANDRA-21508): + // - There is a slim chance we send two frames with the same streamId. Responses + // already queued on this connection will have a chance to flush before the close + // fires. For the majority of cases, each frame will carry its own unique stream + // id, so nothing is misrouted. These are valid responses to requests that were + // correctly-decoded earlier. + if (isFatal) + promise.addListener(future -> ctx.close()); + ctx.writeAndFlush(payload, promise); + } + finally + { + payload.release(); + } } finally { - payload.release(); JVMStabilityInspector.inspectThrowable(cause); } } diff --git a/src/java/org/apache/cassandra/transport/Flusher.java b/src/java/org/apache/cassandra/transport/Flusher.java index d0b76e0a9912..674af5250e79 100644 --- a/src/java/org/apache/cassandra/transport/Flusher.java +++ b/src/java/org/apache/cassandra/transport/Flusher.java @@ -36,7 +36,6 @@ import org.apache.cassandra.net.FrameEncoder; import org.apache.cassandra.net.FrameEncoderCrc; import org.apache.cassandra.net.FrameEncoderLZ4; -import org.apache.cassandra.transport.Message.Response; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.memory.BufferPool; @@ -57,22 +56,23 @@ abstract class Flusher implements Runnable interface OnFlushCleanup { void cleanup(FlushItem item); } - static class FlushItem + + public static class FlushItem { enum Kind {FRAMED, UNFRAMED} final Kind kind; final Channel channel; - final T response; - final Envelope request; + final T responseEnvelope; + final Envelope requestEnvelope; final OnFlushCleanup tidy; - FlushItem(Kind kind, Channel channel, T response, Envelope request, OnFlushCleanup tidy) + FlushItem(Kind kind, Channel channel, T responseEnvelope, Envelope requestEnvelope, OnFlushCleanup tidy) { this.kind = kind; this.channel = channel; - this.request = request; - this.response = response; + this.requestEnvelope = requestEnvelope; + this.responseEnvelope = responseEnvelope; this.tidy = tidy; } @@ -85,21 +85,21 @@ static class Framed extends FlushItem { final FrameEncoder.PayloadAllocator allocator; Framed(Channel channel, - Envelope response, - Envelope request, + Envelope responseEnvelope, + Envelope requestEnvelope, FrameEncoder.PayloadAllocator allocator, OnFlushCleanup tidy) { - super(Kind.FRAMED, channel, response, request, tidy); + super(Kind.FRAMED, channel, responseEnvelope, requestEnvelope, tidy); this.allocator = allocator; } } - static class Unframed extends FlushItem + static class Unframed extends FlushItem { - Unframed(Channel channel, Response response, Envelope request, OnFlushCleanup tidy) + Unframed(Channel channel, Envelope responseEnvelope, Envelope requestEnvelope, OnFlushCleanup tidy) { - super(Kind.UNFRAMED, channel, response, request, tidy); + super(Kind.UNFRAMED, channel, responseEnvelope, requestEnvelope, tidy); } } } @@ -151,13 +151,13 @@ boolean isEmpty() private void processUnframedResponse(FlushItem.Unframed flush) { - flush.channel.write(flush.response, flush.channel.voidPromise()); + flush.channel.write(flush.responseEnvelope, flush.channel.voidPromise()); channels.add(flush.channel); } private void processFramedResponse(FlushItem.Framed flush) { - Envelope outbound = flush.response; + Envelope outbound = flush.responseEnvelope; if (envelopeSize(outbound.header) >= MAX_FRAMED_PAYLOAD_SIZE) { flushLargeMessage(flush.channel, outbound, flush.allocator); @@ -171,7 +171,7 @@ private void processFramedResponse(FlushItem.Framed flush) payloads.put(flushBuffer.channel, flushBuffer); } - flushBuffer.add(flush.response); + flushBuffer.add(flush.responseEnvelope); } } diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index 556a5607934d..e8ddc9e519a8 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -36,6 +36,7 @@ import org.apache.cassandra.transport.messages.SupportedMessage; import io.netty.buffer.ByteBuf; +import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.channel.VoidChannelPromise; @@ -91,8 +92,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li supportedOptions.put(StartupMessage.COMPRESSION, compressions); supportedOptions.put(StartupMessage.PROTOCOL_VERSIONS, ProtocolVersion.supportedVersions()); SupportedMessage supported = new SupportedMessage(supportedOptions); - supported.setStreamId(inbound.header.streamId); - outbound = supported.encode(inbound.header.version); + outbound = supported.encode(inbound.header.version, inbound.header.streamId); ctx.writeAndFlush(outbound); break; @@ -122,7 +122,9 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li if (future.isSuccess()) { logger.trace("Response to STARTUP sent, configuring pipeline for {}", inbound.header.version); - configurator.configureModernPipeline(ctx, serverConnection, allocator, inbound.header.version, startup.options); + configurator.configureModernPipeline(ctx, serverConnection, allocator, + inbound.header.version, + startup.options); allocator.release(inbound.header.bodySizeInBytes); } else @@ -131,8 +133,8 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li if (null == cause) cause = new ServerError("Unexpected error establishing connection"); logger.warn("Writing response to STARTUP failed, unable to configure pipeline", cause); - ErrorMessage error = ErrorMessage.fromException(cause); - Envelope response = error.encode(inbound.header.version); + ErrorMessage error = ErrorMessage.fromExceptionNoStreamId(cause); + Envelope response = error.encode(inbound.header.version, inbound.header.streamId); ChannelPromise closeChannel = AsyncChannelPromise.withListener(ctx, f -> ctx.close()); ctx.writeAndFlush(response, closeChannel); if (ctx.channel().isOpen()) @@ -152,18 +154,20 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li final Message.Response response = Dispatcher.processRequest(ctx.channel(), startup, Overload.NONE, Dispatcher.RequestTime.forImmediateExecution()); - outbound = response.encode(inbound.header.version); + outbound = response.encode(inbound.header.version, inbound.header.streamId); ctx.writeAndFlush(outbound, promise); logger.trace("Configured pipeline: {}", ctx.pipeline()); break; default: ErrorMessage error = - ErrorMessage.fromException( + ErrorMessage.fromTransportException( new ProtocolException(String.format("Unexpected message %s, expecting STARTUP or OPTIONS", inbound.header.type))); - outbound = error.encode(inbound.header.version); - ctx.writeAndFlush(outbound); + outbound = error.encode(inbound.header.version, inbound.header.streamId); + // An unexpected message during initial connection setup leaves the connection in a + // corrupted state; send the error, then close the connection. + ctx.writeAndFlush(outbound).addListener(ChannelFutureListener.CLOSE); } } finally diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index d55e011b3a12..117fdd35ef35 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -65,6 +65,12 @@ public abstract class Message { protected static final Logger logger = LoggerFactory.getLogger(Message.class); + /** + * Sentinel default for a {@link Response}'s stream id; + * must be overwritten before {@link #encode}. + **/ + public static final int UNSET_STREAM_ID = Integer.MIN_VALUE; + public interface Codec extends CBCodec {} public enum Direction @@ -160,7 +166,6 @@ public Codec unsafeSetCodec(Codec codec) throws NoSuchFieldException, Ille public final Type type; protected Connection connection; - private int streamId; private Envelope source; private Map customPayload; protected ProtocolVersion forcedProtocolVersion = null; @@ -180,17 +185,6 @@ public Connection connection() return connection; } - public Message setStreamId(int streamId) - { - this.streamId = streamId; - return this; - } - - public int getStreamId() - { - return streamId; - } - public void setSource(Envelope source) { this.source = source; @@ -214,7 +208,7 @@ public void setCustomPayload(Map customPayload) @Override public String toString() { - return String.format("(%s:%s:%s)", type, streamId, connection == null ? "null" : connection.getVersion().asInt()); + return String.format("(%s:%s)", type, connection == null ? "null" : connection.getVersion().asInt()); } public static abstract class Request extends Message @@ -342,8 +336,16 @@ public List getWarnings() } } - public Envelope encode(ProtocolVersion version) + public Envelope encode(ProtocolVersion version, int streamId) { + // A Response's stream id must be stamped before it is serialized to the wire. UNSET_STREAM_ID here + // means a server code path produced a response without routing information; sending it would risk + // delivering it to an unrelated in-flight request (CASSANDRA-21508). Fail fatally so the connection + // is torn down rather than mis-route a response. Checked before the try below so it is not caught and + // re-wrapped (which would carry the unset id forward). + if (streamId == UNSET_STREAM_ID) + throw ProtocolException.toFatalException(new ProtocolException("Attempted to encode a response with an unset stream id: " + this)); + int flags = Flag.none(); @SuppressWarnings("unchecked") Codec codec = (Codec)this.type.codec; @@ -430,11 +432,11 @@ public Envelope encode(ProtocolVersion version) if (responseVersion.isBeta()) flags = Flag.add(flags, Flag.USE_BETA); - return Envelope.create(type, getStreamId(), responseVersion, flags, body); + return Envelope.create(type, streamId, responseVersion, flags, body); } catch (Throwable e) { - throw ErrorMessage.wrap(e, getStreamId()); + throw ErrorMessage.wrap(e, streamId); } } @@ -455,7 +457,6 @@ static Message decodeMessage(Channel channel, Envelope inbound) throw new ProtocolException("Received frame with CUSTOM_PAYLOAD flag for native protocol version < 4"); Message message = inbound.header.type.codec.decode(inbound.body, inbound.header.version); - message.setStreamId(inbound.header.streamId); message.setSource(inbound); message.setCustomPayload(customPayload); diff --git a/src/java/org/apache/cassandra/transport/PipelineConfigurator.java b/src/java/org/apache/cassandra/transport/PipelineConfigurator.java index 3cb88bbb85ed..2824cfb32a95 100644 --- a/src/java/org/apache/cassandra/transport/PipelineConfigurator.java +++ b/src/java/org/apache/cassandra/transport/PipelineConfigurator.java @@ -409,7 +409,7 @@ public void configureLegacyPipeline(ChannelHandlerContext ctx, ClientResourceLim pipeline.addBefore(INITIAL_HANDLER, MESSAGE_DECOMPRESSOR, Envelope.Decompressor.instance); pipeline.addBefore(INITIAL_HANDLER, MESSAGE_COMPRESSOR, Envelope.Compressor.instance); pipeline.addBefore(INITIAL_HANDLER, MESSAGE_DECODER, PreV5Handlers.ProtocolDecoder.instance); - pipeline.addBefore(INITIAL_HANDLER, MESSAGE_ENCODER, PreV5Handlers.ProtocolEncoder.instance); + pipeline.addBefore(INITIAL_HANDLER, MESSAGE_ENCODER, PreV5Handlers.EventMessageEncoder.instance); pipeline.addBefore(INITIAL_HANDLER, LEGACY_MESSAGE_PROCESSOR, new PreV5Handlers.LegacyDispatchHandler(dispatcher, queueBackpressure, limits)); pipeline.remove(INITIAL_HANDLER); onNegotiationComplete(pipeline); diff --git a/src/java/org/apache/cassandra/transport/PreV5Handlers.java b/src/java/org/apache/cassandra/transport/PreV5Handlers.java index f5b9d82f0223..630e52ae620d 100644 --- a/src/java/org/apache/cassandra/transport/PreV5Handlers.java +++ b/src/java/org/apache/cassandra/transport/PreV5Handlers.java @@ -24,6 +24,7 @@ import javax.net.ssl.SSLException; +import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Predicate; import org.apache.commons.lang3.exception.ExceptionUtils; @@ -40,6 +41,7 @@ import org.apache.cassandra.service.QueryState; import org.apache.cassandra.transport.ClientResourceLimits.Overload; import org.apache.cassandra.transport.messages.ErrorMessage; +import org.apache.cassandra.transport.messages.EventMessage; import org.apache.cassandra.utils.JVMStabilityInspector; import io.netty.channel.Channel; @@ -93,23 +95,26 @@ protected void channelRead0(ChannelHandlerContext ctx, Message.Request request) // The only reason we won't process this message is if checkLimits() throws an OverloadedException. // (i.e. Even if backpressure is applied, the current request is allowed to finish.) checkLimits(ctx, request); - dispatcher.dispatch(ctx.channel(), request, this::toFlushItem, backpressure); + dispatcher.dispatch(ctx.channel(), request, this::toFlushItem, ctx, backpressure); } // Acts as a Dispatcher.FlushItemConverter - private Flusher.FlushItem.Unframed toFlushItem(Channel channel, Message.Request request, Message.Response response) + private Flusher.FlushItem.Unframed toFlushItem(ChannelHandlerContext ctx, Channel channel, Message.Request request, Message.Response response) { - return new Flusher.FlushItem.Unframed(channel, response, request.getSource(), this::releaseItem); + ProtocolVersion version = getConnectionVersion(ctx); + Envelope requestEnvelope = request.getSource(); + Envelope responseEnvelope = response.encode(version, requestEnvelope.header.streamId); + return new Flusher.FlushItem.Unframed(channel, responseEnvelope, requestEnvelope, this::releaseItem); } - private void releaseItem(Flusher.FlushItem item) + private void releaseItem(Flusher.FlushItem item) { // Note: in contrast to the equivalent for V5 protocol, CQLMessageHandler::release(FlushItem item), // this does not release the FlushItem's Message.Response. In V4, the buffers for the response's body // and serialised header are emitted directly down the Netty pipeline from Envelope.Encoder, so // releasing them is handled by the pipeline itself. - long itemSize = item.request.header.bodySizeInBytes; - item.request.release(); + long itemSize = item.requestEnvelope.header.bodySizeInBytes; + item.requestEnvelope.release(); // since the request has been processed, decrement inflight payload at channel, endpoint and global levels channelPayloadBytesInFlight -= itemSize; @@ -307,14 +312,14 @@ public void decode(ChannelHandlerContext ctx, Envelope source, List resu * Simple adaptor to plug CQL message encoding into pre-V5 pipelines */ @ChannelHandler.Sharable - public static class ProtocolEncoder extends MessageToMessageEncoder + public static class EventMessageEncoder extends MessageToMessageEncoder { - public static final ProtocolEncoder instance = new ProtocolEncoder(); - private ProtocolEncoder(){} - public void encode(ChannelHandlerContext ctx, Message source, List results) + public static final EventMessageEncoder instance = new EventMessageEncoder(); + private EventMessageEncoder(){} + public void encode(ChannelHandlerContext ctx, EventMessage source, List results) { ProtocolVersion version = getConnectionVersion(ctx); - results.add(source.encode(version)); + results.add(source.encode(version, EventMessage.EVENT_MESSAGE_STREAM_ID)); } } @@ -337,13 +342,28 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) if (ctx.channel().isOpen()) { Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(ctx.channel(), false); - ErrorMessage errorMessage = ErrorMessage.fromException(cause, handler); - ChannelFuture future = ctx.writeAndFlush(errorMessage.encode(getConnectionVersion(ctx))); - // On protocol exception, close the channel as soon as the message have been sent. - // Most cases of PE are wrapped so the type check below is expected to fail more often than not. - // At this moment Fatal exceptions are not thrown in v4, but just as a precaustion we check for them here - if (isFatal(cause)) - future.addListener((ChannelFutureListener) f -> ctx.close()); + // No request in scope at the channel level; a WrappedException cause carries the frame's + // stream id and overrides this fallback. + ErrorMessage.WithStreamId withStreamId = ErrorMessage.fromException(cause, handler); + + if (withStreamId.streamId == Message.UNSET_STREAM_ID) + { + // No stream id could be recovered, so we have no request to route a response to. + // Close the connection rather than emit an unroutable frame (CASSANDRA-21508). + ctx.close(); + } + else + { + ErrorMessage errorMessage = withStreamId.message; + int streamId = withStreamId.streamId; + + ChannelFuture future = ctx.writeAndFlush(errorMessage.encode(getConnectionVersion(ctx), streamId)); + // On protocol exception, close the channel as soon as the message have been sent. + // Most cases of PE are wrapped so the type check below is expected to fail more often than not. + // At this moment Fatal exceptions are not thrown in v4, but just as a precaustion we check for them here + if (isFatal(cause)) + future.addListener((ChannelFutureListener) f -> ctx.close()); + } } SocketAddress remoteAddress = ctx.channel().remoteAddress(); @@ -393,7 +413,8 @@ private static AuthenticationException maybeExtractAndWrapAuthenticationExceptio } } - private static ProtocolVersion getConnectionVersion(ChannelHandlerContext ctx) + @VisibleForTesting + static ProtocolVersion getConnectionVersion(ChannelHandlerContext ctx) { Connection connection = ctx.channel().attr(Connection.attributeKey).get(); // The only case the connection can be null is when we send the initial STARTUP message diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index 5151a30a96a7..aee18ac227cb 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -352,7 +352,7 @@ public Map execute(List requ for (int i = 0; i < requests.size(); i++) { Message.Request message = requests.get(i); - message.setStreamId(i); + message.setSource(new Envelope(Envelope.Header.dummy(i, message.type), null)); message.attach(connection); } lastWriteFuture = channel.writeAndFlush(requests); @@ -365,7 +365,7 @@ public Map execute(List requ throw new RuntimeException("timeout"); if (msg instanceof ErrorMessage) throw new RuntimeException((Throwable) ((ErrorMessage) msg).error); - rrMap.put(requests.get(msg.getStreamId()), msg); + rrMap.put(requests.get(msg.getSource().header.streamId), msg); } } else @@ -383,6 +383,18 @@ public Map execute(List requ } } + /** + * The stream id to frame an outbound client request with. SimpleClient carries the intended id on the + * request's (dummy) source envelope (see {@link #execute(List)} and callers that pipeline requests). + * When no source has been assigned, we fall back to 0, which is sufficient for the non-pipelined path + * where only a single request is ever in flight. + */ + private static int outboundStreamId(Message message) + { + Envelope source = message.getSource(); + return source == null ? 0 : source.header.streamId; + } + public interface EventHandler { void onEvent(Event event); @@ -438,37 +450,38 @@ private static class InitialHandler extends MessageToMessageDecoder this.largeMessageThreshold = largeMessageThreshold; } - protected void decode(ChannelHandlerContext ctx, Envelope response, List results) + @Override + protected void decode(ChannelHandlerContext ctx, Envelope request, List results) { - switch(response.header.type) + switch(request.header.type) { case READY: case AUTHENTICATE: - if (response.header.version.isGreaterOrEqualTo(ProtocolVersion.V5)) + if (request.header.version.isGreaterOrEqualTo(ProtocolVersion.V5)) { - configureModernPipeline(ctx, response, largeMessageThreshold); + configureModernPipeline(ctx, request, largeMessageThreshold); // consuming the message is done when setting up the pipeline } else { configureLegacyPipeline(ctx); // really just removes self from the pipeline, so pass this message on - ctx.pipeline().context(Envelope.Decoder.class).fireChannelRead(response); + ctx.pipeline().context(Envelope.Decoder.class).fireChannelRead(request); } break; case SUPPORTED: case ERROR: // just pass through - results.add(response); + results.add(request); break; default: - throw new ProtocolException(String.format("Unexpected %s response expecting " + + throw new ProtocolException(String.format("Unexpected %s request expecting " + "READY, AUTHENTICATE, ERROR or SUPPORTED", - response.header.type)); + request.header.type)); } } - private void configureModernPipeline(ChannelHandlerContext ctx, Envelope response, int largeMessageThreshold) + private void configureModernPipeline(ChannelHandlerContext ctx, Envelope request, int largeMessageThreshold) { logger.info("Configuring modern pipeline"); ChannelPipeline pipeline = ctx.pipeline(); @@ -490,7 +503,7 @@ private void configureModernPipeline(ChannelHandlerContext ctx, Envelope respons CQLMessageHandler.MessageConsumer responseConsumer = new CQLMessageHandler.MessageConsumer() { - public void dispatch(Channel channel, Message.Response message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure) + public

void dispatch(Channel channel, Message.Response message, Dispatcher.FlushItemConverter

toFlushItem, P param, Overload backpressure) { responseHandler.handleResponse(channel, message); } @@ -585,15 +598,15 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) ProtocolVersion version = connection == null ? ProtocolVersion.CURRENT : connection.getVersion(); SimpleFlusher flusher = new SimpleFlusher(frameEncoder, largeMessageThreshold); for (Message message : (List) msg) - flusher.enqueue(message.encode(version)); + flusher.enqueue(message.encode(version, outboundStreamId(message))); flusher.maybeWrite(ctx, promise); } }); pipeline.remove(this); - Message.Response message = messageDecoder.decode(ctx.channel(), response); - responseConsumer.dispatch(channel, message, (ch, req, resp) -> null, Overload.NONE); + Message.Response message = messageDecoder.decode(ctx.channel(), request); + responseConsumer.dispatch(channel, message, (p, ch, req, resp) -> null, null, Overload.NONE); } private FrameDecoder frameDecoder(ChannelHandlerContext ctx, BufferPoolAllocator allocator) @@ -638,7 +651,8 @@ public void encode(ChannelHandlerContext ctx, List messages, List unexpectedExceptionHandler) + { + return fromException(e, unexpectedExceptionHandler).message; + } + + public static class WithStreamId + { + public final ErrorMessage message; + public final int streamId; + + WithStreamId(ErrorMessage message, int streamId) + { + this.message = message; + this.streamId = streamId; + } + } /** * @param e the exception * @param unexpectedExceptionHandler a callback for handling unexpected exceptions. If null, or if this * returns false, the error is logged at ERROR level via sl4fj */ - public static ErrorMessage fromException(Throwable e, Predicate unexpectedExceptionHandler) + public static WithStreamId fromException(Throwable e, Predicate unexpectedExceptionHandler) { - int streamId = 0; + int streamId = UNSET_STREAM_ID; // Netty will wrap exceptions during decoding in a CodecException. If the cause was one of our ProtocolExceptions // or some other internal exception, extract that and use it. @@ -469,23 +493,15 @@ else if (e instanceof WrappedException) if (e instanceof TransportException) { - ErrorMessage message = new ErrorMessage((TransportException) e, streamId); - if (e instanceof ProtocolException) - { - // if the driver attempted to connect with a protocol version not supported then - // respond with the appropiate version, see ProtocolVersion.decode() - ProtocolVersion forcedProtocolVersion = ((ProtocolException) e).getForcedProtocolVersion(); - if (forcedProtocolVersion != null) - message.forcedProtocolVersion = forcedProtocolVersion; - } - return message; + ErrorMessage message = fromTransportException((TransportException) e); + return new WithStreamId(message, streamId); } // Unexpected exception if (unexpectedExceptionHandler == null || !unexpectedExceptionHandler.apply(e)) logger.error("Unexpected exception during request", e); - return new ErrorMessage(new ServerError(e), streamId); + return new WithStreamId(new ErrorMessage(new ServerError(e)), streamId); } @Override diff --git a/src/java/org/apache/cassandra/transport/messages/EventMessage.java b/src/java/org/apache/cassandra/transport/messages/EventMessage.java index 2a2df1dcebb2..77d958db5407 100644 --- a/src/java/org/apache/cassandra/transport/messages/EventMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/EventMessage.java @@ -25,6 +25,8 @@ public class EventMessage extends Message.Response { + public static final int EVENT_MESSAGE_STREAM_ID = -1; + public static final Message.Codec codec = new Message.Codec() { public EventMessage decode(ByteBuf body, ProtocolVersion version) @@ -49,7 +51,6 @@ public EventMessage(Event event) { super(Message.Type.EVENT); this.event = event; - this.setStreamId(-1); } @Override diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index 0f6afe0fe123..eb4c384af74c 100644 --- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java @@ -207,7 +207,7 @@ else if (options.skipMetadata()) { QueryEvents.instance.notifyExecuteFailure(prepared, options, state, e); JVMStabilityInspector.inspectThrowable(e); - return ErrorMessage.fromException(e); + return ErrorMessage.fromExceptionNoStreamId(e); } } diff --git a/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java b/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java index e92b69b16e00..baf562127eb6 100644 --- a/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java @@ -51,12 +51,13 @@ public PrepareMessage decode(ByteBuf body, ProtocolVersion version) { String query = CBUtil.readLongString(body); String keyspace = null; - if (version.isGreaterOrEqualTo(ProtocolVersion.V5)) { + if (version.isGreaterOrEqualTo(ProtocolVersion.V5)) + { // If flags grows, we may want to consider creating a PrepareOptions class with an internal codec // class that handles flags and options of the prepare message. Since there's only one right now, // we just take care of business here. - int flags = (int)body.readUnsignedInt(); + int flags = (int) body.readUnsignedInt(); if ((flags & 0x1) == 0x1) { keyspace = CBUtil.readString(body); @@ -74,8 +75,11 @@ public void encode(PrepareMessage msg, ByteBuf dest, ProtocolVersion version) { // If we have no keyspace, write out a 0-valued flag field. if (msg.keyspace == null) + { dest.writeInt(0x0); - else { + } + else + { dest.writeInt(0x1); CBUtil.writeAsciiString(msg.keyspace, dest); } @@ -134,7 +138,7 @@ protected Message.Response execute(QueryState state, Dispatcher.RequestTime requ { QueryEvents.instance.notifyPrepareFailure(null, query, state, e); JVMStabilityInspector.inspectThrowable(e); - return ErrorMessage.fromException(e); + return ErrorMessage.fromExceptionNoStreamId(e); } } diff --git a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java index 61f3a9cfbd83..d858266d7fff 100644 --- a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java @@ -129,7 +129,7 @@ protected Message.Response execute(QueryState state, Dispatcher.RequestTime requ JVMStabilityInspector.inspectThrowable(e); if (!((e instanceof RequestValidationException) || (e instanceof RequestExecutionException))) logger.error("Unexpected error during query", e); - return ErrorMessage.fromException(e); + return ErrorMessage.fromExceptionNoStreamId(e); } } diff --git a/test/burn/org/apache/cassandra/transport/DriverBurnTest.java b/test/burn/org/apache/cassandra/transport/DriverBurnTest.java index 236d9b60dc2c..601ac6be4fb1 100644 --- a/test/burn/org/apache/cassandra/transport/DriverBurnTest.java +++ b/test/burn/org/apache/cassandra/transport/DriverBurnTest.java @@ -340,10 +340,10 @@ public void perfTest(SizeCaps requestCaps, SizeCaps responseCaps, Cluster.Builde SimpleStatement request = generateQueryStatement(0, requestCaps); ResultMessage.Rows response = generateRows(0, responseCaps); QueryMessage requestMessage = generateQueryMessage(0, requestCaps, version); - Envelope message = requestMessage.encode(version); + Envelope message = requestMessage.encode(version, 0); int requestSize = message.body.readableBytes(); message.release(); - message = response.encode(version); + message = response.encode(version, 0); int responseSize = message.body.readableBytes(); message.release(); Message.Type.QUERY.unsafeSetCodec(new Message.Codec() { diff --git a/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java b/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java index cfd54f34e148..5db4b9d928e2 100644 --- a/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java +++ b/test/burn/org/apache/cassandra/transport/SimpleClientPerfTest.java @@ -155,10 +155,10 @@ public void perfTest(SizeCaps requestCaps, SizeCaps responseCaps, AssertUtil.Thr { ResultMessage.Rows response = generateRows(0, responseCaps); QueryMessage requestMessage = generateQueryMessage(0, requestCaps, version); - Envelope message = requestMessage.encode(version); + Envelope message = requestMessage.encode(version, 0); int requestSize = message.body.readableBytes(); message.release(); - message = response.encode(version); + message = response.encode(version, 0); int responseSize = message.body.readableBytes(); message.release(); diff --git a/test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java b/test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java new file mode 100644 index 000000000000..430fdb6de05c --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java @@ -0,0 +1,254 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import net.bytebuddy.ByteBuddy; +import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; +import net.bytebuddy.implementation.MethodDelegation; +import net.bytebuddy.implementation.bind.annotation.SuperCall; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.cql3.QueryOptions; +import org.apache.cassandra.cql3.statements.SelectStatement; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.exceptions.OverloadedException; +import org.apache.cassandra.service.QueryState; +import org.apache.cassandra.transport.Dispatcher; +import org.apache.cassandra.transport.Envelope; +import org.apache.cassandra.transport.Message; +import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.transport.SimpleClient; +import org.apache.cassandra.transport.messages.ErrorMessage; +import org.apache.cassandra.transport.messages.QueryMessage; +import org.apache.cassandra.transport.messages.ResultMessage; + +import static net.bytebuddy.matcher.ElementMatchers.named; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; +import static org.apache.cassandra.config.DatabaseDescriptor.clientInitialization; +import static org.apache.cassandra.config.DatabaseDescriptor.getNativeTransportPort; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NATIVE_PROTOCOL; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; + +/** + * Regression test for CASSANDRA-21508. A coordinator that load-sheds a request which waited past its + * native-transport queue deadline must stamp the OVERLOADED error with the timed-out request's own + * stream id. Before the fix the error went out on stream id 0, which misroutes it to an unrelated + * in-flight request and can escalate into the client-side "column-shift" read corruption (a value + * from one table decoded against another query's column definitions under the v5 skip-metadata + * optimization). + * + *

The original defect: {@code Dispatcher.processRequest} load-sheds a request that has waited in the + * Native-Transport-Requests queue longer than {@code native_transport_timeout} by returning + * + * 

+ *   ErrorMessage.fromException(new OverloadedException("Query timed out before it could start"))
+ *
+ *

+ * without calling {@code setStreamId(request.getStreamId())}. {@code ErrorMessage.streamId} therefore + * kept its default of 0, and the error frame was written to the client on stream id 0 instead of the + * timed-out request's real stream id. The fix routes every response through a central stamping step + * and makes {@code ErrorMessage.fromException} require the stream id at the call site. + * + *

{@link #loadShedErrorIsStampedWithRequestStreamId()} proves the fix on the wire: a request sent on + * a non-zero stream id is load-shed and the OVERLOADED error comes back on that same stream id (not 0). + * Against the unpatched server this assertion fails with the error arriving on stream id 0. + * + *

Why the stream id matters: on a busy connection stream id 0 is almost always in use, so an error + * mis-stamped with 0 is applied to an unrelated in-flight request. The client then frees and reuses + * stream id 0 for a new query, and when the original query's real rows arrive on it they are decoded + * positionally against the reusing query's cached column definitions (skip-metadata carries no column + * info) - shifting each value into the wrong column and either throwing in a codec or silently + * returning a plausible-but-wrong value. + */ +public class StreamIdMisrouteTest extends TestBaseImpl +{ + private static final int TIMED_OUT_STREAM_ID = 42; // any non-zero id; before the fix the reply was forced to 0 + + @BeforeClass + public static void initClientSide() + { + // SimpleClient runs in the test classloader; it needs DatabaseDescriptor initialized here to + // build the native-protocol pipeline. Without this, connect() fails with a ClosedChannelException. + clientInitialization(); + } + + /** + * Proves the fix directly: under native-transport queue backlog, the load-shed OVERLOADED error is + * returned on the stream id of the request that actually timed out, not on stream id 0. Against the + * unpatched server this fails with the error arriving on stream id 0. + */ + @Test + public void loadShedErrorIsStampedWithRequestStreamId() throws Throwable + { + try (Cluster cluster = init(Cluster.build().withNodes(1) + .withInstanceInitializer(SlowSelect::install) + .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL) + // one NTR thread so a slow query blocks the queue head + .set("native_transport_max_threads", 1) + // short deadline so a queued request is shed quickly + .set("native_transport_timeout", "150ms") + .set("read_request_timeout", "500ms") + .set("range_request_timeout", "500ms")) + .start())) + { + cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (pk int PRIMARY KEY, v int)")); + + InetSocketHost host = hostAndPort(cluster); + ExecutorService executor = Executors.newCachedThreadPool(); + + // Two independent connections. A SimpleClient's response queue is a SynchronousQueue, so + // concurrent execute() calls on ONE client would race; we use one client to build the + // backlog and a second, dedicated client for the request whose stream id we assert on. + try (SimpleClient filler = SimpleClient.builder(host.address, host.port) + .protocolVersion(ProtocolVersion.V5).build().connect(false); + SimpleClient victim = SimpleClient.builder(host.address, host.port) + .protocolVersion(ProtocolVersion.V5).build().connect(false)) + { + // Warm up both connections while the server is still fast. + filler.execute(query(withKeyspace("SELECT * FROM %s.tbl"), 1)); + victim.execute(query(withKeyspace("SELECT * FROM %s.tbl"), 1)); + + // From here every non-internal SELECT sleeps 1s, far past the 150ms queue deadline. + cluster.get(1).runOnInstance(() -> Assert.assertTrue(SlowSelect.enabled.compareAndSet(false, true))); + + // Saturate the single NTR worker and pile several requests behind it, so anything + // enqueued now waits well beyond native_transport_timeout before a worker frees up. + List> backlog = new ArrayList<>(); + for (int i = 0; i < 8; i++) + { + int streamId = 10 + i; + backlog.add(executor.submit(() -> + filler.execute(query(withKeyspace("SELECT * FROM %s.tbl"), streamId), false))); + } + + // Let the backlog form and the queue-time clock run past the 150ms deadline. + TimeUnit.MILLISECONDS.sleep(800); + + // This request enqueues behind a queue that is already older than the deadline, so the + // worker load-sheds it immediately when it is dequeued. It goes out on stream id 42. + Message.Response shed = + victim.execute(query(withKeyspace("SELECT * FROM %s.tbl"), TIMED_OUT_STREAM_ID), false); + + Assert.assertTrue("Expected an OVERLOADED error, got: " + shed, + shed instanceof ErrorMessage + && ((ErrorMessage) shed).error instanceof OverloadedException); + + // The fix: the request went out on stream id 42, and the load-shed error now comes back + // on stream id 42 too because Dispatcher stamps every response with the request's id. + Assert.assertEquals("Load-shed OVERLOADED error should carry the request's own stream id (" + + TIMED_OUT_STREAM_ID + ')', + TIMED_OUT_STREAM_ID, shed.getSource().header.streamId); + + // Drain the backlog so the connection can close cleanly. + cluster.get(1).runOnInstance(() -> SlowSelect.enabled.set(false)); + for (Future f : backlog) + { + try + { + f.get(30, TimeUnit.SECONDS); + } + catch (Exception ignored) + { /* shed/timed out */ } + } + } + finally + { + cluster.get(1).runOnInstance(() -> SlowSelect.enabled.set(false)); + executor.shutdownNow(); + } + } + } + + // ------------------------------------------------------------------------------------------------ + // Helpers + // ------------------------------------------------------------------------------------------------ + + private static QueryMessage query(String cql, int streamId) + { + QueryMessage msg = new QueryMessage(cql, QueryOptions.forInternalCalls( + org.apache.cassandra.db.ConsistencyLevel.ONE, Collections.emptyList())); + msg.setSource(new Envelope(Envelope.Header.dummy(streamId, Message.Type.QUERY), null)); + return msg; + } + + private static InetSocketHost hostAndPort(Cluster cluster) + { + // Node 1's native transport binds on its broadcast address; the port is whatever the in-JVM + // provisioning assigned (9042 for the default multi-interface strategy). Derive both from the + // instance config so this works regardless of provisioning strategy. + String address = cluster.get(1).config().broadcastAddress().getAddress().getHostAddress(); + int port = cluster.get(1).callOnInstance( + () -> getNativeTransportPort()); + return new InetSocketHost(address, port); + } + + private static final class InetSocketHost + { + final String address; + final int port; + + InetSocketHost(String address, int port) + { + this.address = address; + this.port = port; + } + } + + /** + * ByteBuddy interceptor that makes client-issued SELECTs sleep past the queue deadline, so a + * request queued behind one is load-shed by Dispatcher. Mirrors OverloadTest.SlowSelect. + */ + public static class SlowSelect + { + static final AtomicBoolean enabled = new AtomicBoolean(false); + + static void install(ClassLoader cl, int nodeNumber) + { + new ByteBuddy().rebase(SelectStatement.class) + .method(named("execute").and(takesArguments(QueryState.class, QueryOptions.class, Dispatcher.RequestTime.class))) + .intercept(MethodDelegation.to(SlowSelect.class)) + .make() + .load(cl, ClassLoadingStrategy.Default.INJECTION); + } + + @SuppressWarnings("unused") + public static ResultMessage.Rows execute(QueryState state, QueryOptions options, + Dispatcher.RequestTime requestTime, + @SuperCall Callable zuper) throws Exception + { + if (enabled.get() && !state.getClientState().isInternal) + TimeUnit.SECONDS.sleep(1); + return zuper.call(); + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/UnableToParseClientMessageTest.java b/test/distributed/org/apache/cassandra/distributed/test/UnableToParseClientMessageTest.java index 2e2bce961141..f6755dae3df5 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/UnableToParseClientMessageTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/UnableToParseClientMessageTest.java @@ -185,9 +185,9 @@ public CustomHeaderMessage(ByteBuf headerEncoded) } @Override - public Envelope encode(ProtocolVersion version) + public Envelope encode(ProtocolVersion version, int streamId) { - Envelope base = super.encode(version); + Envelope base = super.encode(version, streamId); return new CustomHeaderEnvelope(base.header, base.body, headerEncoded); } } @@ -228,7 +228,7 @@ protected CustomBodyMessage(Type type, ByteBuf body) } @Override - public Envelope encode(ProtocolVersion version) + public Envelope encode(ProtocolVersion version, int streamId) { Codec originalCodec = type.codec; try @@ -255,7 +255,7 @@ public int encodedSize(Message message, ProtocolVersion version) } }); - return super.encode(version); + return super.encode(version, streamId); } finally { diff --git a/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java index 6780e4206e7d..1676850e06ba 100644 --- a/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java +++ b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java @@ -20,7 +20,9 @@ import java.io.IOException; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.ServerSocket; +import java.net.SocketAddress; import java.nio.ByteBuffer; import java.security.SecureRandom; import java.util.HashMap; @@ -81,18 +83,24 @@ import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; +import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPipeline; +import io.netty.channel.ChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.handler.codec.MessageToMessageDecoder; +import io.netty.handler.codec.MessageToMessageEncoder; import static org.apache.cassandra.config.EncryptionOptions.TlsEncryptionPolicy.UNENCRYPTED; import static org.apache.cassandra.io.util.FileUtils.ONE_MIB; import static org.apache.cassandra.net.FramingTest.randomishBytes; import static org.apache.cassandra.transport.Flusher.MAX_FRAMED_PAYLOAD_SIZE; +import static org.apache.cassandra.transport.PreV5Handlers.getConnectionVersion; import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition; import static org.apache.cassandra.utils.concurrent.NonBlockingRateLimiter.NO_OP_LIMITER; import static org.assertj.core.api.Assertions.assertThat; @@ -176,6 +184,45 @@ public void handleErrorDuringNegotiation() throws Throwable observer.verifier().accept(0); } + @Test + public void handleIncompleteHeaderErrorDuringNegotiation() throws Throwable + { + // CASSANDRA-21508: a partial header (fewer than Header.LENGTH=9 bytes) carrying an unsupported + // protocol version gives the server no stream id to route an error back to. Rather than emit an + // unroutable error frame (which could be misapplied to another request), the server closes the + // connection. Here the client observes a closed connection with no error frame. + int messageCount = 0; + Codec codec = Codec.crc(alloc); + AllocationObserver observer = new AllocationObserver(); + InboundProxyHandler.Controller controller = new InboundProxyHandler.Controller(); + // Truncate the client's STARTUP to a partial header (< 9 bytes) and set an unsupported version. + controller.withPayloadTransform(msg -> { + ByteBuf bb = (ByteBuf) msg; + ByteBuf truncated = bb.copy(0, 4); + truncated.setByte(0, 99 & Envelope.PROTOCOL_VERSION_MASK); + bb.release(); + return truncated; + }); + + ServerConfigurator configurator = ServerConfigurator.builder() + .withAllocationObserver(observer) + .withProxyController(controller) + .build(); + Server server = server(configurator); + Client client = new Client(codec, messageCount); + server.start(); + client.connect(address, port); + + // The server cannot recover a stream id from the incomplete header, so it closes the connection + // without sending an error frame. + assertFalse(client.isConnected()); + assertThat(client.getConnectionError()).isNull(); + server.stop(); + + // the failure happens before any capacity is allocated + observer.verifier().accept(0); + } + @Test public void handleFrameCorruptionAfterNegotiation() throws Throwable { @@ -247,6 +294,43 @@ public ByteBuf apply(ByteBuf msg) testFrameCorruption(1, Codec.crc(alloc), envelopeProvider, corruptor, totalBytesPerEnvelope, errorCheck); } + @Test + public void fatalErrorFrameIsFlushedBeforeChannelClose() + { + // CASSANDRA-21508: the post-V5 exception handler must close the connection only AFTER the diagnostic + // error frame has been flushed. If it closes synchronously right after writeAndFlush, a write that + // can't drain immediately (TCP backpressure, TLS, large frame) is aborted and the client never sees + // the error. Simulate "write cannot complete synchronously" with an outbound handler that captures + // the write promise without completing it - the handler must not have closed the channel yet. + StallingOutboundHandler stall = new StallingOutboundHandler(); + EmbeddedChannel channel = new EmbeddedChannel() + { + // client_error_reporting_exclusions (SubnetGroups.contains) requires a real InetSocketAddress; + // EmbeddedChannel's default remote address is not one, so supply a loopback address. + @Override + protected SocketAddress remoteAddress0() + { + return new InetSocketAddress("127.0.0.1", 9042); + } + }; + channel.pipeline().addLast(stall); + channel.pipeline().addLast(ExceptionHandlers.postV5Handler(FrameEncoderCrc.instance.allocator(), + ProtocolVersion.V5)); + + // Fatal protocol error, no recoverable stream id -> handler encodes an error frame and fatally closes. + channel.pipeline().fireExceptionCaught(ProtocolException.toFatalException(new ProtocolException("boom"))); + + // The error frame was handed to the outbound pipeline... + assertNotNull("expected the handler to write an error frame", stall.writePromise); + // ...but since that write has not completed, the channel must still be open. The buggy synchronous + // ctx.close() would have already closed it here. + assertTrue("channel must not close before the error frame is flushed", channel.isOpen()); + + // Once the write completes, the deferred close fires. + stall.writePromise.setSuccess(); + assertFalse("channel should close after the error frame is flushed", channel.isOpen()); + } + @Test public void testAquireAndRelease() { @@ -322,6 +406,48 @@ private void testEnvelopeDecodingErrors(int messageCount, IntPredicate shouldErr runTest(configurator, codec, messageCount, envelopeProvider, responseMatcher, observer.verifier()); } + @Test + public void testRecoverableBetaFlagEnvelopeErrors() + { + // CASSANDRA-21508: a V6 (beta) envelope header with the USE_BETA flag unset is an + // invalid but *recoverable* protocol error, exactly like an unknown opcode. The server should + // return an ERROR on the offending stream id and continue processing subsequent envelopes on the + // same connection. + + // every other message advertises V6 with USE_BETA unset and should error while extracting the header + IntPredicate shouldError = i -> i % 2 == 0; + testBetaFlagEnvelopeErrors(10, shouldError, Codec.crc(alloc)); + testBetaFlagEnvelopeErrors(10, shouldError, Codec.lz4(alloc)); + + testBetaFlagEnvelopeErrors(100, shouldError, Codec.crc(alloc)); + } + + private void testBetaFlagEnvelopeErrors(int messageCount, IntPredicate shouldError, Codec codec) + { + TestConsumer consumer = new TestConsumer(new ResultMessage.Void(), codec.encoder); + AllocationObserver observer = new AllocationObserver(false); + Message.Decoder decoder = new FixedDecoder(); + + // Mutate the erroring streams' headers to advertise the V6 (beta) version with the USE_BETA flag + // cleared. In the envelope header, byte 0 is the (direction & version) byte and byte 1 is the flags. + int betaBit = 1 << Envelope.Header.Flag.USE_BETA.ordinal(); + IntFunction envelopeProvider = mutatedEnvelopeProvider(shouldError, b -> { + b.put(0, (byte) ProtocolVersion.V6.asInt()); // REQUEST direction, V6 (beta) + b.put(1, (byte) (b.get(1) & ~betaBit)); // clear USE_BETA + }); + + Predicate responseMatcher = + h -> (shouldError.test(h.streamId) && h.type == Message.Type.ERROR) || h.type == Message.Type.RESULT; + + ServerConfigurator configurator = ServerConfigurator.builder() + .withConsumer(consumer) + .withAllocationObserver(observer) + .withDecoder(decoder) + .build(); + + runTest(configurator, codec, messageCount, envelopeProvider, responseMatcher, observer.verifier()); + } + @Test public void testUnrecoverableEnvelopeDecodingErrors() { @@ -580,6 +706,19 @@ private void flipBit(ByteBuf buf, int index) buf.setByte(index, buf.getByte(index) ^ (1 << 4)); } + private static class StallingOutboundHandler extends ChannelOutboundHandlerAdapter + { + volatile ChannelPromise writePromise; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) + { + // Capture the write but neither complete nor forward it: simulates a socket that cannot drain + // synchronously. The handler's own finally releases the payload, so we don't touch msg. + writePromise = promise; + } + } + private static class MutableEnvelope extends Envelope { public MutableEnvelope(Envelope source) @@ -639,7 +778,6 @@ Message.Request decode(Channel channel, Envelope source) Message.Request request = new OptionsMessage(); request.setSource(source); - request.setStreamId(source.header.streamId); Connection connection = channel.attr(Connection.attributeKey).get(); request.attach(connection); @@ -658,17 +796,17 @@ static class TestConsumer implements MessageConsumer TestConsumer(Message.Response fixedResponse, FrameEncoder frameEncoder) { this.fixedResponse = fixedResponse; - this.responseTemplate = fixedResponse.encode(ProtocolVersion.V5); + this.responseTemplate = fixedResponse.encode(ProtocolVersion.V5, 0); this.frameEncoder = frameEncoder; } - public void dispatch(Channel channel, Message.Request message, Dispatcher.FlushItemConverter toFlushItem, Overload backpressure) + public

void dispatch(Channel channel, Message.Request message, Dispatcher.FlushItemConverter

toFlushItem, P param, Overload backpressure) { if (flusher == null) flusher = new SimpleClient.SimpleFlusher(frameEncoder); Envelope response = Envelope.create(responseTemplate.header.type, - message.getStreamId(), + message.getSource().header.streamId, ProtocolVersion.V5, responseTemplate.header.flags, responseTemplate.body.copy()); @@ -677,7 +815,7 @@ public void dispatch(Channel channel, Message.Request message, Dispatcher.FlushI // and flush them to the outbound pipeline flusher.schedule(channel.pipeline().lastContext()); // this simulates the release of the allocated resources that a real flusher would do - Flusher.FlushItem.Framed item = (Flusher.FlushItem.Framed)toFlushItem.toFlushItem(channel, message, fixedResponse); + Flusher.FlushItem.Framed item = (Flusher.FlushItem.Framed)toFlushItem.toFlushItem(param, channel, message, fixedResponse); item.release(); } @@ -984,6 +1122,21 @@ static Codec crc(BufferPoolAllocator alloc) } } + /** + * Simple adaptor to plug CQL message encoding into pre-V5 pipelines + */ + @ChannelHandler.Sharable + public static class ProtocolEncoder extends MessageToMessageEncoder + { + public static final ProtocolEncoder instance = new ProtocolEncoder(); + private ProtocolEncoder(){} + public void encode(ChannelHandlerContext ctx, Message source, List results) + { + ProtocolVersion version = getConnectionVersion(ctx); + results.add(source.encode(version, 0)); + } + } + static class Client { private final Codec codec; @@ -1022,7 +1175,7 @@ protected void initChannel(Channel channel) throws Exception ChannelPipeline pipeline = channel.pipeline(); // Outbound handlers to enable us to send the initial STARTUP pipeline.addLast("envelopeEncoder", Envelope.Encoder.instance); - pipeline.addLast("messageEncoder", PreV5Handlers.ProtocolEncoder.instance); + pipeline.addLast("messageEncoder", ProtocolEncoder.instance); pipeline.addLast("envelopeDecoder", new Envelope.Decoder()); // Inbound handler to perform the handshake & modify the pipeline on receipt of a READY pipeline.addLast("handshake", new MessageToMessageDecoder() @@ -1100,6 +1253,17 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) th flusher.schedule(channel.pipeline().lastContext()); ready.countDown(); } + + @Override + public void channelInactive(ChannelHandlerContext ctx) + { + // If the server closes the connection during negotiation (e.g. an unroutable + // protocol error that cannot be answered with a stream id), unblock connect() + // so the test observes the disconnection rather than hanging for a READY/ERROR. + connected = false; + ready.countDown(); + ctx.fireChannelInactive(); + } }); } }); diff --git a/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java b/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java index c0e3f5ed91ef..7c708e7f8e27 100644 --- a/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java +++ b/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java @@ -67,7 +67,7 @@ public void testV5ReadFailureSerDeser() boolean dataPresent = false; ReadFailureException rfe = new ReadFailureException(consistencyLevel, receivedBlockFor, receivedBlockFor, dataPresent, failureReasonMap1); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(rfe), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(rfe), ProtocolVersion.V5); ReadFailureException deserializedRfe = (ReadFailureException) deserialized.error; assertEquals(failureReasonMap1, deserializedRfe.failureReasonByEndpoint); @@ -85,7 +85,7 @@ public void testV5WriteFailureSerDeser() WriteType writeType = WriteType.SIMPLE; WriteFailureException wfe = new WriteFailureException(consistencyLevel, receivedBlockFor, receivedBlockFor, writeType, failureReasonMap2); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(wfe), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(wfe), ProtocolVersion.V5); WriteFailureException deserializedWfe = (WriteFailureException) deserialized.error; assertEquals(failureReasonMap2, deserializedWfe.failureReasonByEndpoint); @@ -103,7 +103,7 @@ public void testV5CasWriteTimeoutSerDeser() ConsistencyLevel consistencyLevel = ConsistencyLevel.SERIAL; CasWriteTimeoutException ex = new CasWriteTimeoutException(WriteType.CAS, consistencyLevel, 0, receivedBlockFor, contentions); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(ex), ProtocolVersion.V5); assertTrue(deserialized.error instanceof CasWriteTimeoutException); CasWriteTimeoutException deserializedEx = (CasWriteTimeoutException) deserialized.error; @@ -124,7 +124,7 @@ public void testV4CasWriteTimeoutSerDeser() ConsistencyLevel consistencyLevel = ConsistencyLevel.SERIAL; CasWriteTimeoutException ex = new CasWriteTimeoutException(WriteType.CAS, consistencyLevel, receivedBlockFor, receivedBlockFor, contentions); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V4); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(ex), ProtocolVersion.V4); assertTrue(deserialized.error instanceof WriteTimeoutException); assertFalse(deserialized.error instanceof CasWriteTimeoutException); WriteTimeoutException deserializedEx = (WriteTimeoutException) deserialized.error; @@ -142,7 +142,7 @@ public void testV5CasWriteResultUnknownSerDeser() ConsistencyLevel consistencyLevel = ConsistencyLevel.SERIAL; CasWriteUnknownResultException ex = new CasWriteUnknownResultException(consistencyLevel, receivedBlockFor, receivedBlockFor); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(ex), ProtocolVersion.V5); assertTrue(deserialized.error instanceof CasWriteUnknownResultException); CasWriteUnknownResultException deserializedEx = (CasWriteUnknownResultException) deserialized.error; @@ -160,7 +160,7 @@ public void testV4CasWriteResultUnknownSerDeser() ConsistencyLevel consistencyLevel = ConsistencyLevel.SERIAL; CasWriteUnknownResultException ex = new CasWriteUnknownResultException(consistencyLevel, receivedBlockFor, receivedBlockFor); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V4); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromTransportException(ex), ProtocolVersion.V4); assertTrue(deserialized.error instanceof WriteTimeoutException); assertFalse(deserialized.error instanceof CasWriteUnknownResultException); WriteTimeoutException deserializedEx = (WriteTimeoutException) deserialized.error; diff --git a/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java b/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java index 601a1b1cdbc4..5f73c5d54405 100644 --- a/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java +++ b/test/unit/org/apache/cassandra/transport/MessageDispatcherTest.java @@ -157,7 +157,7 @@ public long tryAuth(Callable check) throws Exception public long tryAuth(Callable check, Message.Request request) throws Exception { long start = check.call(); - dispatch.dispatch(null, request, (channel,req,response) -> null, ClientResourceLimits.Overload.NONE); + dispatch.dispatch(null, request, (p, channel, req, response) -> null, null, ClientResourceLimits.Overload.NONE); // While this is timeout based, we should be *well below* a full second on any of this processing in any sane environment. long timeout = System.currentTimeMillis(); @@ -176,9 +176,10 @@ public AuthTestDispatcher() } @Override - void processRequest(Channel channel, +

void processRequest(Channel channel, Message.Request request, - FlushItemConverter forFlusher, + FlushItemConverter

forFlusher, + P param, ClientResourceLimits.Overload backpressure, RequestTime requestTime) { diff --git a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java index a0c6694b6e2b..616536109e53 100644 --- a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java +++ b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.transport; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; @@ -73,7 +74,12 @@ public void testInvalidProtocolVersion(int version) throws Exception try { dec.decode(null, buf, results); Assert.fail("Expected protocol error"); - } catch (ProtocolException e) { + } catch (ErrorMessage.WrappedException e) { + // The stream id is recovered using the attempted version's header layout (CASSANDRA-21508): + // v1/v2 use a single-byte stream id (header byte 2, which is 0x00 here), whereas v3+ use a + // two-byte id (0x0001). See decodeRecoversSingleByteStreamIdForOldProtocolVersions. + int expectedStreamId = version < ProtocolVersion.V3.asInt() ? 0x00 : 0x01; + Assert.assertEquals(expectedStreamId, e.getStreamId()); Assert.assertTrue(e.getMessage().contains("Invalid or unsupported protocol version")); } } @@ -101,6 +107,108 @@ public void testInvalidProtocolVersionShortBody() throws Exception } } + @Test + public void testIncompleteHeaderWithInvalidProtocolVersion() throws Exception + { + // CASSANDRA-21508: when fewer than a full header's worth of bytes have arrived AND the protocol + // version is unsupported, decode cannot trust/recover a stream id. It defers the protocol error and + // wraps it with Message.UNSET_STREAM_ID, so the channel-level exception handler closes the connection + // rather than emit an unroutable error frame. Exercises Envelope.Decoder.decode() lines 387-398. + Envelope.Decoder dec = new Envelope.Decoder(); + + List results = new ArrayList<>(); + // Unsupported version (two above CURRENT) with only part of a header - fewer than Header.LENGTH (9) bytes. + byte[] bytes = new byte[] { + (byte) REQUEST.addToVersion(ProtocolVersion.CURRENT.asInt() + 2), // direction & unsupported version + 0x00, // flags + 0x00, 0x2a, // partial header, truncated before the length field + }; + ByteBuf buf = Unpooled.wrappedBuffer(bytes); + try { + dec.decode(null, buf, results); + Assert.fail("Expected protocol error"); + } catch (ProtocolException e) { + Assert.assertTrue(e.getMessage().contains("Invalid or unsupported protocol version")); + } + } + + @Test + public void extractHeaderReturnsRecoverableErrorOnBetaFlagViolation() throws Exception + { + // CASSANDRA-21508: on the framed (V5+) path, a V6 (beta) envelope whose USE_BETA flag is + // unset is an invalid but *recoverable* protocol error. Envelope.Decoder.extractHeader documents that + // it never throws, and instead returns a HeaderExtractionResult carrying the frame's stream id, so the + // caller (CQLMessageHandler.processOneContainedMessage) can route an ERROR back on that stream and keep + // processing subsequent envelopes. + Envelope.Decoder dec = new Envelope.Decoder(); + + int streamId = 42; + // A complete 9-byte header advertising V6 (beta) as a REQUEST, with USE_BETA deliberately unset. + byte[] header = new byte[] { + (byte) REQUEST.addToVersion(ProtocolVersion.V6.asInt()), // direction & beta version + 0x00, // flags - USE_BETA deliberately unset + 0x00, 0x2a, // stream id = 42 + 0x05, // opcode = OPTIONS + 0x00, 0x00, 0x00, 0x00, // body length = 0 + }; + ByteBuffer buf = ByteBuffer.wrap(header); + + Envelope.Decoder.HeaderExtractionResult result; + try + { + result = dec.extractHeader(buf); + } + catch (ErrorMessage.WrappedException e) + { + throw new AssertionError("extractHeader must not throw on a USE_BETA violation; a recoverable " + + "protocol error escaped as a WrappedException and would tear down the " + + "connection instead of returning a routable error", e); + } + + Assert.assertFalse("A USE_BETA violation should be reported as a (recoverable) extraction error", + result.isSuccess()); + Assert.assertEquals("the frame's stream id must be preserved so the error can be routed back", + streamId, result.streamId()); + Assert.assertTrue("expected a USE_BETA protocol error, got: " + result.error().getMessage(), + result.error().getMessage().contains("USE_BETA flag is unset")); + } + + @Test + public void decodeRecoversSingleByteStreamIdForOldProtocolVersions() throws Exception + { + // CASSANDRA-21508: protocol versions 1 and 2 use a shorter header whose stream id is a single + // byte. When such a version is rejected, decode must recover the stream id using that layout. + // Reading a 16-bit value would splice the stream id byte together with the following opcode byte and + // recover a bogus id (e.g. stream 0 + STARTUP opcode 1 -> stream 1), routing the error to a stream the + // client never used. A v1/v2 client (which downgrades on such an error) would then never see it and + // its connection would time out - as observed in ProtocolNegotiationTest#olderVersionsAreUnsupported. + Envelope.Decoder dec = new Envelope.Decoder(); + + int streamId = 0x07; + List results = new ArrayList<>(); + // A full (>= Header.LENGTH bytes) v1 frame so decode reaches the stream-id recovery, not the + // incomplete-header path: single-byte stream id, STARTUP opcode, and one body byte. + byte[] bytes = new byte[] { + (byte) REQUEST.addToVersion(1), // unsupported v1 + 0x00, // flags + (byte) streamId, // v1/v2 single-byte stream id + 0x01, // opcode = STARTUP (would be spliced in as the low byte) + 0x00, 0x00, 0x00, 0x01, // body length = 1 + 0x00 // 1 body byte, so readableBytes == Header.LENGTH + }; + ByteBuf buf = Unpooled.wrappedBuffer(bytes); + try { + dec.decode(null, buf, results); + Assert.fail("Expected protocol error"); + } catch (ErrorMessage.WrappedException e) { + Assert.assertEquals("a v1/v2 stream id must be read as a single byte, not spliced with the opcode", + streamId, e.getStreamId()); + Assert.assertTrue("expected a ProtocolException cause, got: " + e.getCause(), + e.getCause() instanceof ProtocolException); + Assert.assertTrue(e.getMessage().contains("Invalid or unsupported protocol version")); + } + } + @Test public void testInvalidDirection() throws Exception { @@ -161,7 +269,7 @@ public void testBodyLengthOverLimit() throws Exception public void testErrorMessageWithNullString() { // test for CASSANDRA-11167 - ErrorMessage msg = ErrorMessage.fromException(new ServerError((String) null)); + ErrorMessage msg = ErrorMessage.fromTransportException(new ServerError((String) null)); assert msg.toString().endsWith("null") : msg.toString(); int size = ErrorMessage.codec.encodedSize(msg, ProtocolVersion.CURRENT); ByteBuf buf = Unpooled.buffer(size); diff --git a/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java b/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java index 7611b513796d..d41e4244363a 100644 --- a/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java +++ b/test/unit/org/apache/cassandra/transport/ProtocolNegotiationTest.java @@ -126,29 +126,31 @@ private void testStreamIdsAcrossNegotiation(ProtocolVersion version) for (int i = 0; i < 100; i++) { int streamId = random.nextInt(254) + 1; - options.setStreamId(streamId); + options.setSource(new Envelope(Envelope.Header.dummy(streamId, options.type), null)); Message.Response response = client.execute(options); + // The stream id is stamped onto the frame at encoding time; verify it round-trips by + // reading it back off the response's source envelope. assertEquals(String.format("StreamId mismatch; version: %s, seed: %s, iter: %s, expected: %s, actual: %s", - version, seed, i, streamId, response.getStreamId()), - streamId, response.getStreamId()); + version, seed, i, streamId, response.getSource().header.streamId), + streamId, response.getSource().header.streamId); } int streamId = random.nextInt(254) + 1; // STARTUP messages are handled by the initial connection handler StartupMessage startup = new StartupMessage(ImmutableMap.of(CQL_VERSION, QueryProcessor.CQL_VERSION.toString())); - startup.setStreamId(streamId); + startup.setSource(new Envelope(Envelope.Header.dummy(streamId, startup.type), null)); Message.Response response = client.execute(startup); assertEquals(String.format("StreamId mismatch after negotiation; version: %s, expected: %s, actual %s", - version, streamId, response.getStreamId()), - streamId, response.getStreamId()); + version, streamId, response.getSource().header.streamId), + streamId, response.getSource().header.streamId); // Following STARTUP, the version specific handlers are fully responsible for processing messages QueryMessage query = new QueryMessage("SELECT * FROM system.local", QueryOptions.DEFAULT); - query.setStreamId(streamId); + query.setSource(new Envelope(Envelope.Header.dummy(streamId, query.type), null)); response = client.execute(query); assertEquals(String.format("StreamId mismatch after negotiation; version: %s, expected: %s, actual %s", - version, streamId, response.getStreamId()), - streamId, response.getStreamId()); + version, streamId, response.getSource().header.streamId), + streamId, response.getSource().header.streamId); } catch (IOException e) { @@ -215,9 +217,9 @@ private void validateMessageVersion(ProtocolVersion version) QueryMessage query = new QueryMessage("SELECT * FROM system.local", QueryOptions.DEFAULT) { @Override - public Envelope encode(ProtocolVersion originalVersion) + public Envelope encode(ProtocolVersion originalVersion, int streamId) { - return super.encode(v); + return super.encode(v, streamId); } }; try From 79af4a6e033c3bd05c367cc86be6ef8151f3425a Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Fri, 24 Jul 2026 16:32:56 -0500 Subject: [PATCH 2/3] Cleanup after manual review remove decorateResponse method restore comments that are no longer applicable restore some line break changes --- .../transport/CQLMessageHandler.java | 13 ++++++------ .../cassandra/transport/Dispatcher.java | 20 +++++++------------ .../apache/cassandra/transport/Envelope.java | 4 ---- .../transport/InitialConnectionHandler.java | 4 +--- 4 files changed, 15 insertions(+), 26 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java index eec2d4d65f1e..4441663cfe5d 100644 --- a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java +++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java @@ -465,6 +465,13 @@ private void handleError(Throwable t, int streamId) errorHandler.accept(ErrorMessage.wrap(t, streamId)); } + /** + * For use in the case where the error can't be mapped to a specific stream id, + * such as a corrupted frame, or when extracting a CQL message from the frame's + * payload fails. This does not attempt to release any resources, as these errors + * should only occur before any capacity acquisition is attempted (e.g. on receipt + * of a corrupt frame, or failure to extract a CQL message from the envelope). + */ private void handleError(Throwable t) { errorHandler.accept(t); @@ -726,12 +733,6 @@ protected void processCorruptFrame(FrameDecoder.CorruptFrame frame) processSubsequentFrameOfLargeMessage(frame); } - // A corrupt frame's bytes can't be reliably mapped back to a specific in-flight request (the - // frame may span several streams, and the stream id bytes may themselves be part of the - // corruption). We fall back to this connection's STARTUP stream id purely so the client still - // receives a diagnostic error frame before we tear the connection down (see the class javadoc - // above and CQLConnectionTest#handleFrameCorruptionAfterNegotiation). The exception is fatal, so - // the channel is closed as soon as the error has been written. handleError(ProtocolException.toFatalException(new ProtocolException(error))); } diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 5c7879074882..2e1b36a4f988 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -444,12 +444,10 @@ private static Message.Response processRequest(ServerConnection connection, Mess */ static Message.Response processRequest(Channel channel, Message.Request request, Overload backpressure, RequestTime requestTime) { + Message.Response response = null; try { - // decorateResponse must run before the finally below resets ClientWarn, so that it can read the - // warnings accumulated during processing. Decorating in the caller (after this method returns) - // would observe the reset state and drop all client warnings. - return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), request); + response = processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); } catch (Throwable t) { @@ -462,14 +460,17 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - return decorateResponse(ErrorMessage.fromExceptionNoStreamId(t, handler), request); + response = ErrorMessage.fromExceptionNoStreamId(t, handler); } finally { + if (response != null) + response.setWarnings(ClientWarn.instance.getWarnings()); CoordinatorWarnings.reset(); CoordinatorWriteWarnings.reset(); ClientWarn.instance.resetWarnings(); } + return response; } /** @@ -483,13 +484,6 @@

void processRequest(Channel channel, Message.Request request, FlushItemConve flush(toFlush); } - private static Message.Response decorateResponse(Message.Response response, Message.Request request) - { - assert response != null; - response.setWarnings(ClientWarn.instance.getWarnings()); - return response; - } - private void flush(FlushItem item) { EventLoop loop = item.channel.eventLoop(); @@ -522,7 +516,7 @@ public static void shutdown() * for delivering events to registered clients is dependent on protocol version and the configuration * of the pipeline. For v5 and newer connections, the event message is encoded into an Envelope, * wrapped in a FlushItem and then delivered via the pipeline's flusher, in a similar way to - * a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, P, Overload, RequestTime)}. + * a Response returned from {@link #processRequest(Channel, Message.Request, FlushItemConverter, Object, Overload, RequestTime)}. * It's worth noting that events are not generally fired as a direct response to a client request, * so this flush item has a null request attribute. The dispatcher itself is created when the * pipeline is first configured during protocol negotiation and is attached to the channel for diff --git a/src/java/org/apache/cassandra/transport/Envelope.java b/src/java/org/apache/cassandra/transport/Envelope.java index b33b054819ab..83ecd52d33f2 100644 --- a/src/java/org/apache/cassandra/transport/Envelope.java +++ b/src/java/org/apache/cassandra/transport/Envelope.java @@ -277,10 +277,6 @@ HeaderExtractionResult extractHeader(ByteBuffer buffer) // cause the channel to be closed. return new HeaderExtractionResult.Error(e, streamId, bodyLength); } - catch (ErrorMessage.WrappedException e) - { - return new HeaderExtractionResult.Error((ProtocolException) e.getCause(), e.getStreamId(), bodyLength); - } } public static abstract class HeaderExtractionResult diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index e8ddc9e519a8..28299363fb7e 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -122,9 +122,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li if (future.isSuccess()) { logger.trace("Response to STARTUP sent, configuring pipeline for {}", inbound.header.version); - configurator.configureModernPipeline(ctx, serverConnection, allocator, - inbound.header.version, - startup.options); + configurator.configureModernPipeline(ctx, serverConnection, allocator, inbound.header.version, startup.options); allocator.release(inbound.header.bodySizeInBytes); } else From 046f9590786f02cd8102630b920283514f3b5488 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Fri, 24 Jul 2026 20:51:06 -0500 Subject: [PATCH 3/3] Fix --- src/java/org/apache/cassandra/transport/Envelope.java | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/java/org/apache/cassandra/transport/Envelope.java b/src/java/org/apache/cassandra/transport/Envelope.java index 83ecd52d33f2..b33b054819ab 100644 --- a/src/java/org/apache/cassandra/transport/Envelope.java +++ b/src/java/org/apache/cassandra/transport/Envelope.java @@ -277,6 +277,10 @@ HeaderExtractionResult extractHeader(ByteBuffer buffer) // cause the channel to be closed. return new HeaderExtractionResult.Error(e, streamId, bodyLength); } + catch (ErrorMessage.WrappedException e) + { + return new HeaderExtractionResult.Error((ProtocolException) e.getCause(), e.getStreamId(), bodyLength); + } } public static abstract class HeaderExtractionResult