From 9b4b177a7584ef7f47ecb32f424086b89bcd56fd Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Mon, 13 Jul 2026 15:05:17 -0500 Subject: [PATCH 1/7] Ensure a Message's Response streamId is always set patch by Francisco Guerrero; reviewed by TBD for CASSANDRA-21508 --- .../cassandra/transport/Dispatcher.java | 24 +- .../transport/ExceptionHandlers.java | 4 +- .../transport/InitialConnectionHandler.java | 5 +- .../apache/cassandra/transport/Message.java | 3 + .../cassandra/transport/PreV5Handlers.java | 4 +- .../transport/messages/AuthResponse.java | 2 +- .../transport/messages/AuthUtil.java | 7 +- .../transport/messages/BatchMessage.java | 2 +- .../transport/messages/ErrorMessage.java | 11 +- .../transport/messages/ExecuteMessage.java | 2 +- .../transport/messages/PrepareMessage.java | 2 +- .../transport/messages/QueryMessage.java | 2 +- .../transport/messages/StartupMessage.java | 2 +- .../test/StreamIdMisrouteTest.java | 253 ++++++++++++++++++ .../transport/CQLConnectionTest.java | 1 + .../cassandra/transport/ErrorMessageTest.java | 12 +- .../transport/ProtocolErrorTest.java | 2 +- 17 files changed, 304 insertions(+), 34 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 3f6468c21825..a2d85abdffef 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -112,8 +112,7 @@ public void dispatch(Channel channel, Message.Request request, FlushItemConverte // 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.fromException(new OverloadedException("Server is shutting down"), request.getStreamId()); response.setWarnings(ClientWarn.instance.getWarnings()); response.attach(request.connection); FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); @@ -374,7 +373,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.fromException(new OverloadedException("Query timed out before it could start"), request.getStreamId()); } if (connection.getVersion().isGreaterOrEqualTo(ProtocolVersion.V4)) @@ -434,8 +433,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 +445,8 @@ static Message.Response processRequest(Channel channel, Message.Request request, { try { - return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); + return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), + request); } catch (Throwable t) { @@ -461,10 +459,8 @@ 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; + ErrorMessage error = ErrorMessage.fromException(t, request.getStreamId(), handler); + return decorateResponse(error, request); } finally { @@ -474,6 +470,14 @@ static Message.Response processRequest(Channel channel, Message.Request request, } } + private static Message.Response decorateResponse(Message.Response response, Message.Request request) + { + assert response != null; + response.setStreamId(request.getStreamId()); + response.setWarnings(ClientWarn.instance.getWarnings()); + return response; + } + /** * Note: this method is not expected to execute on the netty event loop. */ diff --git a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java index 12bfe4190de6..a72216fed731 100644 --- a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java +++ b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java @@ -76,7 +76,9 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) if (ctx.channel().isOpen()) { Predicate handler = getUnexpectedExceptionHandler(ctx.channel(), false); - ErrorMessage errorMessage = ErrorMessage.fromException(cause, handler); + // No request in scope at the channel level; a WrappedException cause carries the frame's stream id + // and overrides this 0 fallback, otherwise the channel is torn down for fatal errors. + ErrorMessage errorMessage = ErrorMessage.fromException(cause, 0, handler); Envelope response = errorMessage.encode(version); FrameEncoder.Payload payload = allocator.allocate(true, CQLMessageHandler.envelopeSize(response.header)); try diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index 556a5607934d..d52378e3e2d1 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -131,7 +131,7 @@ 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); + ErrorMessage error = ErrorMessage.fromException(cause, inbound.header.streamId); Envelope response = error.encode(inbound.header.version); ChannelPromise closeChannel = AsyncChannelPromise.withListener(ctx, f -> ctx.close()); ctx.writeAndFlush(response, closeChannel); @@ -161,7 +161,8 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li ErrorMessage error = ErrorMessage.fromException( new ProtocolException(String.format("Unexpected message %s, expecting STARTUP or OPTIONS", - inbound.header.type))); + inbound.header.type)), + inbound.header.streamId); outbound = error.encode(inbound.header.version); ctx.writeAndFlush(outbound); } diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index d55e011b3a12..98a33c87a271 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -64,6 +64,7 @@ public abstract class Message { protected static final Logger logger = LoggerFactory.getLogger(Message.class); + public static final int UNSET_STREAM_ID = Integer.MIN_VALUE; public interface Codec extends CBCodec {} @@ -314,6 +315,7 @@ public static abstract class Response extends Message protected Response(Type type) { super(type); + setStreamId(UNSET_STREAM_ID); // must be overwritten before encode if (type.direction != Direction.RESPONSE) throw new IllegalArgumentException(); @@ -430,6 +432,7 @@ public Envelope encode(ProtocolVersion version) if (responseVersion.isBeta()) flags = Flag.add(flags, Flag.USE_BETA); + assert getStreamId() != UNSET_STREAM_ID : "Response streamId was never set: " + this; return Envelope.create(type, getStreamId(), responseVersion, flags, body); } catch (Throwable e) diff --git a/src/java/org/apache/cassandra/transport/PreV5Handlers.java b/src/java/org/apache/cassandra/transport/PreV5Handlers.java index f5b9d82f0223..3ea18e0aec0f 100644 --- a/src/java/org/apache/cassandra/transport/PreV5Handlers.java +++ b/src/java/org/apache/cassandra/transport/PreV5Handlers.java @@ -337,7 +337,9 @@ 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); + // No request in scope at the channel level; a WrappedException cause carries the frame's stream id + // and overrides this 0 fallback, otherwise the channel is torn down for fatal errors. + ErrorMessage errorMessage = ErrorMessage.fromException(cause, 0, 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. diff --git a/src/java/org/apache/cassandra/transport/messages/AuthResponse.java b/src/java/org/apache/cassandra/transport/messages/AuthResponse.java index 8be73bb6be9d..102a12598fda 100644 --- a/src/java/org/apache/cassandra/transport/messages/AuthResponse.java +++ b/src/java/org/apache/cassandra/transport/messages/AuthResponse.java @@ -81,6 +81,6 @@ protected Response execute(QueryState queryState, Dispatcher.RequestTime request { return new AuthChallenge(challenge); } - }); + }, getStreamId()); } } diff --git a/src/java/org/apache/cassandra/transport/messages/AuthUtil.java b/src/java/org/apache/cassandra/transport/messages/AuthUtil.java index 1c091c7278a9..6c98c8cb7f16 100644 --- a/src/java/org/apache/cassandra/transport/messages/AuthUtil.java +++ b/src/java/org/apache/cassandra/transport/messages/AuthUtil.java @@ -53,10 +53,13 @@ public final class AuthUtil * @param messageToSendBasedOnNegotiation Determines what response to return on based on whether sasl negotiation * is complete (1st parameter) and the challenege token returned from the * negotiator (2nd parameter). + * @param streamId The stream id of the request being handled, used to stamp any error + * response so it is routed back to the correct in-flight request. * @return the response to send back to the client. */ static Response handleLogin(Connection connection, QueryState queryState, byte[] token, - BiFunction messageToSendBasedOnNegotiation) + BiFunction messageToSendBasedOnNegotiation, + int streamId) { IAuthenticator.SaslNegotiator negotiator = ((ServerConnection) connection).getSaslNegotiator(queryState); try @@ -87,7 +90,7 @@ static Response handleLogin(Connection connection, QueryState queryState, byte[] { ClientMetrics.instance.markAuthFailure(negotiator.getAuthenticationMode()); AuthEvents.instance.notifyAuthFailure(queryState, e); - return ErrorMessage.fromException(e); + return ErrorMessage.fromException(e, streamId); } } diff --git a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java index d40b6d01cbcd..fd7ca6125805 100644 --- a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java @@ -236,7 +236,7 @@ protected Message.Response execute(QueryState state, Dispatcher.RequestTime requ { QueryEvents.instance.notifyBatchFailure(prepared, batchType, queryOrIdList, values, options, state, e); JVMStabilityInspector.inspectThrowable(e); - return ErrorMessage.fromException(e); + return ErrorMessage.fromException(e, getStreamId()); } } diff --git a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java index 36b7f34eb38b..f95ff0b96210 100644 --- a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java @@ -429,20 +429,21 @@ private ErrorMessage(TransportException error, int streamId) setStreamId(streamId); } - public static ErrorMessage fromException(Throwable e) + public static ErrorMessage fromException(Throwable e, int streamId) { - return fromException(e, null); + return fromException(e, streamId, null); } /** * @param e the exception + * @param streamId the stream id of the request this error responds to, so the error frame is routed back + * to the correct in-flight request. A {@link WrappedException} cause overrides this with the + * stream id recovered from the frame header. * @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 ErrorMessage fromException(Throwable e, int streamId, Predicate unexpectedExceptionHandler) { - int streamId = 0; - // 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. if (e instanceof CodecException) diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index 0f6afe0fe123..490c1ef6cbb1 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.fromException(e, getStreamId()); } } diff --git a/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java b/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java index e92b69b16e00..527f8a3804dd 100644 --- a/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/PrepareMessage.java @@ -134,7 +134,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.fromException(e, getStreamId()); } } diff --git a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java index 61f3a9cfbd83..2e0e6f0bd987 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.fromException(e, getStreamId()); } } diff --git a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java index c76db6826e30..12df8cc2de59 100644 --- a/src/java/org/apache/cassandra/transport/messages/StartupMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/StartupMessage.java @@ -158,7 +158,7 @@ else if (compression.equals("lz4")) // authentication, in this case we can just go through the traditional auth flow. return authenticator.getAuthenticateMessage(clientState); } - }); + }, getStreamId()); } } return authenticator.getAuthenticateMessage(clientState); 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..aa7957a289c0 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/StreamIdMisrouteTest.java @@ -0,0 +1,253 @@ +/* + * 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.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.getStreamId()); + + // 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.setStreamId(streamId); + 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/unit/org/apache/cassandra/transport/CQLConnectionTest.java b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java index 6780e4206e7d..79510cc2b39b 100644 --- a/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java +++ b/test/unit/org/apache/cassandra/transport/CQLConnectionTest.java @@ -658,6 +658,7 @@ static class TestConsumer implements MessageConsumer TestConsumer(Message.Response fixedResponse, FrameEncoder frameEncoder) { this.fixedResponse = fixedResponse; + this.fixedResponse.setStreamId(0); this.responseTemplate = fixedResponse.encode(ProtocolVersion.V5); this.frameEncoder = frameEncoder; } diff --git a/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java b/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java index c0e3f5ed91ef..dd9010a56283 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.fromException(rfe, 0), 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.fromException(wfe, 0), 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.fromException(ex, 0), 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.fromException(ex, 0), 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.fromException(ex, 0), 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.fromException(ex, 0), 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/ProtocolErrorTest.java b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java index a0c6694b6e2b..5382c0d179d1 100644 --- a/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java +++ b/test/unit/org/apache/cassandra/transport/ProtocolErrorTest.java @@ -161,7 +161,7 @@ public void testBodyLengthOverLimit() throws Exception public void testErrorMessageWithNullString() { // test for CASSANDRA-11167 - ErrorMessage msg = ErrorMessage.fromException(new ServerError((String) null)); + ErrorMessage msg = ErrorMessage.fromException(new ServerError((String) null), 0); assert msg.toString().endsWith("null") : msg.toString(); int size = ErrorMessage.codec.encodedSize(msg, ProtocolVersion.CURRENT); ByteBuf buf = Unpooled.buffer(size); From 7ea622597dd9c6b779d9cea868079c68cc5d58d6 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Tue, 14 Jul 2026 10:40:26 -0500 Subject: [PATCH 2/7] Address PR comments from Benedict --- .../transport/CQLMessageHandler.java | 23 ++++++------------ .../cassandra/transport/Dispatcher.java | 24 +++++++++---------- .../transport/ExceptionHandlers.java | 6 ++--- .../cassandra/transport/PreV5Handlers.java | 6 ++--- .../transport/messages/ErrorMessage.java | 9 +++++++ 5 files changed, 33 insertions(+), 35 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java index 2486497c2f37..c19ded9221dc 100644 --- a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java +++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java @@ -465,18 +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); - } - // Acts as a Dispatcher.FlushItemConverter private Framed toFlushItem(Channel channel, Message.Request request, Message.Response response) { @@ -524,8 +512,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 +531,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,7 +720,9 @@ protected void processCorruptFrame(FrameDecoder.CorruptFrame frame) processSubsequentFrameOfLargeMessage(frame); } - handleError(ProtocolException.toFatalException(new ProtocolException(error))); + // A corrupt frame's bytes can't be trusted to map back to a single stream (it may span + // several), so there is no usable stream id here; use the explicit no-request sentinel. + handleError(ProtocolException.toFatalException(new ProtocolException(error)), ErrorMessage.NO_REQUEST_STREAM_ID); } protected void fatalExceptionCaught(Throwable cause) diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index a2d85abdffef..5fafa94df00e 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -445,8 +445,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, { try { - return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), - request); + return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); } catch (Throwable t) { @@ -459,8 +458,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - ErrorMessage error = ErrorMessage.fromException(t, request.getStreamId(), handler); - return decorateResponse(error, request); + return ErrorMessage.fromException(t, request.getStreamId(), handler); } finally { @@ -470,25 +468,25 @@ static Message.Response processRequest(Channel channel, Message.Request request, } } - private static Message.Response decorateResponse(Message.Response response, Message.Request request) - { - assert response != null; - response.setStreamId(request.getStreamId()); - response.setWarnings(ClientWarn.instance.getWarnings()); - return response; - } - /** * 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) { Message.Response response = processRequest(channel, request, backpressure, requestTime); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); + FlushItem toFlush = forFlusher.toFlushItem(channel, request, decorateResponse(response, request)); 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.setStreamId(request.getStreamId()); + response.setWarnings(ClientWarn.instance.getWarnings()); + return response; + } + private void flush(FlushItem item) { EventLoop loop = item.channel.eventLoop(); diff --git a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java index a72216fed731..ad3b1bb43a87 100644 --- a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java +++ b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java @@ -76,9 +76,9 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) if (ctx.channel().isOpen()) { Predicate handler = getUnexpectedExceptionHandler(ctx.channel(), false); - // No request in scope at the channel level; a WrappedException cause carries the frame's stream id - // and overrides this 0 fallback, otherwise the channel is torn down for fatal errors. - ErrorMessage errorMessage = ErrorMessage.fromException(cause, 0, handler); + // No request in scope at the channel level; a WrappedException cause carries the frame's + // stream id and overrides this fallback, otherwise the channel is torn down for fatal errors. + ErrorMessage errorMessage = ErrorMessage.fromException(cause, ErrorMessage.NO_REQUEST_STREAM_ID, handler); Envelope response = errorMessage.encode(version); FrameEncoder.Payload payload = allocator.allocate(true, CQLMessageHandler.envelopeSize(response.header)); try diff --git a/src/java/org/apache/cassandra/transport/PreV5Handlers.java b/src/java/org/apache/cassandra/transport/PreV5Handlers.java index 3ea18e0aec0f..15bac24417e8 100644 --- a/src/java/org/apache/cassandra/transport/PreV5Handlers.java +++ b/src/java/org/apache/cassandra/transport/PreV5Handlers.java @@ -337,9 +337,9 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) if (ctx.channel().isOpen()) { Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(ctx.channel(), false); - // No request in scope at the channel level; a WrappedException cause carries the frame's stream id - // and overrides this 0 fallback, otherwise the channel is torn down for fatal errors. - ErrorMessage errorMessage = ErrorMessage.fromException(cause, 0, handler); + // No request in scope at the channel level; a WrappedException cause carries the frame's + // stream id and overrides this fallback, otherwise the channel is torn down for fatal errors. + ErrorMessage errorMessage = ErrorMessage.fromException(cause, ErrorMessage.NO_REQUEST_STREAM_ID, 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. diff --git a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java index f95ff0b96210..955a21fa5221 100644 --- a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java @@ -74,6 +74,15 @@ public class ErrorMessage extends Message.Response { private static final Logger logger = LoggerFactory.getLogger(ErrorMessage.class); + /** + * Stream id used for channel-level errors that have no associated request (e.g. a raw protocol, + * SSL, or other unexpected failure caught in a pipeline {@code exceptionCaught} handler, where no + * request frame - and therefore no stream id - is in scope). A {@link WrappedException} cause, when + * present, carries the originating frame's stream id and overrides this value in + * {@link #fromException(Throwable, int, com.google.common.base.Predicate)}. + */ + public static final int NO_REQUEST_STREAM_ID = 0; + public static final Message.Codec codec = new Message.Codec() { public ErrorMessage decode(ByteBuf body, ProtocolVersion version) From 7614b48411c523c6e8a05b2a8d3a6d850ceda2f0 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Tue, 14 Jul 2026 13:31:01 -0500 Subject: [PATCH 3/7] Fix response decoration misses call site from InitialConnectionHandler --- src/java/org/apache/cassandra/transport/Dispatcher.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 5fafa94df00e..2d81b61ccc89 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -445,7 +445,8 @@ static Message.Response processRequest(Channel channel, Message.Request request, { try { - return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); + return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), + request); } catch (Throwable t) { @@ -458,7 +459,8 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - return ErrorMessage.fromException(t, request.getStreamId(), handler); + ErrorMessage error = ErrorMessage.fromException(t, request.getStreamId(), handler); + return decorateResponse(error, request); } finally { @@ -474,7 +476,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure, RequestTime requestTime) { Message.Response response = processRequest(channel, request, backpressure, requestTime); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, decorateResponse(response, request)); + FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); Message.logger.trace("Responding: {}, v={}", response, request.connection().getVersion()); flush(toFlush); } From 79e0bf28cda15ae310f12121799ff512b2e5da18 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Wed, 15 Jul 2026 12:21:38 -0500 Subject: [PATCH 4/7] Address comments from Benedict --- .../apache/cassandra/transport/CQLMessageHandler.java | 2 +- src/java/org/apache/cassandra/transport/Dispatcher.java | 8 +++----- .../apache/cassandra/transport/ExceptionHandlers.java | 2 +- .../cassandra/transport/InitialConnectionHandler.java | 3 +++ src/java/org/apache/cassandra/transport/Message.java | 9 +++++++++ .../org/apache/cassandra/transport/PreV5Handlers.java | 2 +- .../cassandra/transport/messages/ErrorMessage.java | 9 --------- 7 files changed, 18 insertions(+), 17 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java index c19ded9221dc..3174fab9c504 100644 --- a/src/java/org/apache/cassandra/transport/CQLMessageHandler.java +++ b/src/java/org/apache/cassandra/transport/CQLMessageHandler.java @@ -722,7 +722,7 @@ protected void processCorruptFrame(FrameDecoder.CorruptFrame frame) // A corrupt frame's bytes can't be trusted to map back to a single stream (it may span // several), so there is no usable stream id here; use the explicit no-request sentinel. - handleError(ProtocolException.toFatalException(new ProtocolException(error)), ErrorMessage.NO_REQUEST_STREAM_ID); + handleError(ProtocolException.toFatalException(new ProtocolException(error)), Message.NO_REQUEST_STREAM_ID); } protected void fatalExceptionCaught(Throwable cause) diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 2d81b61ccc89..5fafa94df00e 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -445,8 +445,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, { try { - return decorateResponse(processRequest((ServerConnection) request.connection(), request, backpressure, requestTime), - request); + return processRequest((ServerConnection) request.connection(), request, backpressure, requestTime); } catch (Throwable t) { @@ -459,8 +458,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - ErrorMessage error = ErrorMessage.fromException(t, request.getStreamId(), handler); - return decorateResponse(error, request); + return ErrorMessage.fromException(t, request.getStreamId(), handler); } finally { @@ -476,7 +474,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure, RequestTime requestTime) { Message.Response response = processRequest(channel, request, backpressure, requestTime); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); + FlushItem toFlush = forFlusher.toFlushItem(channel, request, decorateResponse(response, request)); Message.logger.trace("Responding: {}, v={}", response, request.connection().getVersion()); flush(toFlush); } diff --git a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java index ad3b1bb43a87..d3d640084ccd 100644 --- a/src/java/org/apache/cassandra/transport/ExceptionHandlers.java +++ b/src/java/org/apache/cassandra/transport/ExceptionHandlers.java @@ -78,7 +78,7 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) Predicate handler = getUnexpectedExceptionHandler(ctx.channel(), false); // No request in scope at the channel level; a WrappedException cause carries the frame's // stream id and overrides this fallback, otherwise the channel is torn down for fatal errors. - ErrorMessage errorMessage = ErrorMessage.fromException(cause, ErrorMessage.NO_REQUEST_STREAM_ID, handler); + ErrorMessage errorMessage = ErrorMessage.fromException(cause, Message.NO_REQUEST_STREAM_ID, handler); Envelope response = errorMessage.encode(version); FrameEncoder.Payload payload = allocator.allocate(true, CQLMessageHandler.envelopeSize(response.header)); try diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index d52378e3e2d1..5c1b9ad5a1d5 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -42,6 +42,8 @@ import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.util.Attribute; +import static org.apache.cassandra.transport.Message.NO_REQUEST_STREAM_ID; + /** * Added to the Netty pipeline whenever a new Channel is initialized. This handler only processes * the messages which constitute the initial handshake between client and server, namely @@ -151,6 +153,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li } final Message.Response response = Dispatcher.processRequest(ctx.channel(), startup, Overload.NONE, Dispatcher.RequestTime.forImmediateExecution()); + response.setStreamId(NO_REQUEST_STREAM_ID); outbound = response.encode(inbound.header.version); ctx.writeAndFlush(outbound, promise); diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 98a33c87a271..5ac54a49032b 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -64,7 +64,16 @@ public abstract class Message { protected static final Logger logger = LoggerFactory.getLogger(Message.class); + public static final int UNSET_STREAM_ID = Integer.MIN_VALUE; + /** + * Stream id used for channel-level errors that have no associated request (e.g. a raw protocol, + * SSL, or other unexpected failure caught in a pipeline {@code exceptionCaught} handler, where no + * request frame - and therefore no stream id - is in scope). A {@link ErrorMessage.WrappedException} cause, when + * present, carries the originating frame's stream id and overrides this value in + * {@link ErrorMessage#fromException(Throwable, int, com.google.common.base.Predicate)}. + */ + public static final int NO_REQUEST_STREAM_ID = 0; public interface Codec extends CBCodec {} diff --git a/src/java/org/apache/cassandra/transport/PreV5Handlers.java b/src/java/org/apache/cassandra/transport/PreV5Handlers.java index 15bac24417e8..c1e2dc29187a 100644 --- a/src/java/org/apache/cassandra/transport/PreV5Handlers.java +++ b/src/java/org/apache/cassandra/transport/PreV5Handlers.java @@ -339,7 +339,7 @@ public void exceptionCaught(final ChannelHandlerContext ctx, Throwable cause) Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(ctx.channel(), false); // No request in scope at the channel level; a WrappedException cause carries the frame's // stream id and overrides this fallback, otherwise the channel is torn down for fatal errors. - ErrorMessage errorMessage = ErrorMessage.fromException(cause, ErrorMessage.NO_REQUEST_STREAM_ID, handler); + ErrorMessage errorMessage = ErrorMessage.fromException(cause, Message.NO_REQUEST_STREAM_ID, 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. diff --git a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java index 955a21fa5221..f95ff0b96210 100644 --- a/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ErrorMessage.java @@ -74,15 +74,6 @@ public class ErrorMessage extends Message.Response { private static final Logger logger = LoggerFactory.getLogger(ErrorMessage.class); - /** - * Stream id used for channel-level errors that have no associated request (e.g. a raw protocol, - * SSL, or other unexpected failure caught in a pipeline {@code exceptionCaught} handler, where no - * request frame - and therefore no stream id - is in scope). A {@link WrappedException} cause, when - * present, carries the originating frame's stream id and overrides this value in - * {@link #fromException(Throwable, int, com.google.common.base.Predicate)}. - */ - public static final int NO_REQUEST_STREAM_ID = 0; - public static final Message.Codec codec = new Message.Codec() { public ErrorMessage decode(ByteBuf body, ProtocolVersion version) From 8cb7d15f860b17e07ee9b3e5b94adf37a71e691a Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Wed, 15 Jul 2026 13:51:34 -0500 Subject: [PATCH 5/7] Use inbound.header.streamId --- .../apache/cassandra/transport/InitialConnectionHandler.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index 5c1b9ad5a1d5..690764a9a6bd 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -153,7 +153,7 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li } final Message.Response response = Dispatcher.processRequest(ctx.channel(), startup, Overload.NONE, Dispatcher.RequestTime.forImmediateExecution()); - response.setStreamId(NO_REQUEST_STREAM_ID); + response.setStreamId(inbound.header.streamId); outbound = response.encode(inbound.header.version); ctx.writeAndFlush(outbound, promise); From ceb5c4ecbaf40a733572d75dcb78e6d2b851621c Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Wed, 15 Jul 2026 13:57:00 -0500 Subject: [PATCH 6/7] Cleanup and add javadoc --- .../apache/cassandra/transport/InitialConnectionHandler.java | 2 -- src/java/org/apache/cassandra/transport/Message.java | 1 + 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index 690764a9a6bd..959a3e79d24c 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -42,8 +42,6 @@ import io.netty.handler.codec.ByteToMessageDecoder; import io.netty.util.Attribute; -import static org.apache.cassandra.transport.Message.NO_REQUEST_STREAM_ID; - /** * Added to the Netty pipeline whenever a new Channel is initialized. This handler only processes * the messages which constitute the initial handshake between client and server, namely diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 5ac54a49032b..0900ee3d28a1 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -65,6 +65,7 @@ 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} (asserted there). */ public static final int UNSET_STREAM_ID = Integer.MIN_VALUE; /** * Stream id used for channel-level errors that have no associated request (e.g. a raw protocol, From 3adeed85e7f6f437e9ff62a3f322f112d78c26d1 Mon Sep 17 00:00:00 2001 From: Francisco Guerrero Date: Wed, 15 Jul 2026 17:12:27 -0500 Subject: [PATCH 7/7] Address issue reported by Caleb --- src/java/org/apache/cassandra/transport/Dispatcher.java | 9 ++++++--- .../cassandra/transport/InitialConnectionHandler.java | 1 - 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/Dispatcher.java b/src/java/org/apache/cassandra/transport/Dispatcher.java index 5fafa94df00e..b103b8249b8f 100644 --- a/src/java/org/apache/cassandra/transport/Dispatcher.java +++ b/src/java/org/apache/cassandra/transport/Dispatcher.java @@ -445,7 +445,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) { @@ -458,7 +461,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, } Predicate handler = ExceptionHandlers.getUnexpectedExceptionHandler(channel, true); - return ErrorMessage.fromException(t, request.getStreamId(), handler); + return decorateResponse(ErrorMessage.fromException(t, request.getStreamId(), handler), request); } finally { @@ -474,7 +477,7 @@ static Message.Response processRequest(Channel channel, Message.Request request, void processRequest(Channel channel, Message.Request request, FlushItemConverter forFlusher, Overload backpressure, RequestTime requestTime) { Message.Response response = processRequest(channel, request, backpressure, requestTime); - FlushItem toFlush = forFlusher.toFlushItem(channel, request, decorateResponse(response, request)); + FlushItem toFlush = forFlusher.toFlushItem(channel, request, response); Message.logger.trace("Responding: {}, v={}", response, request.connection().getVersion()); flush(toFlush); } diff --git a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java index 959a3e79d24c..d52378e3e2d1 100644 --- a/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java +++ b/src/java/org/apache/cassandra/transport/InitialConnectionHandler.java @@ -151,7 +151,6 @@ protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List li } final Message.Response response = Dispatcher.processRequest(ctx.channel(), startup, Overload.NONE, Dispatcher.RequestTime.forImmediateExecution()); - response.setStreamId(inbound.header.streamId); outbound = response.encode(inbound.header.version); ctx.writeAndFlush(outbound, promise);