gnodet-bot commented on code in PR #28204:
URL: https://github.com/apache/camel/pull/28204#discussion_r4336093126
##########
components/camel-nats/src/main/java/org/apache/camel/component/nats/NatsConsumer.java:
##########
@@ -262,38 +287,221 @@ 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<Message> 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();
return;
}
}
} 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) {
Review Comment:
⚠️ **Minor gap:** Unlike the `JetStreamStatusException` catch above (which
now calls `waitRestOfFetch` for non-CONSUMER_DELETED statuses), this broad
fallback doesn't wait — a transient exception would send pull requests
back-to-back until the connection recovers. Consider adding:
```suggestion
} catch (Exception e) {
NatsConsumer.this.getExceptionHandler().handleException("Error during JetStream
pull fetch", e);
try {
waitRestOfFetch(fetchStarted, fetchTimeout);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
return;
}
}
```
--
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]
