gnodet-bot commented on code in PR #28318: URL: https://github.com/apache/camel/pull/28318#discussion_r4358273545
########## components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java: ########## @@ -0,0 +1,384 @@ +/* + * 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.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; + +import io.nats.client.Connection; +import io.nats.client.ConnectionListener; +import io.nats.client.Dispatcher; +import io.nats.client.JetStreamApiException; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Message; +import io.nats.client.MessageHandler; +import io.nats.client.PushSubscribeOptions; +import io.nats.client.api.AckPolicy; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.RetentionPolicy; +import io.nats.client.api.StreamInfo; +import io.nats.client.impl.AckType; +import org.apache.camel.spi.ExceptionHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Keeps a JetStream push subscription consuming when the server deleted its consumer. + * <p> + * The server deletes an ephemeral push consumer (no durable name) when nobody is subscribed to its deliver subject 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. jnats subscribes the deliver subject again after the reconnect, but it does not create the + * consumer again, so the subscription would never receive anything again. After every reconnect this asks the server + * for the consumer and, when an ephemeral consumer is gone, subscribes again: the new consumer starts at the first + * message that was delivered to the route and not acknowledged (or after the last one delivered), so that nothing is + * skipped and the stream is not replayed from the deliver policy. 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. A WARN says so, and the subscription is kept. + * <p> + * jnats answers the consumer lookup with null (not an exception) when the server has no such consumer (api error + * 10014). On a stream with WorkQueue retention, which accepts only the deliver policy All and has already removed the + * acknowledged messages, and on a stream that was created again since the subscribe (its sequences started again at 1), + * the new consumer uses the consumer configuration as it is. + * <p> + * When the consumer cannot be checked or created again (JetStream not ready yet after a server restart, the stream is + * gone), this keeps trying while the route runs: a few times one second apart, then with a growing delay up to 30 + * seconds, with a WARN after the first attempts. + * <p> + * The consumer is checked only after a reconnect (when jnats has subscribed the deliver subject again). A consumer that + * is deleted while the connection stays up (for example by an administrator) is not detected, and the route receives + * nothing more: an idle heartbeat would detect it, but it is not allowed with a queue group, and its alarm goes to the + * error listener of the connection, which the route does not own when the connection is shared. + * <p> + * The delivered messages that are not settled yet are found with {@link Message#lastAck()} on each delivery. jnats + * writes that field on the thread that acknowledges, without synchronization, so a read here can miss a recent + * acknowledgement: the consumer then resumes at an earlier message and the server delivers it again, nothing is + * skipped. + */ +final class NatsJetStreamPushConsumerKeeper implements MessageHandler, ConnectionListener { + + private static final Logger LOG = LoggerFactory.getLogger(NatsJetStreamPushConsumerKeeper.class); + // the attempts one second apart, before the delay grows up to the maximum + private static final int CHECK_ATTEMPTS = 5; + private static final long CHECK_RETRY_DELAY_MILLIS = 1000; + private static final long MAX_CHECK_RETRY_DELAY_MILLIS = 30000; + + private final NatsConsumer consumer; + private final Connection connection; + private final String streamName; + private final String topic; + private final String queueName; + private final Dispatcher dispatcher; + private final MessageHandler handler; + private final ConsumerConfiguration configuration; + private final ExecutorService executor; + private final Consumer<JetStreamSubscription> onSubscribed; + private final Lock lock = new ReentrantLock(); + // whether a check is submitted and has not started yet, so that reconnects do not queue several checks + private final AtomicBoolean checkQueued = new AtomicBoolean(); + // the JetStream messages delivered to the route that may still need a delivery, by stream sequence + private final ConcurrentSkipListMap<Long, Message> unacknowledged = new ConcurrentSkipListMap<>(); + private final AtomicLong highestDelivered = new AtomicLong(); + // with AckPolicy.None the delivery is the acknowledgement: no delivered message waits for an ack + private final boolean trackUnacknowledged; + // the server delivers a message at most maxDeliver times (when set): a message delivered that often is settled + private final long maxDeliver; + private volatile JetStreamSubscription subscription; + // 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 volatile ZonedDateTime streamCreated; + // the last sequence of the stream at the subscribe (-1 if not known), where a consumer that starts at the new (or + // last) messages started + private volatile long streamLastSequence = -1; + // why the consumer counts as gone, for the log + private volatile String goneReason; + + NatsJetStreamPushConsumerKeeper(NatsConsumer consumer, Connection connection, String streamName, String topic, + String queueName, Dispatcher dispatcher, MessageHandler handler, + ConsumerConfiguration configuration, ExecutorService executor, + Consumer<JetStreamSubscription> onSubscribed) { + this.consumer = consumer; + this.connection = connection; + this.streamName = streamName; + this.topic = topic; + this.queueName = queueName; + this.dispatcher = dispatcher; + this.handler = handler; + this.configuration = configuration; + this.executor = executor; + this.onSubscribed = onSubscribed; + this.trackUnacknowledged = configuration.getAckPolicy() != AckPolicy.None; + this.maxDeliver = configuration.getMaxDeliver(); + } + + /** + * Starts watching the connection for reconnects, once the first subscription is made. + * + * @param streamCreated when the stream was created, as seen before the subscribe (null if not known) + * @param streamLastSequence the last sequence of the stream, as seen before the subscribe (-1 if not known) + */ + void start(JetStreamSubscription subscription, ZonedDateTime streamCreated, long streamLastSequence) { + this.subscription = subscription; + this.streamCreated = streamCreated; + this.streamLastSequence = streamLastSequence; + connection.addConnectionListener(this); + } + + void stop() { + connection.removeConnectionListener(this); + } + + @Override + public void onMessage(Message msg) throws InterruptedException { + if (msg.isJetStream()) { + long seq = msg.metaData().streamSequence(); + if (trackUnacknowledged) { + unacknowledged.values().removeIf(this::settled); Review Comment: 💡 **Performance note:** `unacknowledged.values().removeIf(this::settled)` performs an O(n) scan on every incoming JetStream message (when `trackUnacknowledged` is true). For high-throughput consumers this could become a hotspot as the unacknowledged set grows. Worth noting as a future optimization candidate (e.g. batch the cleanup or trigger it only when the set exceeds a threshold). ########## components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsJetStreamPushConsumerKeeper.java: ########## @@ -0,0 +1,384 @@ +/* + * 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.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; + +import io.nats.client.Connection; +import io.nats.client.ConnectionListener; +import io.nats.client.Dispatcher; +import io.nats.client.JetStreamApiException; +import io.nats.client.JetStreamSubscription; +import io.nats.client.Message; +import io.nats.client.MessageHandler; +import io.nats.client.PushSubscribeOptions; +import io.nats.client.api.AckPolicy; +import io.nats.client.api.ConsumerConfiguration; +import io.nats.client.api.DeliverPolicy; +import io.nats.client.api.RetentionPolicy; +import io.nats.client.api.StreamInfo; +import io.nats.client.impl.AckType; +import org.apache.camel.spi.ExceptionHandler; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Keeps a JetStream push subscription consuming when the server deleted its consumer. + * <p> + * The server deletes an ephemeral push consumer (no durable name) when nobody is subscribed to its deliver subject 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. jnats subscribes the deliver subject again after the reconnect, but it does not create the + * consumer again, so the subscription would never receive anything again. After every reconnect this asks the server + * for the consumer and, when an ephemeral consumer is gone, subscribes again: the new consumer starts at the first + * message that was delivered to the route and not acknowledged (or after the last one delivered), so that nothing is + * skipped and the stream is not replayed from the deliver policy. 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. A WARN says so, and the subscription is kept. + * <p> + * jnats answers the consumer lookup with null (not an exception) when the server has no such consumer (api error + * 10014). On a stream with WorkQueue retention, which accepts only the deliver policy All and has already removed the + * acknowledged messages, and on a stream that was created again since the subscribe (its sequences started again at 1), + * the new consumer uses the consumer configuration as it is. + * <p> + * When the consumer cannot be checked or created again (JetStream not ready yet after a server restart, the stream is + * gone), this keeps trying while the route runs: a few times one second apart, then with a growing delay up to 30 + * seconds, with a WARN after the first attempts. + * <p> + * The consumer is checked only after a reconnect (when jnats has subscribed the deliver subject again). A consumer that + * is deleted while the connection stays up (for example by an administrator) is not detected, and the route receives + * nothing more: an idle heartbeat would detect it, but it is not allowed with a queue group, and its alarm goes to the + * error listener of the connection, which the route does not own when the connection is shared. + * <p> + * The delivered messages that are not settled yet are found with {@link Message#lastAck()} on each delivery. jnats + * writes that field on the thread that acknowledges, without synchronization, so a read here can miss a recent + * acknowledgement: the consumer then resumes at an earlier message and the server delivers it again, nothing is + * skipped. + */ +final class NatsJetStreamPushConsumerKeeper implements MessageHandler, ConnectionListener { + + private static final Logger LOG = LoggerFactory.getLogger(NatsJetStreamPushConsumerKeeper.class); + // the attempts one second apart, before the delay grows up to the maximum + private static final int CHECK_ATTEMPTS = 5; + private static final long CHECK_RETRY_DELAY_MILLIS = 1000; + private static final long MAX_CHECK_RETRY_DELAY_MILLIS = 30000; + + private final NatsConsumer consumer; + private final Connection connection; + private final String streamName; + private final String topic; + private final String queueName; + private final Dispatcher dispatcher; + private final MessageHandler handler; + private final ConsumerConfiguration configuration; + private final ExecutorService executor; + private final Consumer<JetStreamSubscription> onSubscribed; + private final Lock lock = new ReentrantLock(); + // whether a check is submitted and has not started yet, so that reconnects do not queue several checks + private final AtomicBoolean checkQueued = new AtomicBoolean(); + // the JetStream messages delivered to the route that may still need a delivery, by stream sequence + private final ConcurrentSkipListMap<Long, Message> unacknowledged = new ConcurrentSkipListMap<>(); + private final AtomicLong highestDelivered = new AtomicLong(); + // with AckPolicy.None the delivery is the acknowledgement: no delivered message waits for an ack + private final boolean trackUnacknowledged; + // the server delivers a message at most maxDeliver times (when set): a message delivered that often is settled + private final long maxDeliver; + private volatile JetStreamSubscription subscription; + // 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 volatile ZonedDateTime streamCreated; + // the last sequence of the stream at the subscribe (-1 if not known), where a consumer that starts at the new (or + // last) messages started + private volatile long streamLastSequence = -1; + // why the consumer counts as gone, for the log + private volatile String goneReason; + + NatsJetStreamPushConsumerKeeper(NatsConsumer consumer, Connection connection, String streamName, String topic, + String queueName, Dispatcher dispatcher, MessageHandler handler, + ConsumerConfiguration configuration, ExecutorService executor, + Consumer<JetStreamSubscription> onSubscribed) { + this.consumer = consumer; + this.connection = connection; + this.streamName = streamName; + this.topic = topic; + this.queueName = queueName; + this.dispatcher = dispatcher; + this.handler = handler; + this.configuration = configuration; + this.executor = executor; + this.onSubscribed = onSubscribed; + this.trackUnacknowledged = configuration.getAckPolicy() != AckPolicy.None; + this.maxDeliver = configuration.getMaxDeliver(); + } + + /** + * Starts watching the connection for reconnects, once the first subscription is made. + * + * @param streamCreated when the stream was created, as seen before the subscribe (null if not known) + * @param streamLastSequence the last sequence of the stream, as seen before the subscribe (-1 if not known) + */ + void start(JetStreamSubscription subscription, ZonedDateTime streamCreated, long streamLastSequence) { + this.subscription = subscription; + this.streamCreated = streamCreated; + this.streamLastSequence = streamLastSequence; + connection.addConnectionListener(this); + } + + void stop() { + connection.removeConnectionListener(this); + } + + @Override + public void onMessage(Message msg) throws InterruptedException { + if (msg.isJetStream()) { + long seq = msg.metaData().streamSequence(); + if (trackUnacknowledged) { + unacknowledged.values().removeIf(this::settled); + unacknowledged.put(seq, msg); + } + highestDelivered.accumulateAndGet(seq, Math::max); + } + handler.onMessage(msg); + } + + @Override + public void connectionEvent(Connection conn, Events type) { + // jnats has subscribed the deliver subject again: is the consumer still there? + if (type == Events.RESUBSCRIBED && consumer.isRunAllowed() && checkQueued.compareAndSet(false, true)) { + try { + executor.submit(this::checkConsumer); + } catch (Exception e) { + checkQueued.set(false); + LOG.debug("Cannot check the JetStream consumer of the push subscription on topic {}", topic, e); + } + } + } + + void checkConsumer() { + lock.lock(); Review Comment: ⚠️ **Concurrency:** `lock.lock()` is not interruptible — if a thread blocks here while the executor calls `shutdownNow()`, the interrupt will be ignored and shutdown will stall until the lock is eventually acquired. Consider `lock.lockInterruptibly()` so shutdown can propagate the interrupt cleanly through `checkConsumer()`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
