From 7d8e7b8bf7593b8b2c18a72226165c58aa8aa89d Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:38:13 +0530 Subject: [PATCH 1/6] CAMEL-25815: camel-nats - the JetStream pull consumer subscribes again when the server deleted its consumer With jetstreamEnabled=true and the default pull subscription, and without a durableName, the consumer subscribes with an ephemeral JetStream consumer. The NATS server deletes an ephemeral consumer that had no waiting pull request for its inactive threshold (5 seconds by default) once no ack is pending, for example while the client was disconnected for a few seconds, and a server restart loses it. jnats does not give a pull subscription a new consumer: every fetch then ends with an empty list (the pull request has no responder, status 503), or once with 409 Consumer Deleted, which the pull loop logged and continued. The route stayed started and consumed nothing again until it was restarted. The pull loop now asks the server for its consumer when a fetch returned nothing (at most every 5 seconds) and handles 409 Consumer Deleted. When the consumer is gone it subscribes again: an ephemeral consumer is created again starting at the first stream sequence the route did not acknowledge (or after the last one fetched), so that nothing is skipped and the stream is not replayed from the deliver policy; a durable consumer is created again from its configuration. Co-Authored-By: Claude Opus 5.5 --- .../component/nats/DefaultNatsManualAck.java | 9 + .../camel/component/nats/NatsConsumer.java | 119 +++++++- .../NatsConsumerPullConsumerDeletedTest.java | 261 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 11 + 4 files changed, 392 insertions(+), 8 deletions(-) create mode 100644 components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java index 7d3b011663ba7..1ab6b87928293 100644 --- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java +++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java @@ -23,14 +23,22 @@ class DefaultNatsManualAck implements NatsManualAck { private final Message message; + private final Runnable onAcknowledged; DefaultNatsManualAck(Message message) { + this(message, () -> { + }); + } + + DefaultNatsManualAck(Message message, Runnable onAcknowledged) { this.message = message; + this.onAcknowledged = onAcknowledged; } @Override public void ack() { message.ack(); + onAcknowledged.run(); } @Override @@ -46,6 +54,7 @@ public void nakWithDelay(Duration delay) { @Override public void term() { message.term(); + onAcknowledged.run(); } @Override 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 af23bed3aac96..b6c3e45a1b610 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 @@ -19,10 +19,12 @@ import java.io.IOException; import java.time.Duration; import java.util.List; +import java.util.NavigableSet; +import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicLong; import io.nats.client.Connection; -import io.nats.client.Connection.Status; import io.nats.client.Dispatcher; import io.nats.client.JetStreamApiException; import io.nats.client.JetStreamManagement; @@ -34,7 +36,9 @@ import io.nats.client.PushSubscribeOptions; import io.nats.client.api.AckPolicy; import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; import io.nats.client.api.StreamConfiguration; +import io.nats.client.support.Status; import org.apache.camel.Exchange; import org.apache.camel.Processor; import org.apache.camel.spi.HeaderFilterStrategy; @@ -49,6 +53,8 @@ public class NatsConsumer extends DefaultConsumer { private static final Logger LOG = LoggerFactory.getLogger(NatsConsumer.class); + // an idle JetStream pull loop asks the server for its consumer at most this often + private static final long CONSUMER_CHECK_INTERVAL_NANOS = 5_000_000_000L; private final Processor processor; private ExecutorService executor; @@ -126,7 +132,7 @@ protected void doStop() throws Exception { if (ObjectHelper.isEmpty(configuration.getConnection()) && ObjectHelper.isNotEmpty(this.connection)) { LOG.debug("Closing Nats Connection"); - if (!this.connection.getStatus().equals(Status.CLOSED)) { + if (!this.connection.getStatus().equals(Connection.Status.CLOSED)) { this.connection.close(); } } @@ -145,6 +151,11 @@ class NatsConsumingTask implements Runnable { private final Connection connection; private final NatsConfiguration configuration; + // stream sequences of the JetStream messages fetched by the pull loop and not acknowledged yet, and the highest + // fetched: where a new ephemeral consumer must start when the server deleted the one this consumer used + private final NavigableSet unacknowledged = new ConcurrentSkipListSet<>(); + private final AtomicLong highestFetched = new AtomicLong(); + private long lastConsumerCheck = System.nanoTime(); NatsConsumingTask(Connection connection, NatsConfiguration configuration) { this.connection = connection; @@ -235,7 +246,7 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc pullOptions); NatsConsumer.this.setActive(true); - runPullFetchLoop(NatsConsumer.this.jetStreamSubscription, messageHandler); + runPullFetchLoop(NatsConsumer.this.jetStreamSubscription, messageHandler, cc); } else { NatsConsumer.this.dispatcher = this.connection.createDispatcher(messageHandler); @@ -262,17 +273,24 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc } } - private void runPullFetchLoop(JetStreamSubscription sub, CamelNatsMessageHandler messageHandler) { + private void runPullFetchLoop( + JetStreamSubscription subscription, CamelNatsMessageHandler messageHandler, ConsumerConfiguration cc) { + JetStreamSubscription sub = subscription; int batchSize = configuration.getPullBatchSize(); Duration fetchTimeout = Duration.ofMillis(configuration.getPullFetchTimeout()); while (NatsConsumer.this.isRunAllowed() && !Thread.currentThread().isInterrupted()) { try { List messages = sub.fetch(batchSize, fetchTimeout); + if (messages.isEmpty() && consumerCheckDue() && consumerGone(sub)) { + sub = subscribeAgain(sub, cc); + continue; + } for (Message msg : messages) { if (!NatsConsumer.this.isRunAllowed()) { break; } try { + track(msg); messageHandler.onMessage(msg); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -280,10 +298,19 @@ private void runPullFetchLoop(JetStreamSubscription sub, CamelNatsMessageHandler } } } catch (JetStreamStatusException e) { - // Synchronous pull calls may throw JetStreamStatusException for transient - // statuses such as 409 (consumer deleted/push-based). Log and continue so the - // consumer can recover when the consumer state is restored. + // Synchronous pull calls may throw JetStreamStatusException for statuses such as 409 + // (consumer deleted/push-based). Log and continue so the consumer can recover when the consumer + // state is restored. NatsConsumer.this.getExceptionHandler().handleException("JetStream status during pull fetch", e); + Status status = e.getStatus(); + if (status != null && Status.CONSUMER_DELETED.equals(status.getMessage())) { + try { + sub = subscribeAgain(sub, cc); + } catch (Exception ex) { + NatsConsumer.this.getExceptionHandler().handleException( + "Error subscribing again after the JetStream consumer was deleted", ex); + } + } } catch (IllegalStateException e) { // Subscription was closed (e.g. consumer is stopping), exit the loop LOG.debug("Pull subscription closed, stopping fetch loop", e); @@ -294,6 +321,79 @@ private void runPullFetchLoop(JetStreamSubscription sub, CamelNatsMessageHandler } } + /** + * Whether the server no longer has the consumer of this pull subscription. The server deletes an ephemeral + * consumer (no durable name) that saw no pull request for its inactive threshold (5 seconds by default), for + * example while the connection was down for a few seconds, and a server restart loses it. jnats then answers + * every fetch with an empty list (the pull request has no responder), and the subscription never receives + * anything again. + */ + private boolean consumerCheckDue() { + long now = System.nanoTime(); + if (now - lastConsumerCheck < CONSUMER_CHECK_INTERVAL_NANOS) { + return false; + } + lastConsumerCheck = now; + return true; + } + + private boolean consumerGone(JetStreamSubscription sub) { + try { + sub.getConsumerInfo(); + return false; + } catch (JetStreamApiException e) { + return e.getErrorCode() == 404; + } catch (Exception e) { + // not connected, timeout: the consumer may well still be there, the next fetch tells + LOG.debug("Cannot check the JetStream consumer of the pull subscription", e); + return false; + } + } + + /** + * Subscribes again after the server deleted the consumer. An ephemeral consumer is created again starting at + * the first message this consumer has not acknowledged, so that nothing is skipped and the stream is not + * replayed from its deliver policy; a durable consumer is created again from its configuration. + */ + private JetStreamSubscription subscribeAgain(JetStreamSubscription old, ConsumerConfiguration cc) + throws IOException, JetStreamApiException { + ConsumerConfiguration again = cc; + long start = unacknowledged.isEmpty() ? highestFetched.get() + 1 : unacknowledged.first(); + boolean resume = cc.getDurable() == null && highestFetched.get() > 0; + if (resume) { + again = ConsumerConfiguration.builder(cc) + .deliverPolicy(DeliverPolicy.ByStartSequence) + .startSequence(start) + .build(); + } + LOG.warn("The JetStream consumer of the pull subscription on topic {} is gone; subscribing again{}", + configuration.getTopic(), resume ? " from stream sequence " + start : ""); + JetStreamSubscription sub = this.connection.jetStream().subscribe(configuration.getTopic(), + PullSubscribeOptions.builder().configuration(again).build()); + unacknowledged.clear(); + NatsConsumer.this.jetStreamSubscription = sub; + try { + old.unsubscribe(); + } catch (Exception e) { + LOG.debug("Error unsubscribing the pull subscription of the deleted consumer", e); + } + return sub; + } + + private void track(Message msg) { + if (msg.isJetStream()) { + long seq = msg.metaData().streamSequence(); + unacknowledged.add(seq); + highestFetched.accumulateAndGet(seq, Math::max); + } + } + + private void acknowledged(Message msg) { + if (msg.isJetStream()) { + unacknowledged.remove(msg.metaData().streamSequence()); + } + } + private void setupStandardNatsConsumer(String topic, String queueName, Integer maxMessages) { NatsConsumer.this.dispatcher = connection.createDispatcher(new CamelNatsMessageHandler()); if (ObjectHelper.isNotEmpty(queueName)) { @@ -325,6 +425,7 @@ public void onMessage(Message msg) throws InterruptedException { public void onComplete(Exchange exchange) { LOG.debug("ACK"); msg.ack(); + acknowledged(msg); } @Override @@ -333,6 +434,7 @@ public void onFailure(Exchange exchange) { // ACK policy is none which means that we should auto ACK even if the message processed failed in Camel LOG.debug("ACK"); msg.ack(); + acknowledged(msg); } else { LOG.debug("NACK (delay:{})", configuration.getNackWait()); if (configuration.getNackWait() <= 0) { @@ -379,7 +481,8 @@ public void onFailure(Exchange exchange) { evalRedeliveryMessageTrait(msg, exchange)); if (manualAckEnabled) { - exchange.getIn().setHeader(NatsConstants.NATS_MANUAL_ACK, new DefaultNatsManualAck(msg)); + exchange.getIn().setHeader(NatsConstants.NATS_MANUAL_ACK, + new DefaultNatsManualAck(msg, () -> acknowledged(msg))); } NatsConsumer.this.processor.process(exchange); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java new file mode 100644 index 0000000000000..00e071cdcdf3c --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -0,0 +1,261 @@ +/* + * 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.List; +import java.util.concurrent.TimeUnit; + +import io.nats.client.Connection; +import io.nats.client.JetStream; +import io.nats.client.JetStreamApiException; +import io.nats.client.JetStreamManagement; +import io.nats.client.JetStreamStatusException; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Message; +import io.nats.client.PullSubscribeOptions; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.Error; +import io.nats.client.api.ServerInfo; +import io.nats.client.impl.NatsJetStreamMetaData; +import io.nats.client.impl.NatsMessage; +import io.nats.client.support.Status; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * The JetStream pull consumer (the default with jetstreamEnabled=true) uses an ephemeral consumer when no durableName + * is set. The NATS server deletes an ephemeral consumer that saw no pull request for its inactive threshold (5 seconds + * by default), for example while the client was disconnected for a few seconds, and a server restart loses it. The + * consumer must then subscribe again, starting at the first message it did not acknowledge, instead of fetching from + * the deleted consumer for ever. + */ +class NatsConsumerPullConsumerDeletedTest extends CamelTestSupport { + + private final FakeServer server = new FakeServer(); + + @Override + protected RouteBuilder createRouteBuilder() { + context.getRegistry().bind("conn", server.connection()); + return new RouteBuilder() { + @Override + public void configure() { + from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullFetchTimeout=100") + .convertBodyTo(String.class) + .to("mock:result"); + } + }; + } + + @Test + void consumesAgainAfterTheServerDeletedTheEphemeralConsumer() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + // the client was disconnected for longer than the inactive threshold: the server deleted the consumer + server.deleteConsumers(false); + + mock.reset(); + mock.expectedBodiesReceived("order-2", "order-3"); + server.publish("order-2"); + server.publish("order-3"); + mock.assertIsSatisfied(20_000); + } + + @Test + void consumesAgainAfterAWaitingPullGotConsumerDeleted() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + // the consumer is deleted while a pull request waits: that request gets 409 Consumer Deleted + server.deleteConsumers(true); + + mock.reset(); + mock.expectedBodiesReceived("order-2"); + server.publish("order-2"); + mock.assertIsSatisfied(20_000); + assertEquals(2, server.consumersCreated(), "one consumer created again"); + } + + /** + * A JetStream server with one stream, faked at the level of the jnats interfaces the consumer calls. A consumer + * delivers from its start sequence; once deleted, a fetch on it gets nothing (jnats ends the fetch on the 503 "no + * responders" status), and getConsumerInfo fails with 404. + */ + static final class FakeServer { + final List stream = new ArrayList<>(); + final List consumers = new ArrayList<>(); + int consumersCreated; + + final class FakeConsumer { + long next; + boolean deleted; + boolean tellDeleted; + final List redeliver = new ArrayList<>(); + } + + synchronized int consumersCreated() { + return consumersCreated; + } + + synchronized void publish(String body) { + stream.add(body.getBytes(StandardCharsets.UTF_8)); + notifyAll(); + } + + synchronized void deleteConsumers(boolean waitingRequest) { + for (FakeConsumer c : consumers) { + c.deleted = true; + c.tellDeleted = waitingRequest; + } + notifyAll(); + } + + Connection connection() { + JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); + JetStream js = proxy(JetStream.class, (name, args) -> { + if ("subscribe".equals(name) && args.length == 2 && args[1] instanceof PullSubscribeOptions o) { + return subscribe(o.getConsumerConfiguration()); + } + 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 "getStatus" -> Connection.Status.CONNECTED; + default -> null; + }); + } + + private synchronized JetStreamSubscription subscribe(ConsumerConfiguration cc) { + FakeConsumer c = new FakeConsumer(); + c.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.ByStartSequence ? cc.getStartSequence() : 1; + consumers.add(c); + consumersCreated++; + JetStreamSubscription[] self = new JetStreamSubscription[1]; + self[0] = proxy(JetStreamSubscription.class, (name, args) -> switch (name) { + case "fetch" -> fetch(c, self[0], (Integer) args[0], (Duration) args[1]); + case "getConsumerInfo" -> consumerInfo(c); + case "isActive" -> true; + default -> null; + }); + return self[0]; + } + + private synchronized Object consumerInfo(FakeConsumer c) throws JetStreamApiException { + if (c.deleted) { + throw new JetStreamApiException(Error.convert(new Status(404, "consumer not found"))); + } + return null; + } + + private synchronized List fetch(FakeConsumer c, JetStreamSubscription sub, int batch, Duration timeout) { + long deadline = System.nanoTime() + timeout.toNanos(); + List out = new ArrayList<>(); + while (true) { + if (c.deleted) { + if (c.tellDeleted) { + c.tellDeleted = false; + throw new JetStreamStatusException(new Status(409, Status.CONSUMER_DELETED), sub); + } + return out; + } + while (out.size() < batch && (!c.redeliver.isEmpty() || c.next <= stream.size())) { + long seq = c.redeliver.isEmpty() ? c.next++ : c.redeliver.remove(0); + out.add(message(c, sub, seq)); + } + long left = deadline - System.nanoTime(); + if (!out.isEmpty() || left <= 0) { + return out; + } + try { + TimeUnit.NANOSECONDS.timedWait(this, left); + } catch (InterruptedException e) { + // as jnats: an interrupted fetch keeps the interrupt and returns what it has + Thread.currentThread().interrupt(); + return out; + } + } + } + + private Message message(FakeConsumer c, JetStreamSubscription sub, long seq) { + byte[] 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" -> { + synchronized (this) { + c.redeliver.add(seq); + } + 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; + }); + } + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 5b1e7e99e39d9..23b8629d1f10e 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -1472,6 +1472,17 @@ in full and rejected by the broker. A settlement request the broker rejects is now reported through the consumer's exception handler. +=== camel-nats - the JetStream pull consumer subscribes again when its consumer was deleted + +With `jetstreamEnabled=true` and a pull subscription (the default), the `nats` consumer kept fetching from a JetStream +consumer that the server had deleted, and consumed nothing more until the route was restarted. It now subscribes again +when the server no longer has its consumer. An ephemeral consumer (no `durableName`), which the server deletes after +about 5 seconds without a pull request, for example while the client is disconnected, or loses on a restart, is created +again at the first message the route did not acknowledge. A durable consumer that was deleted, for example by an +administrator, is created again from its configuration, so it starts where its deliver policy says: with the default +`DeliverPolicy.All` at the start of the stream, and the route consumes the messages of the stream again. Set a +`consumerConfiguration` with another deliver policy to avoid that. + === camel-telegram - a failed send fails the exchange once When Telegram answers a message with an error status (for example `400` for an unknown chat), or with a body that From 3ac611e69d86403ace608a721434da8fa38ab619 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 11:15:55 +0530 Subject: [PATCH 2/6] CAMEL-25815: camel-nats - address review: subscribe again only for an ephemeral consumer A durable consumer is gone only when an administrator deleted it or after its inactiveThreshold; creating it again from its configuration replayed the stream from its deliver policy (the whole stream with the default DeliverPolicy.All). The pull loop now creates only an ephemeral consumer again; for a durable it logs a WARN once and keeps the subscription, so the fetches work again when the durable is created again with the same name. The empty-fetch path and the 409 path share one helper, which reports a failed subscribe (jnats throws IllegalStateException for some, such as SUB-90007 when the stream is gone) through the exception handler and tries again at the next consumer check, instead of ending the fetch loop. The new consumer is sent without a start time (ByStartTime together with ByStartSequence), a subscription made while the route stops is unsubscribed, the unused DefaultNatsManualAck constructor is removed, and the upgrade guide entry covers the ephemeral consumer only. Co-Authored-By: Claude Opus 5.5 --- .../component/nats/DefaultNatsManualAck.java | 5 - .../camel/component/nats/NatsConsumer.java | 73 +++++-- .../NatsConsumerPullConsumerDeletedTest.java | 26 +++ .../NatsConsumerPullConsumerGoneTest.java | 186 ++++++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 15 +- 5 files changed, 278 insertions(+), 27 deletions(-) create mode 100644 components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java diff --git a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java index 1ab6b87928293..9b0859919817e 100644 --- a/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java +++ b/components/camel-nats/src/main/java/org/apache/camel/component/nats/DefaultNatsManualAck.java @@ -25,11 +25,6 @@ class DefaultNatsManualAck implements NatsManualAck { private final Message message; private final Runnable onAcknowledged; - DefaultNatsManualAck(Message message) { - this(message, () -> { - }); - } - DefaultNatsManualAck(Message message, Runnable onAcknowledged) { this.message = message; this.onAcknowledged = onAcknowledged; 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 b6c3e45a1b610..89f85c207f910 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 @@ -61,7 +61,7 @@ public class NatsConsumer extends DefaultConsumer { private Connection connection; private Dispatcher dispatcher; private boolean active; - private JetStreamSubscription jetStreamSubscription; + private volatile JetStreamSubscription jetStreamSubscription; public NatsConsumer(NatsEndpoint endpoint, Processor processor) { super(endpoint, processor); @@ -156,6 +156,8 @@ class NatsConsumingTask implements Runnable { private final NavigableSet unacknowledged = new ConcurrentSkipListSet<>(); private final AtomicLong highestFetched = new AtomicLong(); private long lastConsumerCheck = System.nanoTime(); + // whether the WARN that the durable consumer is gone was logged, so that it is logged once until it is back + private boolean durableGoneReported; NatsConsumingTask(Connection connection, NatsConfiguration configuration) { this.connection = connection; @@ -282,9 +284,12 @@ private void runPullFetchLoop( try { List messages = sub.fetch(batchSize, fetchTimeout); if (messages.isEmpty() && consumerCheckDue() && consumerGone(sub)) { - sub = subscribeAgain(sub, cc); + sub = consumerDeleted(sub, cc); continue; } + if (!messages.isEmpty()) { + durableGoneReported = false; + } for (Message msg : messages) { if (!NatsConsumer.this.isRunAllowed()) { break; @@ -298,18 +303,12 @@ private void runPullFetchLoop( } } } catch (JetStreamStatusException e) { - // Synchronous pull calls may throw JetStreamStatusException for statuses such as 409 - // (consumer deleted/push-based). Log and continue so the consumer can recover when the consumer - // state is restored. + // Synchronous pull calls may throw JetStreamStatusException for statuses such as 409. Report it and + // continue; for 409 Consumer Deleted an ephemeral consumer is created again (see consumerDeleted). NatsConsumer.this.getExceptionHandler().handleException("JetStream status during pull fetch", e); Status status = e.getStatus(); if (status != null && Status.CONSUMER_DELETED.equals(status.getMessage())) { - try { - sub = subscribeAgain(sub, cc); - } catch (Exception ex) { - NatsConsumer.this.getExceptionHandler().handleException( - "Error subscribing again after the JetStream consumer was deleted", ex); - } + sub = consumerDeleted(sub, cc); } } catch (IllegalStateException e) { // Subscription was closed (e.g. consumer is stopping), exit the loop @@ -340,6 +339,7 @@ private boolean consumerCheckDue() { private boolean consumerGone(JetStreamSubscription sub) { try { sub.getConsumerInfo(); + durableGoneReported = false; return false; } catch (JetStreamApiException e) { return e.getErrorCode() == 404; @@ -351,19 +351,54 @@ private boolean consumerGone(JetStreamSubscription sub) { } /** - * Subscribes again after the server deleted the consumer. An ephemeral consumer is created again starting at - * the first message this consumer has not acknowledged, so that nothing is skipped and the stream is not - * replayed from its deliver policy; a durable consumer is created again from its configuration. + * The server no longer has the consumer of the pull subscription. An ephemeral consumer is created again (see + * {@link #subscribeAgain}); when that fails, the failure goes to the exception handler and the next consumer + * check tries again. A durable consumer is not created again: it is gone only when an administrator deleted it + * or after its inactive threshold, and creating it again from its configuration would replay the stream from + * its deliver policy. Its subscription is kept, so the fetches work again when the durable is created again + * with the same name. + * + * @return the subscription to fetch from + */ + private JetStreamSubscription consumerDeleted(JetStreamSubscription sub, ConsumerConfiguration cc) { + if (cc.getDurable() != null) { + if (!durableGoneReported) { + durableGoneReported = true; + LOG.warn("The durable JetStream consumer {} of the pull subscription on topic {} is gone; it is not" + + " created again, and the fetches resume when it is created again with the same name", + cc.getDurable(), configuration.getTopic()); + } + return sub; + } + try { + return subscribeAgain(sub, cc); + } catch (Exception e) { + NatsConsumer.this.getExceptionHandler().handleException( + "Error subscribing again after the JetStream consumer was deleted", e); + return sub; + } + } + + /** + * Subscribes again after the server deleted the ephemeral consumer, starting at the first message this consumer + * has not acknowledged, so that nothing is skipped and the stream is not replayed from its deliver policy. + *

+ * A sequence that is never acknowledged stays in the unacknowledged set until this resubscribe, for example a + * message nak'd until maxDeliver is used up, or with manualAck and ackPolicy=None when the route never calls + * ack. The new consumer then starts at that old sequence, and the route sees the messages after it again + * (at-least-once). */ private JetStreamSubscription subscribeAgain(JetStreamSubscription old, ConsumerConfiguration cc) throws IOException, JetStreamApiException { ConsumerConfiguration again = cc; long start = unacknowledged.isEmpty() ? highestFetched.get() + 1 : unacknowledged.first(); - boolean resume = cc.getDurable() == null && highestFetched.get() > 0; + boolean resume = highestFetched.get() > 0; if (resume) { + // a start time (DeliverPolicy.ByStartTime) together with a start sequence is rejected by the server again = ConsumerConfiguration.builder(cc) .deliverPolicy(DeliverPolicy.ByStartSequence) .startSequence(start) + .startTime(null) .build(); } LOG.warn("The JetStream consumer of the pull subscription on topic {} is gone; subscribing again{}", @@ -372,6 +407,14 @@ private JetStreamSubscription subscribeAgain(JetStreamSubscription old, Consumer PullSubscribeOptions.builder().configuration(again).build()); unacknowledged.clear(); NatsConsumer.this.jetStreamSubscription = sub; + if (!NatsConsumer.this.isRunAllowed()) { + // stopping: doStop may have unsubscribed the old subscription before this one was assigned + try { + sub.unsubscribe(); + } catch (Exception e) { + LOG.debug("Error unsubscribing the new pull subscription while stopping", e); + } + } try { old.unsubscribe(); } catch (Exception e) { diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java index 00e071cdcdf3c..777125a868b80 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -111,7 +111,9 @@ void consumesAgainAfterAWaitingPullGotConsumerDeleted() throws Exception { static final class FakeServer { final List stream = new ArrayList<>(); final List consumers = new ArrayList<>(); + final List subscribed = new ArrayList<>(); int consumersCreated; + RuntimeException failNextSubscribe; final class FakeConsumer { long next; @@ -137,6 +139,24 @@ synchronized void deleteConsumers(boolean waitingRequest) { notifyAll(); } + /** + * The deleted consumers are created again with the same name (a durable created again by an administrator). + */ + synchronized void restoreConsumers() { + for (FakeConsumer c : consumers) { + c.deleted = false; + } + notifyAll(); + } + + synchronized void failNextSubscribe(RuntimeException e) { + failNextSubscribe = e; + } + + synchronized List subscribed() { + return new ArrayList<>(subscribed); + } + Connection connection() { JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); JetStream js = proxy(JetStream.class, (name, args) -> { @@ -156,6 +176,12 @@ Connection connection() { } private synchronized JetStreamSubscription subscribe(ConsumerConfiguration cc) { + if (failNextSubscribe != null) { + RuntimeException e = failNextSubscribe; + failNextSubscribe = null; + throw e; + } + subscribed.add(cc); FakeConsumer c = new FakeConsumer(); c.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.ByStartSequence ? cc.getStartSequence() : 1; consumers.add(c); diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java new file mode 100644 index 0000000000000..3b541be17476c --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -0,0 +1,186 @@ +/* + * 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.time.ZonedDateTime; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; + +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.nats.NatsConsumerPullConsumerDeletedTest.FakeServer; +import org.apache.camel.spi.ExceptionHandler; +import org.apache.camel.test.junit6.CamelTestSupport; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; +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.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * What the JetStream pull consumer does when the server no longer has its consumer, besides creating an ephemeral + * consumer again (see {@link NatsConsumerPullConsumerDeletedTest}): a durable consumer is not created again, a failed + * subscribe is tried again, and a start time is not sent together with the start sequence. + */ +class NatsConsumerPullConsumerGoneTest extends CamelTestSupport { + + private final FakeServer server = new FakeServer(); + private final List handled = new CopyOnWriteArrayList<>(); + + @Override + protected RouteBuilder createRouteBuilder() { + context.getRegistry().bind("conn", server.connection()); + context.getRegistry().bind("eh", new ExceptionHandler() { + @Override + public void handleException(Throwable exception) { + handled.add(exception); + } + + @Override + public void handleException(String message, Throwable exception) { + handled.add(exception); + } + + @Override + public void handleException(String message, Exchange exchange, Throwable exception) { + handled.add(exception); + } + }); + return new RouteBuilder() { + @Override + public void configure() { + // each test adds its route + } + }; + } + + private void addRoute(String options) throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("nats:orders?connection=#conn&jetstreamEnabled=true&jetstreamName=ORDERS&pullFetchTimeout=100" + + "&exceptionHandler=#eh" + options) + .convertBodyTo(String.class) + .to("mock:result"); + } + }); + } + + @Test + void durableConsumerIsNotCreatedAgain() throws Exception { + List warnings = new CopyOnWriteArrayList<>(); + AbstractAppender appender = new AbstractAppender("CaptureWarn", null, null, true, Property.EMPTY_ARRAY) { + @Override + public void append(LogEvent event) { + if (event.getLevel() == Level.WARN + && event.getMessage().getFormattedMessage().contains("durable JetStream consumer")) { + warnings.add(event.getMessage().getFormattedMessage()); + } + } + }; + appender.start(); + Logger logger = (Logger) LogManager.getLogger(NatsConsumer.class); + logger.addAppender(appender); + try { + addRoute("&durableName=orders"); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + // an administrator deleted the durable: a waiting pull gets 409 Consumer Deleted, then the fetches get + // nothing and the consumer check (every 5 seconds) finds it gone + server.deleteConsumers(true); + await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, warnings.size())); + await().during(6, TimeUnit.SECONDS).atMost(8, TimeUnit.SECONDS).untilAsserted(() -> { + assertEquals(1, server.consumersCreated(), "the durable is not created again"); + assertEquals(1, warnings.size(), "the WARN is logged once"); + }); + + // the durable is created again with the same name: the same subscription fetches again + mock.reset(); + mock.expectedBodiesReceived("order-2"); + server.restoreConsumers(); + server.publish("order-2"); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + assertEquals(1, server.consumersCreated()); + } finally { + logger.removeAppender(appender); + appender.stop(); + } + } + + @Test + void failedSubscribeIsTriedAgainAtTheNextCheck() throws Exception { + addRoute(""); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + // the stream itself is gone for a while (for example a memory stream after a server restart): jnats reports + // that subscribe failure as an IllegalStateException + server.failNextSubscribe(new IllegalStateException("[SUB-90007] No matching streams for subject.")); + server.deleteConsumers(false); + + mock.reset(); + mock.expectedBodiesReceived("order-2"); + server.publish("order-2"); + mock.setResultWaitTime(30_000); + mock.assertIsSatisfied(); + assertEquals(2, server.consumersCreated(), "created again at the check after the failed one"); + assertTrue(handled.stream().anyMatch(IllegalStateException.class::isInstance), + "the failed subscribe goes to the exception handler"); + } + + @Test + void startTimeIsNotSentWithTheStartSequence() throws Exception { + context.getRegistry().bind("cc", ConsumerConfiguration.builder() + .deliverPolicy(DeliverPolicy.ByStartTime) + .startTime(ZonedDateTime.now().minusHours(1)) + .build()); + addRoute("&consumerConfiguration=#cc"); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + server.deleteConsumers(true); + + mock.reset(); + mock.expectedBodiesReceived("order-2"); + server.publish("order-2"); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + ConsumerConfiguration again = server.subscribed().get(1); + assertEquals(DeliverPolicy.ByStartSequence, again.getDeliverPolicy()); + assertEquals(2, again.getStartSequence()); + assertNull(again.getStartTime()); + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 23b8629d1f10e..77a9fb848112f 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -1472,16 +1472,17 @@ in full and rejected by the broker. A settlement request the broker rejects is now reported through the consumer's exception handler. -=== camel-nats - the JetStream pull consumer subscribes again when its consumer was deleted +=== camel-nats - the JetStream pull consumer subscribes again when its ephemeral consumer was deleted With `jetstreamEnabled=true` and a pull subscription (the default), the `nats` consumer kept fetching from a JetStream consumer that the server had deleted, and consumed nothing more until the route was restarted. It now subscribes again -when the server no longer has its consumer. An ephemeral consumer (no `durableName`), which the server deletes after -about 5 seconds without a pull request, for example while the client is disconnected, or loses on a restart, is created -again at the first message the route did not acknowledge. A durable consumer that was deleted, for example by an -administrator, is created again from its configuration, so it starts where its deliver policy says: with the default -`DeliverPolicy.All` at the start of the stream, and the route consumes the messages of the stream again. Set a -`consumerConfiguration` with another deliver policy to avoid that. +when the server no longer has its ephemeral consumer (no `durableName`, and no durable name in the +`consumerConfiguration`), which the server deletes after about 5 seconds without a pull request, for example while the +client is disconnected, or loses on a restart. The new consumer starts at +the first message the route did not acknowledge, so the route can see again the messages after that one. A durable +consumer is not created again: when it is gone (deleted by an administrator, or after its `inactiveThreshold`), the +consumer logs a WARN and keeps fetching, and the fetches work again when the durable is created again with the same +name. === camel-telegram - a failed send fails the exchange once From 939b4a5412c50be4299e3c854c48db915e097529 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 12:31:12 +0530 Subject: [PATCH 3/6] CAMEL-25815: camel-nats - address review: wait for the rest of the fetch timeout after an early empty fetch While the consumer of a pull subscription is gone, jnats ends each fetch at once (status 503, no responders), so the fetch loop sent pull requests back to back: up to the next consumer check for an ephemeral consumer, and for good for a durable that is not created again. After an empty fetch that ended before its timeout, the loop now waits for the rest of pullFetchTimeout; a stop interrupts the wait. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 18 ++++++++++++++++-- .../NatsConsumerPullConsumerDeletedTest.java | 6 ++++++ .../nats/NatsConsumerPullConsumerGoneTest.java | 7 +++++++ 3 files changed, 29 insertions(+), 2 deletions(-) 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 89f85c207f910..0091e899e3fbd 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 @@ -22,6 +22,7 @@ import java.util.NavigableSet; import java.util.concurrent.ConcurrentSkipListSet; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; import io.nats.client.Connection; @@ -282,14 +283,23 @@ private void runPullFetchLoop( Duration fetchTimeout = Duration.ofMillis(configuration.getPullFetchTimeout()); while (NatsConsumer.this.isRunAllowed() && !Thread.currentThread().isInterrupted()) { try { + long fetchStarted = System.nanoTime(); List messages = sub.fetch(batchSize, fetchTimeout); if (messages.isEmpty() && consumerCheckDue() && consumerGone(sub)) { sub = consumerDeleted(sub, cc); continue; } - if (!messages.isEmpty()) { - durableGoneReported = false; + if (messages.isEmpty()) { + // jnats ends a fetch at once when its pull request has no responder (status 503), for example + // while the consumer is gone: wait for the rest of the fetch timeout, so that the loop does + // not send pull requests back to back + long left = fetchTimeout.toNanos() - (System.nanoTime() - fetchStarted); + if (left > 0) { + TimeUnit.NANOSECONDS.sleep(left); + } + continue; } + durableGoneReported = false; for (Message msg : messages) { if (!NatsConsumer.this.isRunAllowed()) { break; @@ -314,6 +324,10 @@ private void runPullFetchLoop( // Subscription was closed (e.g. consumer is stopping), exit the loop LOG.debug("Pull subscription closed, stopping fetch loop", e); return; + } catch (InterruptedException e) { + // the consumer is stopping + Thread.currentThread().interrupt(); + return; } catch (Exception e) { NatsConsumer.this.getExceptionHandler().handleException("Error during JetStream pull fetch", e); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java index 777125a868b80..f0dd332c4d176 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -113,6 +113,7 @@ static final class FakeServer { final List consumers = new ArrayList<>(); final List subscribed = new ArrayList<>(); int consumersCreated; + int fetches; RuntimeException failNextSubscribe; final class FakeConsumer { @@ -126,6 +127,10 @@ synchronized int consumersCreated() { return consumersCreated; } + synchronized int fetches() { + return fetches; + } + synchronized void publish(String body) { stream.add(body.getBytes(StandardCharsets.UTF_8)); notifyAll(); @@ -204,6 +209,7 @@ private synchronized Object consumerInfo(FakeConsumer c) throws JetStreamApiExce } private synchronized List fetch(FakeConsumer c, JetStreamSubscription sub, int batch, Duration timeout) { + fetches++; long deadline = System.nanoTime() + timeout.toNanos(); List out = new ArrayList<>(); while (true) { diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java index 3b541be17476c..19750b7ede2f3 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -117,10 +117,17 @@ public void append(LogEvent event) { // nothing and the consumer check (every 5 seconds) finds it gone server.deleteConsumers(true); await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, warnings.size())); + int fetchesBefore = server.fetches(); + long waitStarted = System.nanoTime(); await().during(6, TimeUnit.SECONDS).atMost(8, TimeUnit.SECONDS).untilAsserted(() -> { assertEquals(1, server.consumersCreated(), "the durable is not created again"); assertEquals(1, warnings.size(), "the WARN is logged once"); }); + // while the durable is gone each fetch ends at once: the loop waits pullFetchTimeout (100 ms) between them + long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); + int fetches = server.fetches() - fetchesBefore; + assertTrue(fetches <= waitedMillis / 100 + 5, + "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); // the durable is created again with the same name: the same subscription fetches again mock.reset(); From f946daeb27ce40e41ce60d14b2ecf48386d6f5a9 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 16:31:41 +0530 Subject: [PATCH 4/6] CAMEL-25815: camel-nats - address review: treat a null consumer info as gone, resume only where the stream allows it jnats 2.26.4 answers getConsumerInfo() with null when the server has no such consumer (lookupConsumerInfo maps api error 10014 to null), so the consumer check never found a deleted consumer: it now treats null as gone, and the fake server answers like jnats (null for a deleted consumer, the info for a live one). subscribeAgain now looks the stream up first. A stream with WorkQueue retention accepts only the deliver policy All and has already removed the acknowledged messages, so the consumer configuration is used as it is. A stream whose create time differs from the one seen at the subscribe was created again (its sequences start at 1 again): the tracked sequences are cleared and the consumer configuration is used. The consumer check runs only after an empty fetch that ended before half of pullFetchTimeout (a gone consumer ends the fetch at once with 503), so an idle healthy consumer no longer asks for CONSUMER.INFO every 5 seconds. Other error statuses than 409 Consumer Deleted wait for the rest of the fetch timeout instead of fetching again back to back. With AckPolicy.None no sequence waits for an ack: a resubscribe starts after the highest fetched one. The tests wait with setResultWaitTime, and a JetStream IT deletes the consumer while the route is busy. Co-Authored-By: Claude Opus 5.5 --- .../camel/component/nats/NatsConsumer.java | 98 +++++++++--- .../NatsConsumerPullConsumerDeletedTest.java | 139 ++++++++++++++++-- .../NatsConsumerPullConsumerGoneTest.java | 57 ++++++- .../NatsJetstreamConsumerPullDeletedIT.java | 107 ++++++++++++++ .../pages/camel-4x-upgrade-guide-4_24.adoc | 12 +- 5 files changed, 373 insertions(+), 40 deletions(-) create mode 100644 components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java 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 0091e899e3fbd..da625166fbeb3 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 @@ -18,6 +18,7 @@ import java.io.IOException; import java.time.Duration; +import java.time.ZonedDateTime; import java.util.List; import java.util.NavigableSet; import java.util.concurrent.ConcurrentSkipListSet; @@ -38,7 +39,9 @@ import io.nats.client.api.AckPolicy; import io.nats.client.api.ConsumerConfiguration; import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.RetentionPolicy; import io.nats.client.api.StreamConfiguration; +import io.nats.client.api.StreamInfo; import io.nats.client.support.Status; import org.apache.camel.Exchange; import org.apache.camel.Processor; @@ -156,6 +159,11 @@ class NatsConsumingTask implements Runnable { // fetched: where a new ephemeral consumer must start when the server deleted the one this consumer used private final NavigableSet unacknowledged = new ConcurrentSkipListSet<>(); private final AtomicLong highestFetched = new AtomicLong(); + // with AckPolicy.None the delivery is the acknowledgement: no fetched sequence waits for an ack + private boolean trackUnacknowledged = true; + // when the stream was created, as seen at the subscribe: a different time later means that the stream was + // created again and its sequences started again at 1 + private ZonedDateTime streamCreated; private long lastConsumerCheck = System.nanoTime(); // whether the WARN that the durable consumer is gone was logged, so that it is logged once until it is back private boolean durableGoneReported; @@ -201,7 +209,8 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc this.configuration.getTopic()); JetStreamManagement jsm = connection.jetStreamManagement(); - if (streamExists(jsm, streamName)) { + StreamInfo streamInfo = lookupStream(jsm, streamName); + if (streamInfo != null) { LOG.debug("JetStream stream '{}' already exists. Skipping creation.", streamName); } else { LOG.debug("Creating JetStream {}", streamName); @@ -209,9 +218,10 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc .name(streamName) .subjects(topic) .build(); - jsm.addStream(streamConfig); + streamInfo = jsm.addStream(streamConfig); LOG.info("JetStream stream '{}' created successfully.", streamName); } + streamCreated = streamInfo != null ? streamInfo.getCreateTime() : null; ConsumerConfiguration cc = configuration.getConsumerConfiguration(); if (cc == null) { @@ -231,6 +241,7 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc } cc = ccBuilder.build(); } + trackUnacknowledged = cc.getAckPolicy() != AckPolicy.None; CamelNatsMessageHandler messageHandler = new CamelNatsMessageHandler(); @@ -282,21 +293,22 @@ private void runPullFetchLoop( int batchSize = configuration.getPullBatchSize(); Duration fetchTimeout = Duration.ofMillis(configuration.getPullFetchTimeout()); while (NatsConsumer.this.isRunAllowed() && !Thread.currentThread().isInterrupted()) { + long fetchStarted = System.nanoTime(); try { - long fetchStarted = System.nanoTime(); List messages = sub.fetch(batchSize, fetchTimeout); - if (messages.isEmpty() && consumerCheckDue() && consumerGone(sub)) { - sub = consumerDeleted(sub, cc); - continue; - } if (messages.isEmpty()) { // jnats ends a fetch at once when its pull request has no responder (status 503), for example - // while the consumer is gone: wait for the rest of the fetch timeout, so that the loop does - // not send pull requests back to back - long left = fetchTimeout.toNanos() - (System.nanoTime() - fetchStarted); - if (left > 0) { - TimeUnit.NANOSECONDS.sleep(left); + // while the consumer is gone, whereas an idle healthy consumer ends it at its expiry, shortly + // before the fetch timeout: only an empty fetch that ended early asks the server for the + // consumer, so that an idle healthy consumer does not send that request every few seconds + boolean endedEarly = System.nanoTime() - fetchStarted < fetchTimeout.toNanos() / 2; + if (endedEarly && consumerCheckDue() && consumerGone(sub)) { + sub = consumerDeleted(sub, cc); + continue; } + // wait for the rest of the fetch timeout, so that the loop does not send pull requests back to + // back + waitRestOfFetch(fetchStarted, fetchTimeout); continue; } durableGoneReported = false; @@ -319,6 +331,14 @@ private void runPullFetchLoop( Status status = e.getStatus(); if (status != null && Status.CONSUMER_DELETED.equals(status.getMessage())) { sub = consumerDeleted(sub, cc); + } else { + // for example 409 Consumer is push based: do not send pull requests back to back + try { + waitRestOfFetch(fetchStarted, fetchTimeout); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } } } catch (IllegalStateException e) { // Subscription was closed (e.g. consumer is stopping), exit the loop @@ -334,6 +354,13 @@ private void runPullFetchLoop( } } + private void waitRestOfFetch(long fetchStarted, Duration fetchTimeout) throws InterruptedException { + long left = fetchTimeout.toNanos() - (System.nanoTime() - fetchStarted); + if (left > 0) { + TimeUnit.NANOSECONDS.sleep(left); + } + } + /** * Whether the server no longer has the consumer of this pull subscription. The server deletes an ephemeral * consumer (no durable name) that saw no pull request for its inactive threshold (5 seconds by default), for @@ -352,7 +379,11 @@ private boolean consumerCheckDue() { private boolean consumerGone(JetStreamSubscription sub) { try { - sub.getConsumerInfo(); + // jnats answers null when the server has no such consumer (api error 10014), and throws for any other + // error, for example 404 stream not found (10059) + if (sub.getConsumerInfo() == null) { + return true; + } durableGoneReported = false; return false; } catch (JetStreamApiException e) { @@ -398,15 +429,33 @@ private JetStreamSubscription consumerDeleted(JetStreamSubscription sub, Consume * has not acknowledged, so that nothing is skipped and the stream is not replayed from its deliver policy. *

* A sequence that is never acknowledged stays in the unacknowledged set until this resubscribe, for example a - * message nak'd until maxDeliver is used up, or with manualAck and ackPolicy=None when the route never calls - * ack. The new consumer then starts at that old sequence, and the route sees the messages after it again - * (at-least-once). + * message nak'd until maxDeliver is used up. The new consumer then starts at that old sequence, and the route + * sees the messages after it again (at-least-once). With ackPolicy=None no sequence is tracked: the new + * consumer starts after the highest one fetched. + *

+ * The consumer configuration is used as it is (no start sequence) for a stream with WorkQueue retention, which + * accepts only the deliver policy All and has already removed the acknowledged messages, and for a stream that + * was created again since the subscribe (its sequences started again at 1). */ private JetStreamSubscription subscribeAgain(JetStreamSubscription old, ConsumerConfiguration cc) throws IOException, JetStreamApiException { + StreamInfo stream = this.connection.jetStreamManagement().getStreamInfo(configuration.getJetstreamName()); + ZonedDateTime created = stream.getCreateTime(); + if (created != null && streamCreated != null && !created.isEqual(streamCreated)) { + LOG.warn("The JetStream stream {} was created again since the subscribe; consuming it from its deliver" + + " policy", + configuration.getJetstreamName()); + unacknowledged.clear(); + highestFetched.set(0); + } + if (created != null) { + streamCreated = created; + } + boolean workQueue = stream.getConfiguration() != null + && stream.getConfiguration().getRetentionPolicy() == RetentionPolicy.WorkQueue; ConsumerConfiguration again = cc; long start = unacknowledged.isEmpty() ? highestFetched.get() + 1 : unacknowledged.first(); - boolean resume = highestFetched.get() > 0; + boolean resume = !workQueue && highestFetched.get() > 0; if (resume) { // a start time (DeliverPolicy.ByStartTime) together with a start sequence is rejected by the server again = ConsumerConfiguration.builder(cc) @@ -440,7 +489,9 @@ private JetStreamSubscription subscribeAgain(JetStreamSubscription old, Consumer private void track(Message msg) { if (msg.isJetStream()) { long seq = msg.metaData().streamSequence(); - unacknowledged.add(seq); + if (trackUnacknowledged) { + unacknowledged.add(seq); + } highestFetched.accumulateAndGet(seq, Math::max); } } @@ -567,13 +618,16 @@ public void onFailure(Exchange exchange) { } } - private boolean streamExists(JetStreamManagement jsm, String streamName) throws IOException, JetStreamApiException { + /** + * @return the stream, or null when the server does not have it + */ + private static StreamInfo lookupStream(JetStreamManagement jsm, String streamName) + throws IOException, JetStreamApiException { try { - jsm.getStreamInfo(streamName); - return true; + return jsm.getStreamInfo(streamName); } catch (JetStreamApiException jsae) { if (jsae.getErrorCode() == 404) { - return false; + return null; } throw jsae; } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java index f0dd332c4d176..0c887151b86f8 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -19,8 +19,11 @@ import java.lang.reflect.Proxy; import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.time.ZonedDateTime; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; +import java.util.Set; import java.util.concurrent.TimeUnit; import io.nats.client.Connection; @@ -31,12 +34,17 @@ import io.nats.client.JetStreamSubscription; import io.nats.client.Message; import io.nats.client.PullSubscribeOptions; +import io.nats.client.api.ApiResponse; import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.ConsumerInfo; import io.nats.client.api.DeliverPolicy; -import io.nats.client.api.Error; +import io.nats.client.api.RetentionPolicy; import io.nats.client.api.ServerInfo; +import io.nats.client.api.StreamInfo; import io.nats.client.impl.NatsJetStreamMetaData; import io.nats.client.impl.NatsMessage; +import io.nats.client.support.DateTimeUtils; +import io.nats.client.support.JsonParser; import io.nats.client.support.Status; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; @@ -83,7 +91,8 @@ void consumesAgainAfterTheServerDeletedTheEphemeralConsumer() throws Exception { mock.expectedBodiesReceived("order-2", "order-3"); server.publish("order-2"); server.publish("order-3"); - mock.assertIsSatisfied(20_000); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); } @Test @@ -99,22 +108,71 @@ void consumesAgainAfterAWaitingPullGotConsumerDeleted() throws Exception { mock.reset(); mock.expectedBodiesReceived("order-2"); server.publish("order-2"); - mock.assertIsSatisfied(20_000); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); assertEquals(2, server.consumersCreated(), "one consumer created again"); } + @Test + void consumesAStreamCreatedAgainFromItsStart() throws Exception { + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1", "order-2"); + server.publish("order-1"); + server.publish("order-2"); + mock.assertIsSatisfied(); + + // a memory stream after a server restart: the consumer is gone and order-3 is sequence 1 of the new stream + server.createStreamAgain(); + + mock.reset(); + mock.expectedBodiesReceived("order-3"); + server.publish("order-3"); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + ConsumerConfiguration again = server.subscribed().get(1); + assertEquals(DeliverPolicy.All, again.getDeliverPolicy(), "the consumer configuration, not a start sequence"); + } + + @Test + void workQueueStreamIsSubscribedAgainWithDeliverPolicyAll() throws Exception { + server.retention(RetentionPolicy.WorkQueue); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + server.deleteConsumers(false); + + mock.reset(); + mock.expectedBodiesReceived("order-2"); + server.publish("order-2"); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + assertEquals(2, server.consumersCreated(), "one consumer created again"); + assertEquals(DeliverPolicy.All, server.subscribed().get(1).getDeliverPolicy(), + "a work queue stream accepts only the deliver policy All"); + } + /** * A JetStream server with one stream, faked at the level of the jnats interfaces the consumer calls. A consumer - * delivers from its start sequence; once deleted, a fetch on it gets nothing (jnats ends the fetch on the 503 "no - * responders" status), and getConsumerInfo fails with 404. + * delivers from its start sequence; once deleted, a fetch on it gets nothing at once (jnats ends the fetch on the + * 503 "no responders" status), and getConsumerInfo answers null, as jnats does when the server answers consumer not + * found (10014). A live consumer answers its info. A stream with WorkQueue retention removes the acknowledged + * messages and refuses a consumer whose deliver policy is not All (400, 10101), and a stream created again starts + * its sequences at 1 again with a new create time. */ static final class FakeServer { final List stream = new ArrayList<>(); + final Set removed = new HashSet<>(); final List consumers = new ArrayList<>(); final List subscribed = new ArrayList<>(); + RetentionPolicy retention = RetentionPolicy.Limits; + ZonedDateTime created = DateTimeUtils.gmtNow().minusHours(1); int consumersCreated; int fetches; + int consumerInfoRequests; RuntimeException failNextSubscribe; + Status fetchStatus; final class FakeConsumer { long next; @@ -131,6 +189,32 @@ synchronized int fetches() { return fetches; } + synchronized int consumerInfoRequests() { + return consumerInfoRequests; + } + + synchronized void retention(RetentionPolicy retention) { + this.retention = retention; + } + + /** + * Every fetch fails with this status, for example 409 Consumer is push based. + */ + synchronized void failFetches(Status status) { + fetchStatus = status; + } + + /** + * The stream is created again, for example a memory stream after a server restart: its consumers are gone and + * its sequences start again at 1. + */ + synchronized void createStreamAgain() { + stream.clear(); + removed.clear(); + created = created.plusMinutes(1); + deleteConsumers(false); + } + synchronized void publish(String body) { stream.add(body.getBytes(StandardCharsets.UTF_8)); notifyAll(); @@ -163,7 +247,10 @@ synchronized List subscribed() { } Connection connection() { - JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> null); + JetStreamManagement jsm = proxy(JetStreamManagement.class, (name, args) -> switch (name) { + case "getStreamInfo", "addStream" -> streamInfo(); + default -> null; + }); JetStream js = proxy(JetStream.class, (name, args) -> { if ("subscribe".equals(name) && args.length == 2 && args[1] instanceof PullSubscribeOptions o) { return subscribe(o.getConsumerConfiguration()); @@ -180,12 +267,28 @@ Connection connection() { }); } - private synchronized JetStreamSubscription subscribe(ConsumerConfiguration cc) { + private synchronized StreamInfo streamInfo() { + return new StreamInfo( + JsonParser.parseUnchecked("{\"config\":{\"name\":\"ORDERS\",\"subjects\":[\"orders\"]," + + "\"retention\":\"" + retention + "\"},\"created\":\"" + + DateTimeUtils.toRfc3339(created) + "\",\"state\":{}}")); + } + + private synchronized JetStreamSubscription subscribe(ConsumerConfiguration cc) throws JetStreamApiException { if (failNextSubscribe != null) { RuntimeException e = failNextSubscribe; failNextSubscribe = null; throw e; } + if (retention == RetentionPolicy.WorkQueue && cc.getDeliverPolicy() != DeliverPolicy.All) { + // the server's answer as jnats reads it (ApiResponse is abstract) + throw new JetStreamApiException( + new ApiResponse( + JsonParser.parseUnchecked( + "{\"error\":{\"code\":400,\"err_code\":10101," + + "\"description\":\"consumer must be deliver all on workqueue stream\"}}")) { + }); + } subscribed.add(cc); FakeConsumer c = new FakeConsumer(); c.next = cc != null && cc.getDeliverPolicy() == DeliverPolicy.ByStartSequence ? cc.getStartSequence() : 1; @@ -201,17 +304,21 @@ private synchronized JetStreamSubscription subscribe(ConsumerConfiguration cc) { return self[0]; } - private synchronized Object consumerInfo(FakeConsumer c) throws JetStreamApiException { + private synchronized ConsumerInfo consumerInfo(FakeConsumer c) { + consumerInfoRequests++; if (c.deleted) { - throw new JetStreamApiException(Error.convert(new Status(404, "consumer not found"))); + return null; } - return null; + return new ConsumerInfo(JsonParser.parseUnchecked("{\"stream_name\":\"ORDERS\",\"name\":\"c\",\"config\":{}}")); } private synchronized List fetch(FakeConsumer c, JetStreamSubscription sub, int batch, Duration timeout) { fetches++; long deadline = System.nanoTime() + timeout.toNanos(); List out = new ArrayList<>(); + if (fetchStatus != null) { + throw new JetStreamStatusException(fetchStatus, sub); + } while (true) { if (c.deleted) { if (c.tellDeleted) { @@ -222,7 +329,9 @@ private synchronized List fetch(FakeConsumer c, JetStreamSubscription s } while (out.size() < batch && (!c.redeliver.isEmpty() || c.next <= stream.size())) { long seq = c.redeliver.isEmpty() ? c.next++ : c.redeliver.remove(0); - out.add(message(c, sub, seq)); + if (!removed.contains(seq)) { + out.add(message(c, sub, seq)); + } } long left = deadline - System.nanoTime(); if (!out.isEmpty() || left <= 0) { @@ -262,6 +371,14 @@ public boolean isJetStream() { } yield null; } + case "ack", "term" -> { + synchronized (this) { + if (retention == RetentionPolicy.WorkQueue) { + removed.add(seq); + } + } + yield null; + } default -> null; }); } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java index 19750b7ede2f3..4542e485e51f5 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -21,8 +21,10 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.TimeUnit; +import io.nats.client.api.AckPolicy; import io.nats.client.api.ConsumerConfiguration; import io.nats.client.api.DeliverPolicy; +import io.nats.client.support.Status; import org.apache.camel.Exchange; import org.apache.camel.builder.RouteBuilder; import org.apache.camel.component.mock.MockEndpoint; @@ -151,8 +153,8 @@ void failedSubscribeIsTriedAgainAtTheNextCheck() throws Exception { server.publish("order-1"); mock.assertIsSatisfied(); - // the stream itself is gone for a while (for example a memory stream after a server restart): jnats reports - // that subscribe failure as an IllegalStateException + // the subscribe fails, for example while the stream itself is gone: jnats reports that failure as an + // IllegalStateException server.failNextSubscribe(new IllegalStateException("[SUB-90007] No matching streams for subject.")); server.deleteConsumers(false); @@ -166,6 +168,57 @@ void failedSubscribeIsTriedAgainAtTheNextCheck() throws Exception { "the failed subscribe goes to the exception handler"); } + @Test + void idleHealthyConsumerIsNotAskedForItsInfo() throws Exception { + addRoute(""); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1"); + server.publish("order-1"); + mock.assertIsSatisfied(); + + // each empty fetch of a healthy consumer lasts the whole fetch timeout: no consumer check after it + await().during(6, TimeUnit.SECONDS).atMost(8, TimeUnit.SECONDS) + .untilAsserted(() -> assertEquals(0, server.consumerInfoRequests())); + } + + @Test + void otherErrorStatusIsNotFetchedAgainBackToBack() throws Exception { + server.failFetches(new Status(409, "Consumer is push based")); + addRoute(""); + await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty()); + int fetchesBefore = server.fetches(); + long waitStarted = System.nanoTime(); + Thread.sleep(1000); + long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); + int fetches = server.fetches() - fetchesBefore; + // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch + assertTrue(fetches <= waitedMillis / 100 + 5, + "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + } + + @Test + void ackPolicyNoneResumesAfterTheHighestFetched() throws Exception { + // the route never calls ack: with ackPolicy=None the delivery is the acknowledgement + addRoute("&manualAck=true&ackPolicy=None"); + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedBodiesReceived("order-1", "order-2"); + server.publish("order-1"); + server.publish("order-2"); + mock.assertIsSatisfied(); + + server.deleteConsumers(false); + + mock.reset(); + mock.expectedBodiesReceived("order-3"); + server.publish("order-3"); + mock.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + ConsumerConfiguration again = server.subscribed().get(1); + assertEquals(AckPolicy.None, again.getAckPolicy()); + assertEquals(DeliverPolicy.ByStartSequence, again.getDeliverPolicy()); + assertEquals(3, again.getStartSequence(), "starts after the highest fetched sequence"); + } + @Test void startTimeIsNotSentWithTheStartSequence() throws Exception { context.getRegistry().bind("cc", ConsumerConfiguration.builder() diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java new file mode 100644 index 0000000000000..9d747783652ef --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/jetstream/NatsJetstreamConsumerPullDeletedIT.java @@ -0,0 +1,107 @@ +/* + * 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.jetstream; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import io.nats.client.Connection; +import io.nats.client.JetStreamManagement; +import io.nats.client.Nats; +import io.nats.client.Options; +import org.apache.camel.EndpointInject; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.component.nats.NatsComponent; +import org.apache.camel.component.nats.integration.NatsITSupport; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Isolated; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * The server deletes the ephemeral consumer of a pull subscription while the route is busy with a message, so that no + * pull request is waiting and no 409 Consumer Deleted arrives: the consumer must find out that its consumer is gone and + * subscribe again after the message it routed. + */ +@Isolated +class NatsJetstreamConsumerPullDeletedIT extends NatsITSupport { + + private static final String STREAM = "mystream-pull-deleted"; + + private final CountDownLatch busy = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @EndpointInject("mock:result") + MockEndpoint mockResultEndpoint; + + @Test + void consumesAgainAfterTheEphemeralConsumerWasDeleted() throws Exception { + waitForNatsConsumers(1); + mockResultEndpoint.expectedBodiesReceived("order-1", "order-2"); + + template.sendBody("direct:send", "order-1"); + assertTrue(busy.await(20, TimeUnit.SECONDS), "order-1 is routed"); + + String[] servers = Arrays.stream( + context.getComponent("nats", NatsComponent.class).getConfiguration().getServers().split(",")) + .map(server -> server.contains("://") ? server : "nats://" + server) + .toArray(String[]::new); + try (Connection connection = Nats.connect(new Options.Builder().servers(servers).build())) { + JetStreamManagement jsm = connection.jetStreamManagement(); + List consumers = jsm.getConsumerNames(STREAM); + assertEquals(1, consumers.size(), "the ephemeral consumer of the route"); + jsm.deleteConsumer(STREAM, consumers.get(0)); + } + release.countDown(); + + template.sendBody("direct:send", "order-2"); + mockResultEndpoint.setResultWaitTime(30_000); + mockResultEndpoint.assertIsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + String producerUri = "nats:mytopic-pull-deleted?jetstreamEnabled=true&jetstreamName=" + STREAM + + "&jetstreamAsync=false"; + String consumerUri = "nats:mytopic-pull-deleted?jetstreamEnabled=true&jetstreamName=" + STREAM + + "&pullSubscription=true&pullFetchTimeout=500"; + + from("direct:send") + // when running full test suite then send can fail due to nats server setup/teardown + .errorHandler(defaultErrorHandler().maximumRedeliveries(5)) + .to(producerUri); + + from(consumerUri) + .convertBodyTo(String.class) + .process(exchange -> { + if ("order-1".equals(exchange.getIn().getBody(String.class))) { + busy.countDown(); + release.await(30, TimeUnit.SECONDS); + } + }) + .to(mockResultEndpoint); + } + }; + } +} diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc index 77a9fb848112f..6de130d3f6e89 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc @@ -1478,11 +1478,13 @@ With `jetstreamEnabled=true` and a pull subscription (the default), the `nats` c consumer that the server had deleted, and consumed nothing more until the route was restarted. It now subscribes again when the server no longer has its ephemeral consumer (no `durableName`, and no durable name in the `consumerConfiguration`), which the server deletes after about 5 seconds without a pull request, for example while the -client is disconnected, or loses on a restart. The new consumer starts at -the first message the route did not acknowledge, so the route can see again the messages after that one. A durable -consumer is not created again: when it is gone (deleted by an administrator, or after its `inactiveThreshold`), the -consumer logs a WARN and keeps fetching, and the fetches work again when the durable is created again with the same -name. +client is disconnected, or loses on a restart. The new consumer starts at the first message the route did not +acknowledge, so the route can see again the messages after that one (with `ackPolicy=None`, after the last message +fetched). On a stream with `WorkQueue` retention, and on a stream that was created again since the route subscribed, the +new consumer uses the consumer configuration as it is. The consumer asks the server for its consumer only after a fetch +that ended empty well before `pullFetchTimeout`, at most every 5 seconds. A durable consumer is not created again: when +it is gone (deleted by an administrator, or after its `inactiveThreshold`), the consumer logs a WARN and keeps fetching, +and the fetches work again when the durable is created again with the same name. === camel-telegram - a failed send fails the exchange once From abb0d972fbeca4c9cbc5194d91cb036c87bb2abb Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:12:28 +0530 Subject: [PATCH 5/6] CAMEL-25815: camel-nats - address review: wait for the rest of the fetch timeout after an unexpected fetch failure too From gnodet-bot's review of #28204: the generic catch of runPullFetchLoop sent the next pull request at once after an unexpected exception (for example while the connection recovers), unlike the error statuses, which wait for the rest of pullFetchTimeout since the previous commit. It now waits as well. New test unexpectedFetchFailureIsNotFetchedAgainBackToBack: every fetch fails with a RuntimeException. Without the change the loop made 96635 fetches in one second. Co-Authored-By: Claude Opus 5.5 --- .../apache/camel/component/nats/NatsConsumer.java | 7 +++++++ .../nats/NatsConsumerPullConsumerDeletedTest.java | 11 +++++++++++ .../nats/NatsConsumerPullConsumerGoneTest.java | 15 +++++++++++++++ 3 files changed, 33 insertions(+) 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 da625166fbeb3..88026b0dbb2c2 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 @@ -350,6 +350,13 @@ private void runPullFetchLoop( return; } catch (Exception e) { NatsConsumer.this.getExceptionHandler().handleException("Error during JetStream pull fetch", e); + // as for the other error statuses: do not send the next pull request at once + try { + waitRestOfFetch(fetchStarted, fetchTimeout); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + return; + } } } } diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java index 0c887151b86f8..ce09e17c508b2 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -173,6 +173,7 @@ static final class FakeServer { int consumerInfoRequests; RuntimeException failNextSubscribe; Status fetchStatus; + RuntimeException fetchFailure; final class FakeConsumer { long next; @@ -204,6 +205,13 @@ synchronized void failFetches(Status status) { fetchStatus = status; } + /** + * Every fetch fails with this exception, for example an error of the connection while it recovers. + */ + synchronized void failFetches(RuntimeException failure) { + fetchFailure = failure; + } + /** * The stream is created again, for example a memory stream after a server restart: its consumers are gone and * its sequences start again at 1. @@ -319,6 +327,9 @@ private synchronized List fetch(FakeConsumer c, JetStreamSubscription s if (fetchStatus != null) { throw new JetStreamStatusException(fetchStatus, sub); } + if (fetchFailure != null) { + throw fetchFailure; + } while (true) { if (c.deleted) { if (c.tellDeleted) { diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java index 4542e485e51f5..8d88c05a19f0b 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -196,6 +196,21 @@ void otherErrorStatusIsNotFetchedAgainBackToBack() throws Exception { "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); } + @Test + void unexpectedFetchFailureIsNotFetchedAgainBackToBack() throws Exception { + server.failFetches(new RuntimeException("connection is recovering")); + addRoute(""); + await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty()); + int fetchesBefore = server.fetches(); + long waitStarted = System.nanoTime(); + Thread.sleep(1000); + long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); + int fetches = server.fetches() - fetchesBefore; + // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch + assertTrue(fetches <= waitedMillis / 100 + 5, + "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + } + @Test void ackPolicyNoneResumesAfterTheHighestFetched() throws Exception { // the route never calls ack: with ackPolicy=None the delivery is the acknowledgement From 72628e000a80747dbe336fda92ae5de10f5f5e63 Mon Sep 17 00:00:00 2001 From: smjain <49463903+allthingssecurity@users.noreply.github.com> Date: Wed, 21 Oct 2026 20:54:59 +0530 Subject: [PATCH 6/6] CAMEL-25815: camel-nats - address review: no Thread.sleep in the back-to-back fetch tests Both tests now check the bound with Awaitility during the whole second. Co-Authored-By: Claude Opus 5.5 --- .../NatsConsumerPullConsumerGoneTest.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java index 8d88c05a19f0b..72456842c1f3e 100644 --- a/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -188,12 +188,13 @@ void otherErrorStatusIsNotFetchedAgainBackToBack() throws Exception { await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty()); int fetchesBefore = server.fetches(); long waitStarted = System.nanoTime(); - Thread.sleep(1000); - long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); - int fetches = server.fetches() - fetchesBefore; // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch - assertTrue(fetches <= waitedMillis / 100 + 5, - "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); + int fetches = server.fetches() - fetchesBefore; + assertTrue(fetches <= waitedMillis / 100 + 5, + "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + }); } @Test @@ -203,12 +204,13 @@ void unexpectedFetchFailureIsNotFetchedAgainBackToBack() throws Exception { await().atMost(5, TimeUnit.SECONDS).until(() -> !handled.isEmpty()); int fetchesBefore = server.fetches(); long waitStarted = System.nanoTime(); - Thread.sleep(1000); - long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); - int fetches = server.fetches() - fetchesBefore; // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch - assertTrue(fetches <= waitedMillis / 100 + 5, - "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + await().during(1, TimeUnit.SECONDS).atMost(2, TimeUnit.SECONDS).untilAsserted(() -> { + long waitedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - waitStarted); + int fetches = server.fetches() - fetchesBefore; + assertTrue(fetches <= waitedMillis / 100 + 5, + "no pull requests back to back: " + fetches + " fetches in " + waitedMillis + " ms"); + }); } @Test