gnodet-bot commented on code in PR #28730:
URL: https://github.com/apache/camel/pull/28730#discussion_r4435136515
##########
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;
+ try {
+ if (!metadata.awaitAsynchronousWireTaps(waitTimeoutMillis)) {
+ exchange.setException(new IllegalStateException(
+ "Timed out waiting for a potentially
Kafka-producing WireTap route to finish; "
+ + "the
source record must be replayed"));
+ return;
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ exchange.setException(new IllegalStateException(
+ "Interrupted while waiting for a potentially
Kafka-producing WireTap route to finish", e));
+ return;
+ }
+ }
+ if (metadata != null && metadata.isDeadLetterSendFailed()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "A Kafka dead-letter send failed; the source record
must be replayed"));
+ }
+ return;
+ }
+ if (metadata != null && metadata.isClaimRejected()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "The Kafka exactlyOnce source metadata was rejected by
another exchange or producer: "
+ +
metadata.getClaimRejectionReason()));
+ }
+ return;
+ }
+
+ boolean transactionAborted = Boolean.TRUE.equals(
+
exchange.getProperty(KafkaExactlyOnceMetadata.TRANSACTION_ABORTED_PROPERTY));
+ if (transactionAborted || exchange.isRollbackOnly()) {
+ if (exchange.getException() == null) {
+ exchange.setException(new IllegalStateException(
+ "The Kafka exactlyOnce transaction was rolled back;
the source record must be replayed"));
+ }
+ return;
+ }
+ if (exchange.getException() != null) {
+ return;
+ }
+
+ Throwable sendFailure = metadata != null ? metadata.getSendFailure() :
null;
+ if (sendFailure != null && !metadata.isClaimed()) {
+ // A producer failed before it could register a transaction
synchronization. Do not mistake a handled
+ // configuration/startup failure for a filter branch and commit
its source offset.
+ exchange.setException(new IllegalStateException(
+ "The exactlyOnce Kafka producer failed before it could
claim the source offsets", sendFailure));
+ return;
+ }
+ if (metadata == null) {
+ exchange.setException(new IllegalStateException(
+ "exactlyOnce source metadata was not captured for the
consumed record"));
+ return;
+ }
+ if (!metadata.isClaimed()) {
+ if (hasAsynchronousHandoff(exchange)) {
+ exchange.setException(new IllegalStateException(
+ "An exactlyOnce Kafka source cannot commit offsets
directly when the route hands the exchange "
+ + "to an
asynchronous endpoint before an exactly-once Kafka producer claims it; "
+ + "invoke the
destination synchronously from the source exchange"));
+ return;
+ }
+ try {
+ metadata.commitOffsets();
+ } catch (Exception commitFailure) {
+ exchange.setException(commitFailure);
+ }
+ }
+ }
+
+ /** Detect only hand-offs that this source exchange actually performed. */
+ boolean hasAsynchronousHandoff(Exchange exchange) {
+ String toEndpoint =
exchange.getProperty(ExchangePropertyKey.TO_ENDPOINT, String.class);
+ if (toEndpoint != null && endpointKey(toEndpoint).startsWith("seda:"))
{
Review Comment:
💡 **Coverage gap: `hasAsynchronousHandoff` only detects `seda:` as an async
hand-off endpoint.**
`endpointKey(toEndpoint).startsWith("seda:")` correctly catches the common
case, but `vm:`, `disruptor:`, and `reactive-streams:` endpoints are also
asynchronous and would pass the check undetected. A route like:
```java
from(eosSource).filter(predicate).to(eosSink).end().to("vm:audit");
```
would allow filtered records to call `commitOffsets()` while the `vm:`
thread is still being dispatched. This is lower risk than the old route-tree
scan, but worth documenting in a `@implNote` or `@see` which async endpoint
schemes are recognized, so future contributors know to extend the check when
adding a new transport.
--
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]
