davsclaus commented on code in PR #27431:
URL: https://github.com/apache/camel/pull/27431#discussion_r4195110768
##########
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:
Optional: with the defaults this can be up to 128 WARN lines for one stop.
Counting them and logging one line with the count (for example when the last
one completes, or at the next start) would be quieter; I see the per-item
choice is deliberate, so fine as it is.
--
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]
