Ensure a Message's Response streamId is always set by frankgh · Pull Request #4936 · apache/cassandra

frankgh · 2026-07-14T02:00:39Z

patch by Francisco Guerrero; reviewed by TBD for CASSANDRA-21508

belliottsmith · 2026-07-14T07:17:05Z

-                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);


I would probably not use a valid streamId as a placeholder you intend to overwrite later

Yeah, that's a very good point. I will update it

One thing to consider here is that we won't always overwrite the stream id. I considered using the sentinel value, but this would trip the assertion during the encoding. So we'll need to keep a valid value here.

The cases where the streamId will not be overwritten are:

ExceptionHandlers.java:81 / PreV5Handlers.java:342 : This occurs during channel negotiations (SSL/TLS handshake failures) and a stream ID is not available

Unexpected non-wrapped pipeline throwables where no frame/stream ID exists at all

Everything else arrives in a wrapped exception and carries a real stream ID :

CQLMessageHandler:517 : extracted.streamId() (best-effort)

CQLMessageHandler:534 (oversized auth) : header.streamId

CQLMessageHandler:725 (corrupt frame) : NO_REQUEST_STREAM_ID — explicit, the one true context-free CQL path

I think keeping 0 for the NO_REQUEST_STREAM_ID is reasonable because the only sources of missing stream ids are when an error occurs before a stream id is available or when a a corrupt frame does not allow us to extract the stream id.

belliottsmith · 2026-07-15T21:39:50Z

