This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 9.0.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit c7c0cec50580932c137bf1f1a77afaef9ea39a1d Author: opencode <[email protected]> AuthorDate: Fri Oct 9 13:26:40 2026 +0200 Fix MessageDispatchInterceptor queue size accounting drift The async send path added the message length to currentSize when the message was queued, but subtracted msg.getMessage().getLength() after super.sendMessage() had run the message through the remaining interceptors. EncryptInterceptor replaces the payload with IV, ciphertext and tag, and FragmentationInterceptor appends a fragment flag byte, so every async message over-decremented the counter. The recorded size drifted steadily negative until the maxQueueSize guard could no longer fire, silently removing the only backpressure on the unbounded dispatch queue. Use one snapshot of the message length for both the increment and the decrement, apply the increment before the task is submitted (rolling it back if the executor rejects the task) so a racing worker can never mutate the value before it is recorded, and reset the counter when the queue starts so residual adjustments from a stopped run cannot carry over. Add a regression test using a downstream interceptor that extends the message payload. --- .../interceptors/MessageDispatchInterceptor.java | 26 ++++++++--- .../TestMessageDispatchInterceptor.java | 52 ++++++++++++++++++++++ webapps/docs/changelog.xml | 13 ++++++ 3 files changed, 86 insertions(+), 5 deletions(-) diff --git a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java index 129f7c6d88..fe3dc341e0 100644 --- a/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java +++ b/java/org/apache/catalina/tribes/group/interceptors/MessageDispatchInterceptor.java @@ -122,10 +122,15 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase implement // data, so the queued message is independent of that buffer. msg = ((ChannelData) msg).clone(); } - if (!addToQueue(msg, destination, payload)) { + // Record the queue size before the send task is submitted. Downstream + // interceptors may replace or extend the message payload while the message + // is being sent, so the accounting must use a single snapshot of the length. + long queuedLength = msg.getMessage().getLength(); + addAndGetCurrentSize(queuedLength); + if (!addToQueue(msg, destination, payload, queuedLength)) { + addAndGetCurrentSize(-queuedLength); throw new ChannelException(sm.getString("messageDispatchInterceptor.unableAdd.queue")); } - addAndGetCurrentSize(msg.getMessage().getLength()); } else { super.sendMessage(destination, msg, payload); } @@ -140,8 +145,14 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase implement * @return true if added */ public boolean addToQueue(final ChannelMessage msg, final Member[] destination, final InterceptorPayload payload) { + return addToQueue(msg, destination, payload, msg.getMessage().getLength()); + } + + + private boolean addToQueue(final ChannelMessage msg, final Member[] destination, + final InterceptorPayload payload, final long queuedLength) { try { - executor.execute(() -> sendAsyncData(msg, destination, payload)); + executor.execute(() -> sendAsyncData(msg, destination, payload, queuedLength)); return true; } catch (RejectedExecutionException ree) { return false; @@ -162,6 +173,9 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase implement } executor = ExecutorFactory.newThreadPool(maxSpareThreads, maxThreads, keepAliveTime, TimeUnit.MILLISECONDS, new TcclThreadFactory("MessageDispatchInterceptor.MessageDispatchThread" + channelName)); + // Tasks that were still in flight when a previous queue run was stopped may have + // adjusted the recorded size after stopQueue() reset it. Start from a clean state. + setAndGetCurrentSize(0); run = true; } @@ -332,8 +346,10 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase implement * @param msg the message * @param destination the destination * @param payload the payload + * @param queuedLength the message length as recorded when the message was enqueued */ - protected void sendAsyncData(ChannelMessage msg, Member[] destination, InterceptorPayload payload) { + protected void sendAsyncData(ChannelMessage msg, Member[] destination, InterceptorPayload payload, + long queuedLength) { ErrorHandler handler = null; if (payload != null) { handler = payload.getErrorHandler(); @@ -365,7 +381,7 @@ public class MessageDispatchInterceptor extends ChannelInterceptorBase implement log.error(sm.getString("messageDispatchInterceptor.errorMessage.failed"), ex); } } finally { - addAndGetCurrentSize(-msg.getMessage().getLength()); + addAndGetCurrentSize(-queuedLength); } } diff --git a/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java b/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java index b3b5efd2a4..dc823a34a2 100644 --- a/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java +++ b/test/org/apache/catalina/tribes/group/interceptors/TestMessageDispatchInterceptor.java @@ -34,6 +34,7 @@ import org.apache.catalina.tribes.group.InterceptorPayload; import org.apache.catalina.tribes.io.ChannelData; import org.apache.catalina.tribes.io.XByteBuffer; import org.apache.catalina.tribes.membership.MemberImpl; +import org.apache.catalina.tribes.util.UUIDGenerator; public class TestMessageDispatchInterceptor { @@ -128,6 +129,57 @@ public class TestMessageDispatchInterceptor { Assert.assertSame("Synchronous sends must pass the original message through", data, collector.message); } + @Test + public void testQueueSizeReturnsToZeroWhenDownstreamMutatesPayload() throws Exception { + GroupChannel channel = new GroupChannel(); + MessageDispatchInterceptor interceptor = new MessageDispatchInterceptor(); + MutatingCollector collector = new MutatingCollector(); + interceptor.setChannel(channel); + interceptor.setNext(collector); + interceptor.startQueue(); + try { + byte[] payloadBytes = "PAYLOAD".getBytes(StandardCharsets.UTF_8); + Member destination = new MemberImpl("localhost", 4000, -1); + Member[] dest = new Member[] { destination }; + int count = 20; + for (int i = 0; i < count; i++) { + ChannelData data = new ChannelData(false); + data.setUniqueId(UUIDGenerator.randomUUID(false)); + data.setAddress(destination); + data.setOptions(Channel.SEND_OPTIONS_ASYNCHRONOUS); + XByteBuffer buffer = new XByteBuffer(64, false); + buffer.append(payloadBytes, 0, payloadBytes.length); + data.setMessage(buffer); + interceptor.sendMessage(dest, data, null); + } + + long deadline = System.currentTimeMillis() + 10000; + long size = interceptor.getCurrentSize(); + while (size != 0 && System.currentTimeMillis() < deadline) { + Thread.sleep(50); + size = interceptor.getCurrentSize(); + } + Assert.assertEquals( + "Queue size accounting drifted when a downstream interceptor changed the message length", + 0, size); + } finally { + interceptor.stopQueue(); + } + } + + /* + * Simulates interceptors such as EncryptInterceptor or FragmentationInterceptor that replace or extend the + * message payload while the message is being sent downstream. + */ + private static class MutatingCollector extends ChannelInterceptorBase { + + @Override + public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) + throws ChannelException { + msg.getMessage().append(true); + } + } + private static class RecordingCollector extends ChannelInterceptorBase { private volatile ChannelMessage message; diff --git a/webapps/docs/changelog.xml b/webapps/docs/changelog.xml index b556af9e29..7d42710cfc 100644 --- a/webapps/docs/changelog.xml +++ b/webapps/docs/changelog.xml @@ -2460,6 +2460,19 @@ <fix> Avoid NPE in manager webapp displaying certificate information. (remm) </fix> + <fix> + Fix the queue size accounting of the + <code>MessageDispatchInterceptor</code>. The message length + recorded when a message is queued is now also the length used + to release the accounting when the send completes. Previously, + when a downstream interceptor like + <code>EncryptInterceptor</code> or + <code>FragmentationInterceptor</code> replaced or extended the + message payload while the message was sent, the recorded queue + size drifted towards negative values and the + <code>maxQueueSize</code> limit eventually stopped being + enforced. (remm) + </fix> </changelog> </subsection> <subsection name="Other"> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
