davsclaus commented on code in PR #28320:
URL: https://github.com/apache/camel/pull/28320#discussion_r4357367111
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -54,6 +54,11 @@ public class NatsConsumer extends DefaultConsumer {
private ExecutorService executor;
private Connection connection;
private Dispatcher dispatcher;
+ // the messages that the handler is processing, and the close of the
dispatcher of a stopped consumer that waits
+ // until none is: closing a dispatcher interrupts its thread, which runs
the route
+ private final Object inFlightLock = new Object();
+ private int messagesInFlight;
+ private Runnable closeDispatcherWhenIdle;
Review Comment:
Optional: this state belongs to one dispatcher, but it lives on the
consumer, which can be restarted in place with a new dispatcher and a new
`CamelNatsMessageHandler`. If an exchange of the stopped run is still in flight
(after a forced shutdown timeout) when the route starts again, the old
dispatcher only closes once the new run is idle too, and a second stop
overwrites `closeDispatcherWhenIdle`, which leaks the first dispatcher. Moving
`inFlightLock`/`messagesInFlight`/`closeDispatcherWhenIdle` into
`CamelNatsMessageHandler`, and keeping the handler next to `this.dispatcher` so
that `doStop` can call something like `handler.closeWhenIdle(connection,
dispatcher)`, would make it per dispatcher.
--
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]
