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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
