davsclaus commented on code in PR #28204:
URL: https://github.com/apache/camel/pull/28204#discussion_r4333295738
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -262,28 +273,44 @@ 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<Message> messages = sub.fetch(batchSize,
fetchTimeout);
+ if (messages.isEmpty() && consumerCheckDue() &&
consumerGone(sub)) {
+ sub = subscribeAgain(sub, cc);
Review Comment:
Not blocking: unlike the 409 path, a failure here is not caught. jnats
throws `IllegalStateException` for some subscribe errors (for example SUB-90007
`No matching streams for subject` when the stream is gone). The `catch
(IllegalStateException e)` below takes that as "subscription closed" and ends
the loop with a DEBUG log. A shared helper that reports the failure through the
exception handler would let the next check try again.
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -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;
Review Comment:
Blocking: when `cc.getDurable() != null` this still subscribes again with
the original configuration, so a durable that an administrator deleted (or that
hit its `inactiveThreshold`) is created again. With the default
`DeliverPolicy.All` the route replays the whole stream. I would only subscribe
again for an ephemeral consumer. For a durable, log a WARN and keep fetching:
the fetches work again if the durable is created again with the same name.
##########
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) {
Review Comment:
Optional: this constructor is no longer used, so it can be removed.
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -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)
Review Comment:
Optional: a user `consumerConfiguration` with `DeliverPolicy.ByStartTime`
keeps its start time in the copy, and I believe the server rejects a start time
together with `ByStartSequence`.
```suggestion
.deliverPolicy(DeliverPolicy.ByStartSequence)
.startTime(null)
```
--
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]