+    private static Message.Response decorateResponse(Message.Response response, Message.Request request)
+    {
+        assert response != null;
+        response.setStreamId(request.getStreamId());


If we're setting it here, I'm still unsure why we're setting it at all callers to ErrorMessage as well...

This method is not the only sink for responses, and that's why we still need to set it in the callers to ErrorMessage . For example, the callsites for ExceptionHandlers:81, PreV5Handlers:342, InitialConnectionHandler:134 / 163, and Dispatcher:115 do not flow through processRequest in the Dispatcher, and the stream ID would not be set. I'd prefer to have the redundancy here since it is a harmless operation.

The first two have no streamId. I suspect that InitialConnectionHandler must have a zero streamId as well. I am not sure it is even correct to set it there - it isn't set currently, have you checked the change is valid? That is, it's possible the streamId relates to work to be done after the connection is established, and the error message is meant to be at the connection level? I don't know, but since it is a change in behaviour we should check. Either way, we close the connection immediately in the first case, so we are not at risk of this bug, and we probably should close the connection in the second case as well since it is an unexpected message on startup so the connection is in a corrupted state. I would prefer to simply do this for these two cases anyway, since it's safer without extra validation (no possible change to behaviour, and we won't be at risk of the bug).

That leaves only the Dispatcher case, and you're right that we should address this in a uniform manner. However, this suggests to me that calls to FlushItemConverter should be forced to go through a method that performs this work for us, since this is the final step before writing the reply to the wire, so captures all normal replies. Ideally, we would do something similar for the out-of-band flushes in InitialConnectionHandler and ExceptionHandlers, so that there are at most two routes to send a message. This might be annoying to have the compiler enforce, but I think it would be OK to settle for a uniform invoker of toFlushItem, especially if we are diligent about only setting streamId when we know it, so that we will fail if we hit UNSET_STREAM_ID.

Regarding redundancy: I am not fond of redundancy for its own sake - our goal should always be to make sure the code is well structured so that we must use it correctly, and so we can have confidence that all edge cases are correctly handled (and future edge cases should not be easily added). Redundancy of the kind discussed here implies we aren't confident we can do that, which suggests we need to improve our abstractions instead.

This particular redundancy also has downsides, as it has lead to the use of NO_REQUEST_STREAM_ID in one place where it seems to weaken the protection we are introducing with UNSET_STREAM_ID, because we're enforcing that it be supplied in at least one place where it doesn't seem to make semantic sense.

That said, I won't push hard on this.

maedhroz · 2026-07-15T21:41:28Z

+    {
+        assert response != null;
+        response.setStreamId(request.getStreamId());
+        response.setWarnings(ClientWarn.instance.getWarnings());


I think this is too late to set warnings. The finally that resets warnings happens before this. The following test passes again if we move the warning setting back to the pre-patch location:

@Test public void warningsAreDeliveredToClient() throws Throwable { try (Cluster cluster = init(Cluster.build().withNodes(1) .withConfig(config -> config.with(GOSSIP, NETWORK, NATIVE_PROTOCOL) .set("tombstone_warn_threshold", 0)).start())) { cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int PRIMARY KEY, v int)"); // insert a row so a tombstone is generated cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, v) VALUES (1, 1)", ConsistencyLevel.ONE); cluster.coordinator(1).execute("DELETE v FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.ONE); String address = cluster.get(1).config().broadcastAddress().getAddress().getHostAddress(); int port = cluster.get(1).callOnInstance(() -> getNativeTransportPort()); try (SimpleClient client = SimpleClient.builder(address, port) .protocolVersion(ProtocolVersion.V5) .build().connect(false)) { QueryMessage query = new QueryMessage("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", QueryOptions.DEFAULT); Message.Response response = client.execute(query); Assert.assertNotNull("Expected warnings on response, got none", response.getWarnings()); Assert.assertFalse("Expected non-empty warnings list", response.getWarnings().isEmpty()); } } }

Actually, I think ClientWarningsTest already catches this.

this is now fixed after the last commit

maedhroz

LGTM, once we fix the bits around warning propagation (see https://github.com/apache/cassandra/pull/4936/changes#r3590923562) and CI is otherwise clean

belliottsmith · 2026-07-16T10:24:58Z

-                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 fallback, otherwise the channel is torn down for fatal errors.
+                ErrorMessage errorMessage = ErrorMessage.fromException(cause, Message.NO_REQUEST_STREAM_ID, handler);


If we expect this to be overridden, why are we specifying NO_REQUEST_STREAM_ID, rather than UNSET_STREAM_ID?

There's one case where the negotiation fails during the protocol negotiation (app layer negotiation). This causes the org.apache.cassandra.transport.PreV5Handlers.ExceptionHandler#exceptionCaught code path above. And errorMessage.encode will hit the assertion in org.apache.cassandra.transport.Message#encode. If we set the stream Id to UNSET_STREAM_ID here, this would trigger the assertion error. So we'll need to do special handling for that case. This can be showcased in the org.apache.cassandra.transport.CQLConnectionTest#handleErrorDuringNegotiation test.

Ok, I spent some time looking at this and I think we can always get a stream id except for one case. I will push a commit with these changes in a bit.

belliottsmith · 2026-07-16T10:42:08Z

            if (responseVersion.isBeta())
                flags = Flag.add(flags, Flag.USE_BETA);

+            assert getStreamId() != UNSET_STREAM_ID : "Response streamId was never set: " + this;


This should not be an assertion, it should be a normal check and throw.

Ensure a Message's Response streamId is always set

9b4b177

patch by Francisco Guerrero; reviewed by TBD for CASSANDRA-21508

frankgh force-pushed the CASSANDRA-21508 branch from c85bea2 to 9b4b177 Compare July 14, 2026 02:32

belliottsmith reviewed Jul 14, 2026

View reviewed changes

frankgh added 5 commits July 14, 2026 10:40

Address PR comments from Benedict

7ea6225

Fix response decoration misses call site from InitialConnectionHandler

7614b48

Address comments from Benedict

79e0bf2

Use inbound.header.streamId

8cb7d15

Cleanup and add javadoc

ceb5c4e

belliottsmith reviewed Jul 15, 2026

View reviewed changes

maedhroz reviewed Jul 15, 2026

View reviewed changes

maedhroz approved these changes Jul 15, 2026

View reviewed changes

Address issue reported by Caleb

3adeed8

belliottsmith reviewed Jul 16, 2026

View reviewed changes

Uh oh!

Conversation

frankgh commented Jul 14, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

frankgh Jul 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

maedhroz left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

frankgh Jul 16, 2026 •

edited

Loading