allthingssecurity commented on code in PR #27431:
URL: https://github.com/apache/camel/pull/27431#discussion_r4195179972
##########
components/camel-reactive-streams/src/main/java/org/apache/camel/component/reactive/streams/ReactiveStreamsConsumer.java:
##########
@@ -98,13 +115,26 @@ private boolean doSend(Exchange exchange, AsyncCallback
callback) {
ExecutorService executorService = this.executor;
if (executorService != null && this.isRunAllowed()) {
- executorService.execute(() ->
this.getAsyncProcessor().process(exchange, doneSync -> {
- if (exchange.getException() != null) {
- getExceptionHandler().handleException("Error processing
exchange", exchange, exchange.getException());
+ executorService.execute(() -> {
+ ROUTING.set(this);
+ try {
+ this.getAsyncProcessor().process(exchange, doneSync -> {
+ Exception cause = exchange.getException();
+ if (cause instanceof RejectedExecutionException &&
!isRunAllowed()) {
+ // an item queued when the consumer was stopped
from one of its own exchanges: one line
+ // per item, as there can be up to
maxInflightExchanges of them
+ LOG.warn("Item {} of stream {} not routed as the
consumer is stopped",
Review Comment:
Thanks. I kept one line per item on purpose: each line names the exchange id
of an item that was taken from the stream and not routed, which is what someone
tracing a lost message needs, and it only happens for a stop from the
consumer's own pool. Happy to switch to a single count line in a follow-up if
that turns out to be noisy in practice.
_Claude Code on behalf of allthingssecurity_
--
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]
