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..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 @@ -23,14 +23,17 @@ class DefaultNatsManualAck implements NatsManualAck { private final Message message; + private final Runnable onAcknowledged; - DefaultNatsManualAck(Message message) { + DefaultNatsManualAck(Message message, Runnable onAcknowledged) { this.message = message; + this.onAcknowledged = onAcknowledged; } @Override public void ack() { message.ack(); + onAcknowledged.run(); } @Override @@ -46,6 +49,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..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 @@ -18,11 +18,15 @@ 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; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +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 +38,11 @@ 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.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; import org.apache.camel.spi.HeaderFilterStrategy; @@ -49,13 +57,15 @@ 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; 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); @@ -126,7 +136,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 +155,18 @@ 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(); + // 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; NatsConsumingTask(Connection connection, NatsConfiguration configuration) { this.connection = connection; @@ -187,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); @@ -195,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) { @@ -217,6 +241,7 @@ private void setupJetStreamConsumer(String topic, String queueName) throws IOExc } cc = ccBuilder.build(); } + trackUnacknowledged = cc.getAckPolicy() != AckPolicy.None; CamelNatsMessageHandler messageHandler = new CamelNatsMessageHandler(); @@ -235,7 +260,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 +287,37 @@ 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()) { + long fetchStarted = System.nanoTime(); try { List messages = sub.fetch(batchSize, fetchTimeout); + 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, 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; for (Message msg : messages) { if (!NatsConsumer.this.isRunAllowed()) { break; } try { + track(msg); messageHandler.onMessage(msg); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -280,17 +325,187 @@ 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. 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())) { + 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 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); + // 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; + } + } + } + } + + 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 + * 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 { + // 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) { + 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; + } + } + + /** + * 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. 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 = !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) + .deliverPolicy(DeliverPolicy.ByStartSequence) + .startSequence(start) + .startTime(null) + .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; + 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) { + 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(); + if (trackUnacknowledged) { + unacknowledged.add(seq); + } + highestFetched.accumulateAndGet(seq, Math::max); + } + } + + private void acknowledged(Message msg) { + if (msg.isJetStream()) { + unacknowledged.remove(msg.metaData().streamSequence()); } } @@ -325,6 +540,7 @@ public void onMessage(Message msg) throws InterruptedException { public void onComplete(Exchange exchange) { LOG.debug("ACK"); msg.ack(); + acknowledged(msg); } @Override @@ -333,6 +549,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 +596,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); @@ -407,13 +625,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 new file mode 100644 index 0000000000000..ce09e17c508b2 --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerDeletedTest.java @@ -0,0 +1,421 @@ +/* + * 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.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; +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.ApiResponse; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.ConsumerInfo; +import io.nats.client.api.DeliverPolicy; +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; +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.setResultWaitTime(20_000); + mock.assertIsSatisfied(); + } + + @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.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 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; + RuntimeException fetchFailure; + + final class FakeConsumer { + long next; + boolean deleted; + boolean tellDeleted; + final List redeliver = new ArrayList<>(); + } + + synchronized int consumersCreated() { + return consumersCreated; + } + + 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; + } + + /** + * 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. + */ + 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(); + } + + synchronized void deleteConsumers(boolean waitingRequest) { + for (FakeConsumer c : consumers) { + c.deleted = true; + c.tellDeleted = 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) -> 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()); + } + 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 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; + 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 ConsumerInfo consumerInfo(FakeConsumer c) { + consumerInfoRequests++; + if (c.deleted) { + 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); + } + if (fetchFailure != null) { + throw fetchFailure; + } + 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); + if (!removed.contains(seq)) { + 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; + } + case "ack", "term" -> { + synchronized (this) { + if (retention == RetentionPolicy.WorkQueue) { + removed.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/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..72456842c1f3e --- /dev/null +++ b/components/camel-nats/src/test/java/org/apache/camel/component/nats/NatsConsumerPullConsumerGoneTest.java @@ -0,0 +1,263 @@ +/* + * 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.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; +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())); + 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(); + 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 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); + + 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 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(); + // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch + 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 + 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(); + // the loop waits for the rest of pullFetchTimeout (100 ms) after each failed fetch + 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 + 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() + .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/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 5b1e7e99e39d9..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 @@ -1472,6 +1472,20 @@ 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 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 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 (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 When Telegram answers a message with an error status (for example `400` for an unknown chat), or with a body that