gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4435138302
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/consumer/support/streaming/KafkaRecordStreamingProcessor.java:
##########
@@ -104,6 +135,308 @@ public ProcessingResult processExchange(
return result;
}
+ /**
+ * A route that did not claim the source metadata with an exactly-once
producer still has to advance the source
+ * offset (for example, a filter branch). Run this after route completion
on the consumer polling thread.
+ */
+ private void completeExactlyOnceProcessing(Exchange exchange) {
+ if (!configuration.isExactlyOnce()) {
+ return;
+ }
+
+ KafkaExactlyOnceMetadata metadata = exchange.getProperty(
+ KafkaExactlyOnceMetadata.EXCHANGE_PROPERTY,
KafkaExactlyOnceMetadata.class);
+ if (metadata != null && exchange.getException() == null) {
+ Integer configuredMaxPollInterval =
configuration.getMaxPollIntervalMs();
+ long waitTimeoutMillis = configuredMaxPollInterval != null
+ ? Math.max(1L, configuredMaxPollInterval.longValue() -
1000L) : 299_000L;
Review Comment:
💡 **`awaitAsynchronousWireTaps` timeout leaves a 1-second margin before
Kafka rebalance.**
`maxPollIntervalMs - 1000` (default 299 s) gives only 1 second of slack
between the wait timing out and Kafka's `max.poll.interval.ms` firing a
rebalance. Under a loaded broker or a slow wire-tap route, a near-miss triggers
a replay (correct behaviour), but the margin is very thin. Consider a
proportional cushion or a separate configurable `exactlyOnce.wireTapTimeoutMs`
option so users with a high `max.poll.interval.ms` aren't constrained by the 1
s floor.
--
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]
