This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 44955b4920f1d27577182b1632671748ee5ec640 Author: Aurélien Pupier <[email protected]> AuthorDate: Fri Oct 9 12:04:12 2026 +0200 CAMEL-25348: fix race condition in ACK sending by counting all completed messages Previously, the ACK for a window was sent only when the callback for the message whose sequence number equalled the window size completed: if (sequenceNumber == window.getSize()) { ctx.writeAndFlush(new LumberjackAck(...)); } All messages in a window are dispatched concurrently to Camel's async processor via a shared DefaultEventExecutorGroup. Their callbacks can therefore fire in any order — there is no guarantee that the message with the highest sequence number completes last. Under concurrent load (e.g. LumberjackMultiThreadIT with multiple connections sharing the same executor pool) the last-sequenced message could complete before earlier ones, or its processing could be delayed long enough that the client timed out and closed the connection before the ACK was ever written. Fix by replacing the sequence-number gate with an AtomicInteger counter initialised to the window size. Each successful callback decrements it; the ACK is written only when the counter reaches zero, i.e. once every message in the window has been processed regardless of completion order. The premature ctx.flush() in the finally block (which fired before any callback could complete) is also removed since writeAndFlush handles its own flush. Co-authored-by: IBM Bob 2.2.1 Signed-off-by: Aurélien Pupier <[email protected]> --- .../lumberjack/io/LumberjackMessageHandler.java | 42 ++++++++-------------- 1 file changed, 15 insertions(+), 27 deletions(-) diff --git a/components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java b/components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java index 767803e582ee..ff0f9aa6595a 100644 --- a/components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java +++ b/components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java @@ -17,6 +17,7 @@ package org.apache.camel.component.lumberjack.io; import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; @@ -49,35 +50,22 @@ final class LumberjackMessageHandler extends SimpleChannelInboundHandler<Lumberj @Override protected void channelRead0(ChannelHandlerContext ctx, LumberjackWindow window) throws Exception { - try { - for (LumberjackMessage msg : window) { - if (process) { - messageProcessor.onMessageReceived(msg.getPayload(), success -> { - if (success) { - notifyMessageProcessed(ctx, msg.getSequenceNumber(), window); - } else { - ctx.close(); - // Mark that we shouldn't process the next messages that are already decoded and are waiting in netty queues - process = false; + // Count down from window size; when it reaches zero all messages have been processed successfully. + AtomicInteger remaining = new AtomicInteger(window.getSize()); + for (LumberjackMessage msg : window) { + if (process) { + messageProcessor.onMessageReceived(msg.getPayload(), success -> { + if (success) { + if (remaining.decrementAndGet() == 0) { + ctx.writeAndFlush(new LumberjackAck(window.getVersion(), window.getSize())); } - }); - } + } else { + ctx.close(); + // Mark that we shouldn't process the next messages that are already decoded and are waiting in netty queues + process = false; + } + }); } - } finally { - ctx.flush(); - } - } - - /** - * Notify message processed if end of window - * - * @param ctx - * @param sequenceNumber - * @param window - */ - private void notifyMessageProcessed(ChannelHandlerContext ctx, int sequenceNumber, LumberjackWindow window) { - if (sequenceNumber == window.getSize()) { - ctx.writeAndFlush(new LumberjackAck(window.getVersion(), sequenceNumber)); } }
