You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "apupier (via GitHub)" <gi...@apache.org> on 2026/10/09 10:11:12 UTC
[PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
apupier opened a new pull request, #27616:
URL: https://github.com/apache/camel/pull/27616
# Description
<!--
- Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
-->
# Target
- [ ] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [ ] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it).
<!--
# *Note*: trivial changes like, typos, minor documentation fixes and other small items do not require a JIRA issue. In this case your pull request should address just this issue, without pulling in other changes.
-->
# Apache Camel coding standards and style
- [ ] I checked that each commit in the pull request has a meaningful subject line and body.
<!--
If you're unsure, you can format the pull request title like `[CAMEL-XXX] Fixes bug in camel-file component`, where you replace `CAMEL-XXX` with the appropriate JIRA issue.
-->
- [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
<!--
You can run the aforementioned command in your module so that the build auto-formats your code. This will also be verified as part of the checks and your PR may be rejected if if there are uncommited changes after running `mvn clean install -DskipTests`.
You can learn more about the contribution guidelines at https://github.com/apache/camel/blob/main/CONTRIBUTING.md
-->
# AI-assisted contributions
- [ ] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27616:
URL: https://github.com/apache/camel/pull/27616#discussion_r4247509723
##########
components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackOutOfOrderAckTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.camel.component.lumberjack;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@DisabledOnOs(value = { OS.LINUX },
+ architectures = { "s390x" },
+ disabledReason = "This test does not run reliably on multiple platforms (see CAMEL-21438)")
+@Isolated
+class LumberjackOutOfOrderAckTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "lumberjack-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+ private static final String MOCK_OUTPUT = "mock:output-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+
+ private final CountDownLatch firstMessageProcessing = new CountDownLatch(1);
+ private final CountDownLatch holdFirstMessage = new CountDownLatch(1);
+ private final AtomicBoolean first = new AtomicBoolean(true);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("lumberjack:0.0.0.0:0").routeId(ROUTE_ID)
+ .threads(5)
+ .process(exchange -> {
+ if (first.compareAndSet(true, false)) {
+ firstMessageProcessing.countDown();
+ if (!holdFirstMessage.await(5, TimeUnit.SECONDS)) {
+ throw new AssertionError("Timed out waiting for holdFirstMessage to be released");
+ }
+ }
+ })
+ .to(MOCK_OUTPUT);
+ }
+ };
+ }
+
+ private int getActualPort() {
+ return ((LumberjackConsumer) context.getRoute(ROUTE_ID).getConsumer()).getLocalPort();
+ }
+
+ @Test
+ void shouldNotSendAckUntilAllMessagesComplete() throws Exception {
+ MockEndpoint mock = getMockEndpoint(MOCK_OUTPUT);
+ mock.expectedMessageCount(14);
+
+ NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
+ try {
+ List<Integer> responses = new CopyOnWriteArrayList<>();
+ ChannelInitializer<Channel> initializer = new ChannelInitializer<>() {
+ @Override
+ protected void initChannel(Channel ch) {
+ ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+ assertThat(msg.readUnsignedByte()).isEqualTo((short) '2');
+ assertThat(msg.readUnsignedByte()).isEqualTo((short) 'A');
+ responses.add(msg.readInt());
+ }
+ });
+ }
+ };
+
+ Channel channel = new Bootstrap()
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .handler(initializer)
+ .connect("127.0.0.1", getActualPort()).sync().channel();
+
+ try {
+ // Send window of 15 messages
+ channel.writeAndFlush(LumberjackUtil.readSample("io/window15.bin"));
+
+ // Wait until the first message has arrived and is blocked
+ boolean firstStarted = firstMessageProcessing.await(5, TimeUnit.SECONDS);
+ assertThat(firstStarted).as("First message should start processing").isTrue();
+
+ // Wait until the other 14 messages (including the 15th with sequenceNumber == windowSize) complete
Review Comment:
💡 **Suggestion:** `Thread.sleep(200)` is technically redundant under the new countdown logic — an ACK is structurally impossible while the first message is still held (remaining will be 1, never 0). It still serves as a regression guard. Consider adding a comment to explain why it's there:
```suggestion
// Give the Netty pipeline a moment to deliver any in-flight response.
// Under the new countdown logic this sleep is technically redundant (ACK is
// structurally blocked), but it guards against regressions to the old
// sequence-number-equality gate where a premature ACK would be in-flight.
Thread.sleep(200);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27616:
URL: https://github.com/apache/camel/pull/27616#issuecomment-6080119281
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-lumberjack`
---
<details><summary>:microscope: Scalpel shadow comparison — Scalpel: 9 of 704 tested, 3 compile-only — current: 9 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **9 affected modules** (current approach: 9).
**Skip-tests mode would test 9 modules** (1 direct + 0 downstream), **skip tests for 3** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (9)</summary>
- `camel-jbang-mcp` ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
- `camel-jbang-plugin-mcp` ← depends on affected reactor module org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` ← depends on affected reactor module org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
- `camel-jbang-plugin-validate` ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` ← depends on affected reactor module org.apache.camel:camel-launcher
- `camel-lumberjack` ← components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java, components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackUtil.java, own pom components/camel-lumberjack/pom.xml changed
- `camel-yaml-dsl-validator` ← depends on affected reactor module org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` ← depends on affected reactor module org.apache.camel:camel-yaml-dsl-validator
</details>
<details><summary>Modules with tests skipped (3)</summary>
- `camel-itest`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
</details>
> :information_source: Shadow mode — Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
:warning: **Some tests are disabled on GitHub Actions** (`@DisabledIfSystemProperty(named = "ci.env.name")`) and require manual verification:
- `components/camel-lumberjack`: 1 test(s) disabled on GitHub Actions
<details><summary>All tested modules (36 modules, 6m 52s total)</summary>
**Total reactor time:** 6m 52s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 53.2s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 48.5s | SUCCESS |
| Camel :: JBang :: MCP | 44.2s | SUCCESS |
| Camel :: Lumberjack | 42.8s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 25.8s | SUCCESS |
| Camel :: YAML DSL :: Validator | 22.0s | SUCCESS |
| Camel :: Component DSL | 21.8s | SUCCESS |
| Camel :: YAML DSL | 21.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 17.1s | SUCCESS |
| Camel :: Docs | 17.0s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 14.5s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 12.3s | SUCCESS |
| Camel :: Kamelet Main | 11.3s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 8.8s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 8.4s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 7.5s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 6.9s | SUCCESS |
| Camel :: All Components Sync point | 6.0s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.7s | SUCCESS |
| Camel :: Catalog :: Maven | 3.4s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.5s | SUCCESS |
| Camel :: Assembly | 1.6s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.5s | SUCCESS |
| Camel :: Coverage | 1.4s | SUCCESS |
| Camel :: JBang :: Main | 1.2s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.0s | SUCCESS |
| Camel :: JBang :: Integration tests | 1.0s | SUCCESS |
| Camel :: Catalog :: Console | 0.9s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.7s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.6s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (53.2s)
- `Camel :: JBang :: Plugin :: TUI` (48.5s)
- `Camel :: JBang :: MCP` (44.2s)
- `Camel :: Lumberjack` (42.8s)
- `Camel :: Catalog :: Camel Catalog` (25.8s)
- `Camel :: YAML DSL :: Validator` (22.0s)
- `Camel :: Component DSL` (21.8s)
- `Camel :: YAML DSL` (21.5s)
- `Camel :: JBang :: Plugin :: Validate` (17.1s)
- `Camel :: Docs` (17.0s)
- `Camel :: JBang :: Plugin :: Testing` (14.5s)
- `Camel :: JBang :: Plugin :: Kubernetes` (12.3s)
- `Camel :: Kamelet Main` (11.3s)
- `Camel :: YAML DSL :: Deserializers` (8.8s)
- `Camel :: Catalog :: Camel Route Parser` (8.4s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (7.5s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (6.9s)
- `Camel :: All Components Sync point` (6.0s)
- `Camel :: YAML DSL :: Maven Plugins` (3.7s)
- `Camel :: Catalog :: Maven` (3.4s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37916039974)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27616:
URL: https://github.com/apache/camel/pull/27616#discussion_r4246112234
##########
components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackUtil.java:
##########
@@ -97,7 +98,17 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
// send 5 frame windows, without pausing
windows.stream().forEach(window -> channel.writeAndFlush(readSample(String.format("io/window%s.bin", window))));
if (waitForResult) {
- Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> windows.size() == responses.size());
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(
+ () -> {
+ List<Integer> received;
+ synchronized (responses) {
+ received = new ArrayList<>(responses);
+ }
+ Assertions.assertThat(responses)
+ .as("Expected one ACK per window frame sent (%d), but received: %s", windows.size(),
+ received)
+ .hasSize(windows.size());
Review Comment:
⚠️ **Thread-safety gap — snapshot taken but assertion reads the live list.** The `received` snapshot is created under the lock, but `Assertions.assertThat(responses)` still reads `responses` directly. On every failed poll Awaitility retries, AssertJ builds the failure message (the `%s` argument and the actual value) by iterating `responses` while the Netty I/O thread may be adding to it. This can throw a `ConcurrentModificationException`, which Awaitility does **not** retry (it only retries `AssertionError`), so the test would fail with a confusing stack trace. The fix is to assert on `received` throughout:
```suggestion
Assertions.assertThat(received)
.as("Expected one ACK per window frame sent (%d), but received: %s", windows.size(),
received)
.hasSize(windows.size());
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "gnodet-bot (via GitHub)" <gi...@apache.org>.
gnodet-bot commented on code in PR #27616:
URL: https://github.com/apache/camel/pull/27616#discussion_r4246934540
##########
components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackOutOfOrderAckTest.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.camel.component.lumberjack;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@DisabledOnOs(value = { OS.LINUX },
+ architectures = { "s390x" },
+ disabledReason = "This test does not run reliably on multiple platforms (see CAMEL-21438)")
+@Isolated
+public class LumberjackOutOfOrderAckTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "lumberjack-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+ private static final String MOCK_OUTPUT = "mock:output-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+
+ private final CountDownLatch firstMessageProcessing = new CountDownLatch(1);
+ private final CountDownLatch holdFirstMessage = new CountDownLatch(1);
+ private final AtomicBoolean first = new AtomicBoolean(true);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("lumberjack:0.0.0.0:0").routeId(ROUTE_ID)
+ .threads(5)
+ .process(exchange -> {
+ if (first.compareAndSet(true, false)) {
+ firstMessageProcessing.countDown();
+ holdFirstMessage.await(5, TimeUnit.SECONDS);
Review Comment:
⚠️ **Silent timeout degradation** — `CountDownLatch.await()` returns `boolean` (`true` if it counted down, `false` on timeout). Discarding the return value means that if the first message never arrives within 5 seconds, the test continues silently and the subsequent `isEmpty()` check becomes meaningless noise.
```suggestion
if (!holdFirstMessage.await(5, TimeUnit.SECONDS)) {
throw new AssertionError("Timed out waiting for holdFirstMessage to be released");
}
```
##########
components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackOutOfOrderAckTest.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.camel.component.lumberjack;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.DisabledOnOs;
+import org.junit.jupiter.api.condition.OS;
+import org.junit.jupiter.api.parallel.Isolated;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@DisabledOnOs(value = { OS.LINUX },
+ architectures = { "s390x" },
+ disabledReason = "This test does not run reliably on multiple platforms (see CAMEL-21438)")
+@Isolated
+public class LumberjackOutOfOrderAckTest extends CamelTestSupport {
+
+ private static final String ROUTE_ID = "lumberjack-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+ private static final String MOCK_OUTPUT = "mock:output-" + LumberjackOutOfOrderAckTest.class.getSimpleName();
+
+ private final CountDownLatch firstMessageProcessing = new CountDownLatch(1);
+ private final CountDownLatch holdFirstMessage = new CountDownLatch(1);
+ private final AtomicBoolean first = new AtomicBoolean(true);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("lumberjack:0.0.0.0:0").routeId(ROUTE_ID)
+ .threads(5)
+ .process(exchange -> {
+ if (first.compareAndSet(true, false)) {
+ firstMessageProcessing.countDown();
+ holdFirstMessage.await(5, TimeUnit.SECONDS);
+ }
+ })
+ .to(MOCK_OUTPUT);
+ }
+ };
+ }
+
+ private int getActualPort() {
+ return ((LumberjackConsumer) context.getRoute(ROUTE_ID).getConsumer()).getLocalPort();
+ }
+
+ @Test
+ public void shouldNotSendAckUntilAllMessagesComplete() throws Exception {
+ MockEndpoint mock = getMockEndpoint(MOCK_OUTPUT);
+ mock.expectedMessageCount(14);
+
+ NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup(1);
+ try {
+ List<Integer> responses = new CopyOnWriteArrayList<>();
+ ChannelInitializer<Channel> initializer = new ChannelInitializer<>() {
+ @Override
+ protected void initChannel(Channel ch) {
+ ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
+ assertThat(msg.readUnsignedByte()).isEqualTo((short) '2');
+ assertThat(msg.readUnsignedByte()).isEqualTo((short) 'A');
+ responses.add(msg.readInt());
+ }
+ });
+ }
+ };
+
+ Channel channel = new Bootstrap()
+ .group(eventLoopGroup)
+ .channel(NioSocketChannel.class)
+ .handler(initializer)
+ .connect("127.0.0.1", getActualPort()).sync().channel();
+
+ try {
+ // Send window of 15 messages
+ channel.writeAndFlush(LumberjackUtil.readSample("io/window15.bin"));
+
+ // Wait until the first message has arrived and is blocked
+ boolean firstStarted = firstMessageProcessing.await(5, TimeUnit.SECONDS);
+ assertThat(firstStarted).as("First message should start processing").isTrue();
+
+ // Wait until the other 14 messages (including the 15th with sequenceNumber == windowSize) complete
+ mock.assertIsSatisfied(5, TimeUnit.SECONDS);
+
+ // Verify that no ACK has been sent yet because the first message is still in flight
+ assertThat(responses).as("No ACK should be sent before all messages in the window complete").isEmpty();
Review Comment:
⚠️ **Race window on `isEmpty()` assertion** — `mock.assertIsSatisfied()` confirms that 14 messages reached the mock endpoint, but it says nothing about the Netty ACK pipeline. On the old (buggy) code a premature ACK could be in-flight through the loopback socket at this point; depending on scheduling, the `isEmpty()` check might win the race and report a false pass. Wrapping this in a short Awaitility settle period makes the assertion deterministic:
```suggestion
// Wait until the other 14 messages (including the 15th with sequenceNumber == windowSize) complete
mock.assertIsSatisfied(5, TimeUnit.SECONDS);
// Give the Netty pipeline a moment to deliver any in-flight response
Thread.sleep(200);
// Verify that no ACK has been sent yet because the first message is still in flight
assertThat(responses).as("No ACK should be sent before all messages in the window complete").isEmpty();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "apupier (via GitHub)" <gi...@apache.org>.
apupier merged PR #27616:
URL: https://github.com/apache/camel/pull/27616
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27616:
URL: https://github.com/apache/camel/pull/27616#discussion_r4232854948
##########
components/camel-lumberjack/src/main/java/org/apache/camel/component/lumberjack/io/LumberjackMessageHandler.java:
##########
@@ -49,35 +50,22 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
@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) {
Review Comment:
Question on the root cause (non-blocking): in `LumberjackComponentTest` the route is `to("mock:output")`, which completes synchronously, so these callbacks fire in loop order on the channel's executor thread and the old `sequenceNumber == window.getSize()` gate already ACKed correctly there. The countdown is still the right behaviour for async routes; a test with a route where the last message completes before the first would cover it.
##########
components/camel-lumberjack/src/test/java/org/apache/camel/component/lumberjack/LumberjackUtil.java:
##########
@@ -97,7 +98,10 @@ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
// send 5 frame windows, without pausing
windows.stream().forEach(window -> channel.writeAndFlush(readSample(String.format("io/window%s.bin", window))));
if (waitForResult) {
- Awaitility.await().atMost(30, TimeUnit.SECONDS).until(() -> windows.size() == responses.size());
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(
+ () -> Assertions.assertThat(responses)
+ .as("Expected one ACK per window frame sent (%d), but received: %s", windows.size(), responses)
+ .hasSize(windows.size()));
Review Comment:
`responses` is read here without the `synchronized (responses)` used elsewhere in this method. On every failed poll AssertJ builds the failure message (the `%s` argument and the actual value), iterating the list while the Netty thread may be adding to it; a `ConcurrentModificationException` is not retried by Awaitility (only `AssertionError` is), so the test would fail with a confusing stack trace instead of this message. Asserting on a snapshot avoids that:
```suggestion
Awaitility.await().atMost(30, TimeUnit.SECONDS).untilAsserted(() -> {
List<Integer> received;
synchronized (responses) {
received = new ArrayList<>(responses);
}
Assertions.assertThat(received)
.as("Expected one ACK per window frame sent (%d), but received: %s", windows.size(), received)
.hasSize(windows.size());
});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Camel 25348 - improve LumberJack when ack in windows are arriving in different order [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27616:
URL: https://github.com/apache/camel/pull/27616#issuecomment-6078844913
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org