From a3e7cccf6a42ff95d8f5bf61ea20e5e16b558c91 Mon Sep 17 00:00:00 2001 From: Maxim Muzafarov Date: Mon, 27 Jul 2026 18:06:57 +0200 Subject: [PATCH] fixup! handle stop daemon cql command properly --- .../cassandra/transport/SimpleClient.java | 5 +- .../cassandra/transport/ErrorMessageTest.java | 8 ++-- .../MessageManagementDispatcherTest.java | 47 ++++++++++--------- 3 files changed, 31 insertions(+), 29 deletions(-) diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index 3df7be720d..deb5141b7c 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -412,7 +412,6 @@ public class SimpleClient implements Closeable } } -<<<<<<< HEAD /** * 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). @@ -423,7 +422,8 @@ public class SimpleClient implements Closeable { Envelope source = message.getSource(); return source == null ? 0 : source.header.streamId; -======= + } + private long responseDeadlineNanos() { return requestTimeoutSeconds > 0 ? nanoTime() + TimeUnit.SECONDS.toNanos(requestTimeoutSeconds) : Long.MAX_VALUE; @@ -442,7 +442,6 @@ public class SimpleClient implements Closeable if (nanoTime() - deadlineNanos >= 0) return null; } ->>>>>>> f9202d6b13 (handle stop daemon cql command properly) } public interface EventHandler diff --git a/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java b/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java index 4d875df722..0a42d67de2 100644 --- a/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java +++ b/test/unit/org/apache/cassandra/transport/ErrorMessageTest.java @@ -202,7 +202,7 @@ public class ErrorMessageTest extends EncodeAndDecodeTestBase Throwable cause = new RuntimeException("Underlying cause"); CommandRequestExecutionException ex = new CommandRequestExecutionException(executionId, errorMessage, cause); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromExceptionNoStreamId(ex), ProtocolVersion.V5); assertTrue(deserialized.error instanceof CommandRequestExecutionException); CommandRequestExecutionException deserializedEx = (CommandRequestExecutionException) deserialized.error; @@ -219,7 +219,7 @@ public class ErrorMessageTest extends EncodeAndDecodeTestBase Throwable cause = new RuntimeException("Underlying cause"); CommandRequestExecutionException ex = new CommandRequestExecutionException(executionId, errorMessage, cause); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V4); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromExceptionNoStreamId(ex), ProtocolVersion.V4); assertTrue(deserialized.error instanceof ServerError); ServerError deserializedEx = (ServerError) deserialized.error; @@ -235,7 +235,7 @@ public class ErrorMessageTest extends EncodeAndDecodeTestBase String errorMessage = "Command execution failed: test command"; CommandRequestExecutionException ex = new CommandRequestExecutionException(executionId, errorMessage); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V5); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromExceptionNoStreamId(ex), ProtocolVersion.V5); assertTrue(deserialized.error instanceof CommandRequestExecutionException); CommandRequestExecutionException deserializedEx = (CommandRequestExecutionException) deserialized.error; @@ -251,7 +251,7 @@ public class ErrorMessageTest extends EncodeAndDecodeTestBase String errorMessage = "Command execution failed: test command"; CommandRequestExecutionException ex = new CommandRequestExecutionException(executionId, errorMessage); - ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromException(ex), ProtocolVersion.V3); + ErrorMessage deserialized = encodeThenDecode(ErrorMessage.fromExceptionNoStreamId(ex), ProtocolVersion.V3); assertTrue(deserialized.error instanceof ServerError); ServerError deserializedEx = (ServerError) deserialized.error; diff --git a/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java b/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java index 1afb700dca..5310f8caaa 100644 --- a/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java +++ b/test/unit/org/apache/cassandra/transport/MessageManagementDispatcherTest.java @@ -400,11 +400,12 @@ public class MessageManagementDispatcherTest Dispatcher managementDispatcher = new ManagementTestDispatcher(true) { @Override - void processRequest(Channel channel, - Message.Request request, - FlushItemConverter forFlusher, - ClientResourceLimits.Overload backpressure, - RequestTime requestTime) +

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

forFlusher, + P param, + ClientResourceLimits.Overload backpressure, + RequestTime requestTime) { isDoneInsideTask.set(isDone()); entered.countDown(); @@ -413,8 +414,8 @@ public class MessageManagementDispatcherTest }; Message.Request request = createManagementRequest(Message.Type.STARTUP); - managementDispatcher.dispatch(request.connection().channel(), request, (channel, req, response) -> null, - ClientResourceLimits.Overload.NONE); + managementDispatcher.dispatch(request.connection().channel(), request, (param, channel, req, response) -> null, + null, ClientResourceLimits.Overload.NONE); try { assertTrue(entered.await(10, TimeUnit.SECONDS)); @@ -464,18 +465,19 @@ public class MessageManagementDispatcherTest Dispatcher queuedProcessor = new ManagementTestDispatcher(true) { @Override - void processRequest(Channel channel, - Message.Request request, - FlushItemConverter forFlusher, - ClientResourceLimits.Overload backpressure, - RequestTime requestTime) +

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

forFlusher, + P param, + ClientResourceLimits.Overload backpressure, + RequestTime requestTime) { processed.countDown(); } }; Message.Request request = createManagementRequest(Message.Type.STARTUP); - queuedProcessor.dispatch(request.connection().channel(), request, (channel, req, response) -> null, - ClientResourceLimits.Overload.NONE); + queuedProcessor.dispatch(request.connection().channel(), request, (param, channel, req, response) -> null, + null, ClientResourceLimits.Overload.NONE); awaitTrue("A queued management request should trigger management backpressure", () -> !managementDispatcher.hasQueueCapacity()); @@ -518,8 +520,8 @@ public class MessageManagementDispatcherTest private long tryRequest(Callable check, Message.Request request) throws Exception { long start = check.call(); - dispatch.dispatch(request.connection().channel(), request, (channel, req, response) -> null, - ClientResourceLimits.Overload.NONE); + dispatch.dispatch(request.connection().channel(), request, (param, channel, req, response) -> null, + null, ClientResourceLimits.Overload.NONE); long timeout = Clock.Global.currentTimeMillis(); while (start == check.call() && Clock.Global.currentTimeMillis() - timeout < 1000) @@ -589,7 +591,7 @@ public class MessageManagementDispatcherTest return managementConnectionMock(); } }; - msg.setStreamId(1); + msg.setSource(new Envelope(Envelope.Header.dummy(1, msg.type), null)); return msg; } @@ -606,11 +608,12 @@ public class MessageManagementDispatcherTest } @Override - void processRequest(Channel channel, - Message.Request request, - FlushItemConverter forFlusher, - ClientResourceLimits.Overload backpressure, - RequestTime requestTime) +

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

forFlusher, + P param, + ClientResourceLimits.Overload backpressure, + RequestTime requestTime) { // noop - just for testing routing }