Croway commented on code in PR #26850:
URL: https://github.com/apache/camel/pull/26850#discussion_r4093241071
##########
core/camel-core-processor/src/main/java/org/apache/camel/processor/PollEnricher.java:
##########
@@ -467,6 +466,8 @@ public boolean process(Exchange exchange, AsyncCallback
callback) {
} catch (Exception e) {
exchange.setException(new CamelExchangeException("Error occurred
during aggregation", exchange, e));
+ // handover any synchronization, so the polled resource is
released (such as a file being rolled back)
+ handoverCompletions(resourceExchange, exchange);
Review Comment:
Handing the resource's completions over to the original exchange on this
failure path ties the polled file's commit or rollback to how the *original*
exchange ends. If the error is then recovered by redelivery or by
`doTry/doCatch`, a file that was never aggregated gets committed (moved to
`.camel` or deleted). This is data loss that `main` doesn't have: there the
completion is dropped and the file only stays in progress until a restart.
Minimal reproducer (put `a.txt` and `b.txt` in the inbox, send one message
to `direct:start`):
```java
errorHandler(defaultErrorHandler().maximumRedeliveries(1).redeliveryDelay(0));
from("direct:start")
.pollEnrich(fileUri("inbox?initialDelay=0&delay=10"), 2000, (original,
resource) -> {
if (attempts.incrementAndGet() == 1) {
throw new IllegalStateException("Transient failure"); // first
aggregation fails
}
original.getMessage().setBody(resource.getMessage().getBody(String.class));
return original;
});
```
1. The first attempt polls `b.txt` and the aggregation fails. b's on
completion is handed over to the exchange, and `b.txt` stays in progress.
2. On redelivery the consumer skips `b.txt` because it is in progress, and
returns `a.txt`, which aggregates fine (body = `A`).
3. The exchange completes successfully, so both completions run
`onComplete`. `a.txt` is committed (correct), and `b.txt` is committed too,
although its content never reached any exchange.
As a test asserting that the non-aggregated file is still in the inbox, this
fails on this branch (3/3 runs: `b.txt was not aggregated (out=A) but was
committed (moved to .camel)`) and passes with `PollEnricher` from `main`.
The same applies to the `variableReceive` early return at line 412.
Suggestion: on the failure paths, don't hand over. Complete the resource's
synchronizations right away as a failure, so the file is rolled back and can be
polled again, for example:
```java
resourceExchange.setException(cause);
UnitOfWorkHelper.doneSynchronizations(resourceExchange,
resourceExchange.getExchangeExtension().handoverCompletions());
```
`PollEnrichFileAggregationFailureTest` should still pass with this, as it
only checks that the file can be polled again.
_Claude Code on behalf of Croway_
--
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]
