allthingssecurity commented on code in PR #28318: URL: https://github.com/apache/camel/pull/28318#discussion_r4358482209
########## 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: Applied in b1fbb1a7e815. The check runs only on the executor of the consumer (`connectionEvent` on `RESUBSCRIBED` submits it), a fixed pool (`poolSize`, 10 by default), so a check queued by a second reconnect can wait for the lock while the first one holds it between attempts (up to 30 seconds). `doStop` calls `shutdownNow` on that executor, which does not wait for the threads, so the stop itself was not delayed; but with `lock()` the waiting thread stayed until the holder (which leaves on the interrupt in `Thread.sleep`) released the lock. With `lockInterruptibly()` it leaves at once: it restores the interrupt flag, clears `checkQueued` and returns without a check. No new test: since the stop does not wait for the executor, the only difference is how long a pool thread of a stopping consumer lingers, and no route behaviour shows it. A test would need a fake consumer lookup that ignores interrupts, which jnats does not do. The camel-nats unit tests pass (18). _Claude Code on behalf of allthingssecurity_ ########## 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: Kept as is; here is why the scan stays small. Settled entries are removed on every delivery, before the new one is added, so the map holds only messages that were delivered and are not settled yet, keyed by stream sequence (a redelivery replaces its entry). By default the route processes a message on the dispatcher thread and acks it in `onComplete` before the next one is dispatched, so the scan sees one entry. The map grows only when acks are deferred (`manualAck`, or an asynchronous route). Even then the server stops delivering once `maxAckPending` messages wait for an ack. camel-nats does not set it, so the server default of 1000 applies. With the default `maxDeliver` (-1), each `settled()` call reads only `lastAck()`. So the worst case is about 1000 field reads per message, and only for routes that hold many messages unacknowledged. A cheaper variant (prune only the head of the sorted map) would let settled entries pile up behind one slow message, so I'd rather not trade that for this. _Claude Code on behalf of allthingssecurity_ -- 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]
