diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java index 567fe9673e91f..7a31579e1d618 100644 --- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java +++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java @@ -65,6 +65,8 @@ public class NatsConsumer extends DefaultConsumer { private ExecutorService executor; private Connection connection; private Dispatcher dispatcher; + // the message handler of the dispatcher, which keeps the messages in flight of that dispatcher (one per start) + private volatile NatsConsumingTask.CamelNatsMessageHandler dispatcherHandler; private boolean active; // set by the consuming task, and again by the push consumer keeper, and read when the consumer stops private volatile JetStreamSubscription jetStreamSubscription; @@ -127,6 +129,33 @@ protected void doStop() throws Exception { } catch (final Exception e) { this.getExceptionHandler().handleException("Error during unsubscribing", e); } + // a JetStream push subscription belongs to the dispatcher (it cannot unsubscribe directly) and is on its + // deliver subject, not on the topic: end it through the dispatcher, which matters when the connection is + // shared and stays open. This does not interrupt the dispatcher thread, which may still run an exchange + JetStreamSubscription pushSubscription = this.jetStreamSubscription; + if (pushSubscription != null && pushSubscription.isActive()) { + try { + this.dispatcher.unsubscribe(pushSubscription); + } catch (final Exception e) { + this.getExceptionHandler().handleException("Error during unsubscribing", e); + } + } + // the dispatcher was created by this consumer: close it (which ends its thread) once no message is in + // flight, as closing it interrupts its thread, and Camel stops a consumer before it waits for the exchanges + // in flight + final Dispatcher stoppedDispatcher = this.dispatcher; + final Connection stoppedConnection = this.connection; + Runnable close = () -> closeDispatcher(stoppedConnection, stoppedDispatcher); + NatsConsumingTask.CamelNatsMessageHandler handler = this.dispatcherHandler; + if (handler != null) { + // the handler of this dispatcher only: a message of an earlier run that is still in flight (after a + // forced stop) closes the dispatcher of its own run + handler.closeWhenIdle(close); + } else { + close.run(); + } + this.dispatcher = null; + this.dispatcherHandler = null; } if (ObjectHelper.isNotEmpty(this.jetStreamSubscription) && this.jetStreamSubscription.isActive()) { @@ -156,6 +185,19 @@ protected void doStop() throws Exception { super.doStop(); } + /** + * Closes a dispatcher of this consumer, unless its connection is closed (which closed it). + */ + private void closeDispatcher(Connection c, Dispatcher d) { + if (c != null && c.getStatus() != Connection.Status.CLOSED) { + try { + c.closeDispatcher(d); + } catch (final Exception e) { + LOG.debug("Error closing the dispatcher of the stopped consumer", e); + } + } + } + public boolean isActive() { return this.active; } @@ -204,7 +246,14 @@ public void run() { setupStandardNatsConsumer(topic, queueName, maxMessages); } } catch (final Exception e) { - NatsConsumer.this.getExceptionHandler().handleException("Error during processing", e); + if (!NatsConsumer.this.isRunAllowed()) { + // the consumer stopped meanwhile (its stop closes the dispatcher that a JetStream push subscribe + // uses, which then fails): nothing to report + LOG.debug("Not consuming from topic {} as the consumer is stopping", + getEndpoint().getConfiguration().getTopic(), e); + } else { + NatsConsumer.this.getExceptionHandler().handleException("Error during processing", e); + } } } @@ -276,6 +325,7 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc runPullFetchLoop(NatsConsumer.this.jetStreamSubscription, messageHandler, cc); } else { NatsConsumer.this.dispatcher = this.connection.createDispatcher(messageHandler); + NatsConsumer.this.dispatcherHandler = messageHandler; PushSubscribeOptions pushOptions = PushSubscribeOptions.builder() .configuration(cc) @@ -538,7 +588,9 @@ private void acknowledged(Message msg) { } private void setupStandardNatsConsumer(String topic, String queueName, Integer maxMessages) { - NatsConsumer.this.dispatcher = connection.createDispatcher(new CamelNatsMessageHandler()); + CamelNatsMessageHandler messageHandler = new CamelNatsMessageHandler(); + NatsConsumer.this.dispatcher = connection.createDispatcher(messageHandler); + NatsConsumer.this.dispatcherHandler = messageHandler; if (ObjectHelper.isNotEmpty(queueName)) { NatsConsumer.this.dispatcher = NatsConsumer.this.dispatcher.subscribe(topic, queueName); } else { @@ -556,9 +608,51 @@ class CamelNatsMessageHandler implements MessageHandler { final boolean ackPolicyNone = configuration.getAckPolicy() == AckPolicy.None; final boolean manualAckEnabled = configuration.isManualAck() && configuration.isJetstreamEnabled(); + // the messages that this handler is processing, and the close of its dispatcher (the consumer stopped) + // 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 pendingClose; + + /** + * Closes the dispatcher of this handler now, or once the message in flight is done. + */ + void closeWhenIdle(Runnable close) { + synchronized (inFlightLock) { + if (messagesInFlight > 0) { + // closed by onMessage once the message is done + pendingClose = close; + return; + } + } + // no message in flight: close now (outside the lock, as the close may block) + close.run(); + } @Override public void onMessage(Message msg) throws InterruptedException { + synchronized (inFlightLock) { + messagesInFlight++; + } + try { + processMessage(msg); + } finally { + Runnable close = null; + synchronized (inFlightLock) { + if (--messagesInFlight == 0) { + close = pendingClose; + pendingClose = null; + } + } + if (close != null) { + // the consumer stopped while this message was processed: close its dispatcher now (on this + // thread, which the close interrupts once the message is done) + close.run(); + } + } + } + + private void processMessage(Message msg) throws InterruptedException { LOG.debug("Received Message: {}", msg); final Exchange exchange = NatsConsumer.this.createExchange(false); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java new file mode 100644 index 0000000000000..ea1f942f82b49 --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerJetStreamPushStopTest.java @@ -0,0 +1,438 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.nats; + +import java.lang.reflect.Proxy; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import io.nats.client.Connection; +import io.nats.client.Dispatcher; +import io.nats.client.JetStream; +import io.nats.client.JetStreamManagement; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Message; +import io.nats.client.MessageHandler; +import io.nats.client.PushSubscribeOptions; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.ServerInfo; +import io.nats.client.impl.NatsJetStreamMetaData; +import io.nats.client.impl.NatsMessage; +import org.apache.camel.Consumer; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.spi.ExceptionHandler; +import org.apache.camel.support.DefaultConsumer; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A JetStream push subscription belongs to the dispatcher of the consumer, and jnats refuses to unsubscribe it directly + * ("Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly"). When the connection is shared + * (connection option) it stays open when the route stops, so the consumer must unsubscribe through its dispatcher; + * otherwise the stopped route keeps getting messages, and after a restart every message is routed twice. + */ +class NatsConsumerJetStreamPushStopTest extends CamelTestSupport { + + private final FakeServer server = new FakeServer(); + private final CountDownLatch slowEntered = new CountDownLatch(1); + private final CountDownLatch releaseSlow = new CountDownLatch(1); + private final AtomicBoolean slowInterrupted = new AtomicBoolean(); + private final AtomicBoolean slowDone = new AtomicBoolean(); + + @Override + protected RouteBuilder createRouteBuilder() { + context.getRegistry().bind("conn", server.connection()); + context.getRegistry().bind("cc", ConsumerConfiguration.builder().deliverPolicy(DeliverPolicy.New).build()); + return new RouteBuilder() { + @Override + public void configure() { + from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullSubscription=false" + + "&consumerConfiguration=#cc") + .routeId("orders") + .convertBodyTo(String.class) + .process(e -> { + if ("slow".equals(e.getMessage().getBody())) { + // an exchange in flight while the route stops, such as a call that waits on I/O + slowEntered.countDown(); + try { + releaseSlow.await(20, TimeUnit.SECONDS); + } catch (InterruptedException ie) { + slowInterrupted.set(true); + Thread.currentThread().interrupt(); + throw ie; + } + slowDone.set(true); + } + }) + .to("mock:result"); + } + }; + } + + @Test + void stoppedRouteGetsNoMessagesAndRestartedRouteGetsEachMessageOnce() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + context.getRouteController().stopRoute("orders"); + int afterStop = server.activeSubscriptions(); + + // the route is started again: one subscription, each message routed once + context.getRouteController().startRoute("orders"); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.activeSubscriptions() > 0); + mock.reset(); + mock.expectedBodiesReceived("order-2"); + // and no second copy shortly after + mock.setAssertPeriod(500); + server.publish("order-2"); + mock.assertIsSatisfied(); + assertEquals(0, afterStop, "no subscription left once the route stopped"); + assertEquals(1, server.activeSubscriptions(), "one subscription once the route started again"); + assertEquals(1, server.closedDispatchers(), "the dispatcher of the first run is closed"); + } + + @Test + void stoppingTheRouteDoesNotInterruptTheExchangeInFlight() throws Exception { + server.publish("slow"); + assertTrue(slowEntered.await(10, TimeUnit.SECONDS), "the exchange is in flight"); + + // the shutdown strategy stops the consumer (it is not Suspendable) and then waits for the exchange in flight + Thread stopper = new Thread(() -> { + try { + context.getRouteController().stopRoute("orders"); + } catch (Exception e) { + throw new IllegalStateException(e); + } + }, "route-stopper"); + stopper.start(); + await().atMost(10, TimeUnit.SECONDS) + .until(() -> ServiceHelper.isStopped(context.getRoute("orders").getConsumer())); + assertEquals(0, server.activeSubscriptions(), "no subscription left once the consumer stopped"); + releaseSlow.countDown(); + stopper.join(TimeUnit.SECONDS.toMillis(20)); + + assertFalse(slowInterrupted.get(), "the step in flight must not be interrupted by the stop"); + assertTrue(slowDone.get(), "the step in flight completes"); + // the dispatcher (and its thread) is closed once the message is done + await().atMost(10, TimeUnit.SECONDS).until(() -> server.closedDispatchers() == 1); + } + + @Test + void eachRunOfTheConsumerClosesItsOwnDispatcher() throws Exception { + server.publish("slow"); + assertTrue(slowEntered.await(10, TimeUnit.SECONDS), "the exchange is in flight"); + + // a route policy (such as a throttling policy) stops and starts the same consumer, which is not Suspendable, + // while an exchange is in flight: each start makes a new dispatcher + Consumer consumer = context.getRoute("orders").getConsumer(); + ServiceHelper.stopService(consumer); + assertEquals(0, server.closedDispatchers(), "the dispatcher of the exchange in flight is not closed yet"); + ServiceHelper.startService(consumer); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.activeSubscriptions() > 0); + + // the second run has no message in flight, so its dispatcher is closed when it stops + ServiceHelper.stopService(consumer); + assertEquals(1, server.closedDispatchers(), "the dispatcher of the second run is closed when it stops"); + + // the exchange of the first run is done: the dispatcher of the first run is closed too + releaseSlow.countDown(); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.closedDispatchers() == 2); + assertFalse(slowInterrupted.get(), "the step in flight must not be interrupted by the stops"); + } + + @Test + void stopWhileSubscribingIsNotReportedAsAnError() throws Exception { + DefaultConsumer consumer = (DefaultConsumer) context.getRoute("orders").getConsumer(); + ServiceHelper.stopService(consumer); + List reported = new CopyOnWriteArrayList<>(); + consumer.setExceptionHandler(new ExceptionHandler() { + @Override + public void handleException(Throwable exception) { + reported.add(exception); + } + + @Override + public void handleException(String message, Throwable exception) { + reported.add(exception); + } + + @Override + public void handleException(String message, Exchange exchange, Throwable exception) { + reported.add(exception); + } + }); + + // the consumer stops while its task subscribes: the stop closes the dispatcher that the task passes to the + // subscribe, which then fails with "Dispatcher is closed" + server.beforeSubscribe = () -> ServiceHelper.stopService(consumer); + ServiceHelper.startService(consumer); + await().atMost(10, TimeUnit.SECONDS).until(() -> server.refusedSubscribes() == 1); + + // the default exception handler would log it at WARN, but a stop is not an error + await().during(Duration.ofMillis(500)).atMost(5, TimeUnit.SECONDS) + .untilAsserted(() -> assertTrue(reported.isEmpty(), () -> "reported: " + reported)); + assertEquals(0, server.activeSubscriptions(), "no subscription left once the consumer stopped"); + } + + /** + * A JetStream server with one stream, faked at the level of the jnats interfaces the consumer calls. Each push + * subscription has its own consumer (DeliverPolicy.New starts after the last published message) which delivers on + * its own thread while the subscription is active. Like jnats, a subscription that belongs to a dispatcher cannot + * unsubscribe directly, the dispatcher unsubscribes it by subscription (not by the topic: a push subscription is on + * its deliver subject), and closing the dispatcher unsubscribes all its subscriptions and interrupts the thread + * that delivers its messages (jnats NatsDispatcher.stop cancels its thread with Future.cancel(true)). A closed + * dispatcher refuses to subscribe or unsubscribe ("Dispatcher is closed"). + */ + static final class FakeServer { + final List stream = new ArrayList<>(); + final List subscriptions = new ArrayList<>(); + final Map bySubscription = new IdentityHashMap<>(); + final Set closed = Collections.newSetFromMap(new IdentityHashMap<>()); + // runs on the thread of the consumer, before a push subscribe reaches the server + volatile Runnable beforeSubscribe; + int refusedSubscribes; + + final class FakeSubscription { + final Dispatcher dispatcher; + Thread thread; + long next; + boolean active = true; + final List redeliver = new ArrayList<>(); + final Map naks = new HashMap<>(); + + FakeSubscription(Dispatcher dispatcher) { + this.dispatcher = dispatcher; + } + } + + synchronized void publish(String body) { + stream.add(body.getBytes(StandardCharsets.UTF_8)); + notifyAll(); + } + + synchronized int refusedSubscribes() { + return refusedSubscribes; + } + + synchronized int closedDispatchers() { + return closed.size(); + } + + synchronized int activeSubscriptions() { + return (int) subscriptions.stream().filter(s -> s.active).count(); + } + + Connection connection() { + JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); + JetStream js = proxy(JetStream.class, (name, args) -> { + if ("subscribe".equals(name) && args.length == 6 && args[5] instanceof PushSubscribeOptions o) { + Runnable hook = beforeSubscribe; + if (hook != null) { + hook.run(); + } + return subscribe(o.getConsumerConfiguration(), (Dispatcher) args[2], (MessageHandler) args[3]); + } + throw new UnsupportedOperationException(name); + }); + ServerInfo info = new ServerInfo("INFO {\"server_id\":\"fake\",\"jetstream\":true}"); + return proxy(Connection.class, (name, args) -> switch (name) { + case "getServerInfo" -> info; + case "jetStreamManagement" -> jsm; + case "jetStream" -> js; + case "createDispatcher" -> createDispatcher(); + case "closeDispatcher" -> { + closeDispatcher((Dispatcher) args[0]); + yield null; + } + case "getStatus" -> Connection.Status.CONNECTED; + default -> null; + }); + } + + private Dispatcher createDispatcher() { + Dispatcher[] self = new Dispatcher[1]; + self[0] = proxy(Dispatcher.class, (name, args) -> { + synchronized (this) { + if (closed.contains(self[0])) { + throw new IllegalStateException("Dispatcher is closed"); + } + } + if ("unsubscribe".equals(name) && args[0] instanceof JetStreamSubscription s) { + unsubscribe(s); + } + // unsubscribe by subject: a push subscription is on its deliver subject, not on the topic + return self[0]; + }); + return self[0]; + } + + private synchronized void unsubscribe(JetStreamSubscription s) { + FakeSubscription f = bySubscription.get(s); + if (f != null) { + f.active = false; + notifyAll(); + } + } + + private synchronized void closeDispatcher(Dispatcher d) { + if (!closed.add(d)) { + throw new IllegalArgumentException("Dispatcher is already closed."); + } + for (FakeSubscription f : subscriptions) { + if (f.dispatcher == d) { + f.active = false; + // the thread of the dispatcher is cancelled with an interrupt + f.thread.interrupt(); + } + } + notifyAll(); + } + + private synchronized JetStreamSubscription subscribe( + ConsumerConfiguration cc, Dispatcher dispatcher, MessageHandler handler) { + if (closed.contains(dispatcher)) { + refusedSubscribes++; + throw new IllegalStateException("Dispatcher is closed"); + } + FakeSubscription f = new FakeSubscription(dispatcher); + f.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.New ? stream.size() + 1 : 1; + subscriptions.add(f); + JetStreamSubscription sub = proxy(JetStreamSubscription.class, (name, args) -> switch (name) { + case "unsubscribe" -> throw new IllegalStateException( + "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly."); + case "isActive" -> { + synchronized (this) { + yield f.active; + } + } + case "getDispatcher" -> f.dispatcher; + default -> null; + }); + bySubscription.put(sub, f); + Thread t = new Thread(() -> deliver(f, sub, handler), "fake-push-" + subscriptions.size()); + t.setDaemon(true); + f.thread = t; + t.start(); + return sub; + } + + private void deliver(FakeSubscription f, JetStreamSubscription sub, MessageHandler handler) { + while (true) { + long seq; + synchronized (this) { + while (f.active && f.redeliver.isEmpty() && f.next > stream.size()) { + try { + wait(); + } catch (InterruptedException e) { + return; + } + } + if (!f.active) { + return; + } + seq = f.redeliver.isEmpty() ? f.next++ : f.redeliver.remove(0); + } + try { + handler.onMessage(message(f, sub, seq)); + } catch (InterruptedException e) { + return; + } + } + } + + private Message message(FakeSubscription f, JetStreamSubscription sub, long seq) { + byte[] data; + synchronized (this) { + data = stream.get((int) seq - 1); + } + // a JetStream message carries its metadata in the reply subject of the acknowledgement + NatsJetStreamMetaData meta = new NatsJetStreamMetaData( + new NatsMessage("orders", "$JS.ACK.ORDERS.c.1." + seq + "." + seq + ".0.0", data) { + @Override + public boolean isJetStream() { + return true; + } + }); + return proxy(Message.class, (name, args) -> switch (name) { + case "getData" -> data; + case "getSubject" -> "orders"; + case "getSID" -> "1"; + case "getSubscription" -> sub; + case "isJetStream" -> true; + case "isStatusMessage" -> false; + case "metaData" -> meta; + case "nak", "nakWithDelay" -> { + synchronized (this) { + // like maxDeliver=3: a message nak'ed three times is not delivered again + if (f.naks.merge(seq, 1, Integer::sum) < 3 && f.active) { + f.redeliver.add(seq); + notifyAll(); + } + } + yield null; + } + default -> null; + }); + } + + @FunctionalInterface + interface Handler { + Object handle(String method, Object[] args) throws Exception; + } + + @SuppressWarnings("unchecked") + static T proxy(Class type, Handler handler) { + return (T) Proxy.newProxyInstance(FakeServer.class.getClassLoader(), new Class[] { type }, + (p, m, a) -> { + if (m.getDeclaringClass() == Object.class) { + return switch (m.getName()) { + case "hashCode" -> System.identityHashCode(p); + case "equals" -> p == a[0]; + default -> type.getSimpleName() + "@fake"; + }; + } + Object r = handler.handle(m.getName(), a == null ? new Object[0] : a); + if (r == null && m.getReturnType() == boolean.class) { + return false; + } + return r; + }); + } + } +}