ramu11 commented on PR #28730: URL: https://github.com/apache/camel/pull/28730#issuecomment-6582222936
> ## TL;DR > Thanks @ramu11, this is a big step forward. The `EXCEPTION_CAUGHT` blocker is fixed, and the reflection-based wireTap tracking is replaced by a startup check, which is much simpler. **Blockers: None.** I have 2 important findings and 1 suggestion. All three are in code that the latest commit ([599871d](https://github.com/apache/camel/commit/599871dfcf6e134a53d1c97bad8ca343eca5b70e)) adds, so please fix them in this PR. CI is green. > > Reviewed: [599871d](https://github.com/apache/camel/commit/599871dfcf6e134a53d1c97bad8ca343eca5b70e) and [2ce2479](https://github.com/apache/camel/commit/2ce2479dd3e48551394fa8e65d175247536e47d4), on top of [4adad62](https://github.com/apache/camel/commit/4adad620d7c44c8ab68d69d9bc54595662bd1eb1) (added commits, not a re-squash). > > ### 1. [Important] A wireTap copy can now make a filtered record replay > `KafkaRecordStreamingProcessor.java:257` now records an asynchronous hand-off on the **shared** `KafkaExactlyOnceMetadata`. A wireTap copy shares that object. So when a wireTap copy sends to `seda`/`vm`/`disruptor`/..., the hand-off is recorded for the source exchange too. This includes a wireTap target that the new startup check allows (one that cannot reach Kafka): > > ```java > from(eosSource) > .wireTap("seda:audit") // allowed: seda:audit never reaches Kafka > .filter(header("type").isEqualTo("order")) > .to(eosSink) > .end(); > from("seda:audit").to("log:audit"); > ``` > > For a record the filter drops, nothing claims the offsets, so `completeExactlyOnceProcessing` checks `hasAsynchronousHandoff` (`:201-208`). The answer depends on whether the tap thread has already sent to `seda:audit`. So the record is replayed some of the time, and each replay runs the tap again, which sends another audit message. The same happens when the wireTap target is a `direct:` route that hops through `seda:`. > > I suggested this shared marker in my last review, sorry: I did not see this side effect. A fix that keeps the marker for copies: only record a hand-off when the asynchronous endpoint can actually lead to Kafka. You can reuse `routesToKafka(context, uri, ...)`, and cache the result per endpoint key so it does not run on every send. As a bonus, this also lets `filter(...).to(eosSink).end().to("seda:audit")` commit filtered records, which today replay forever. Could we add an IT with filter + `wireTap("seda:...")` to a non-Kafka route? > > ### 2. [Important] A Kafka producer inside `onException` is no longer a DLQ send > `FAILURE_HANDLER_DELIVERY_PROPERTY` is only set when `ExchangeFailureHandlingEvent.isDeadLetterChannel()` is true (`KafkaRecordStreamingProcessor.java:264`, `:279`). In `RedeliveryErrorHandler` that flag is `processor == deadLetter` (`:712` on main). An `onException` block with outputs takes precedence over the DLC (`target = failureProcessor != null ? failureProcessor : deadLetter`, `:584` on main). So these are now treated as normal route sends: > > ```java > onException(ValidationException.class).handled(true) > .to("kafka:dlq?exactlyOnce=true&transactionalId=dlq"); // or a plain kafka:dlq > ``` > > * If the source-output producer has already started, a second exactly-once producer is rejected with "Only one exactlyOnce Kafka producer" (`KafkaProducer.java:616`). The record is replayed forever. > * A plain Kafka producer is rejected with "requires every Kafka destination producer to set exactlyOnce=true" (`:752-762`). It also replays forever. > > With the old `EXCEPTION_CAUGHT` check both cases worked. The doc still says "A Kafka producer invoked by an error handler as a dead-letter destination is a separate supported case" (`kafka-component.adoc:619-620`), which reads as if `onException` is included. Please choose one: > > * document that only `errorHandler(deadLetterChannel(...))` is a supported Kafka DLQ, and that an `onException` block must not call a second Kafka producer, or > * also accept the `FailureHandling`..`FailureHandled` window of a non-DLC failure processor when the failure is handled. The window covers only the failure processor itself, so sends after `doCatch` and after `continued(true)` are still classified correctly. > > Either way, please add an IT for `onException(...).handled(true).to(kafka DLQ)`. > > ### 3. [Suggestion] Gaps in the startup wireTap check > A Kafka-directed wireTap that the check misses still falls back to the runtime claim rejection. That fallback is racy: it only helps if the copy runs before the source commits. So it is worth closing these gaps: > > * `containsKafkaWireTap` (`:360`) only walks the source route. A wireTap in a sub-route that the source calls with `to("direct:x")` is not checked. `routeMayReachKafka` already follows `direct:` targets, so the source scan can do the same. > * `RecipientList`, `RoutingSlip` and `DynamicRouter` do not implement `EndpointSending`. So they are not rejected inside a wireTap target, although the comment at `:426` and the doc ("dynamic send EIPs inside a WireTap target are rejected conservatively") say they are. Since `camel-core-processor` is now a dependency, an `instanceof` check for them works. > * The check treats a `kamelet:` target (for example the `kafka-sink` Kamelet) and a `vm:` target with no consumer in this CamelContext as non-Kafka. Rejecting targets it cannot resolve would be safer, or the doc could list these cases. > > ### Nits > * `kafka-component.adoc:642-650`: "Event notification is enabled at CamelContext scope" appears twice, and the paragraph mixes the hand-off rule with the notifier cost. Could it be one short sentence on the cost? > * The `doCatch` IT calls `.process(e -> e.setException(null))` inside the catch block (`KafkaExactlyOnceIT`). `doCatch` already clears the exception, so this line can go. > > ### Earlier points > From my review on [8266d89](https://github.com/apache/camel/commit/8266d894e9e69b14333c41026e210bc1990c5201): > > * [x] 1. `EXCEPTION_CAUGHT` made sends after `doCatch` / `continued(true)` count as DLQ sends: fixed. Classification now uses the DLC failure events, and there are ITs for both cases (`kafkaSendAfterDoTryCatchSucceedsNormally`, `kafkaSendAfterContinuedExceptionSucceedsNormally`). See finding 2 for the `onException` side effect. > * [x] 2. Reflection-based wireTap instrumentation, static route map, forced `ASYNC_WAIT`, blocking wait: all removed and replaced by `validateNoKafkaWireTaps` at consumer start. See finding 3 for the gaps. > * [x] `Locale.ROOT` is now imported. The doc no longer names `TO_ENDPOINT`. > > From my review on [4adad62](https://github.com/apache/camel/commit/4adad620d7c44c8ab68d69d9bc54595662bd1eb1): > > * [x] Context-wide event notification is documented, and `notify` returns early for exchanges without exactly-once metadata. > * [x] The hand-off is recorded on the shared metadata (see finding 1 for the side effect). > * [x] The notifier is reference-counted and removed when the last exactly-once consumer stops, including when `doStart` fails part-way (`KafkaConsumer.java:176-194`, `:313-316`). > * [x] The doc lists the five detected schemes. > * [x] gnodet-bot's duplicate `hasSimpleFunction` nit is fixed in [2ce2479](https://github.com/apache/camel/commit/2ce2479dd3e48551394fa8e65d175247536e47d4). > > _Claude Code on behalf of davsclaus. This review was generated by an AI agent and may contain inaccuracies; please verify all suggestions before applying._ Thanks for the review. I've addressed the feedback, updated the implementation and documentation, and added regression coverage for the discussed cases. The changes are committed as `57b8327e4bbd`. -- 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]
