[
https://issues.apache.org/jira/browse/CAMEL-26815?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Federico Mariani updated CAMEL-26815:
-------------------------------------
Fix Version/s: 4.24.0
(was: 4.23.0)
> camel-kafka - allow a per-fetcher poll exception strategy
> ---------------------------------------------------------
>
> Key: CAMEL-26815
> URL: https://issues.apache.org/jira/browse/CAMEL-26815
> Project: Camel
> Issue Type: Improvement
> Components: camel-kafka
> Reporter: Andreas Mueller
> Priority: Major
> Labels: pull-request-available
> Fix For: 4.22.3, 4.24.0
>
>
> *Use case*
> A custom `PollExceptionStrategy` that wraps the built-in reconnect behavior
> to emit poll-monitoring events cannot be used when an endpoint runs multiple
> consumer threads (`consumersCount` \> 1).
> *Why "Improvement" and not "Bug"*
> The current behavior matches its documented contract. The stock component
> works at any `consumersCount`, and the single-instance
> `setPollExceptionStrategy` option serves a stateless strategy (such as the
> retry one) correctly. No defect is reproducible with the strategies that ship.
> What is missing is a capability: binding a custom strategy to its fetcher,
> which the built-in reconnect, stop, and bridge strategies need. This enhances
> an existing extension point, so it is an improvement rather than a bug.
> *Description*
> The consumer builds its `PollExceptionStrategy` per fetcher thread, in
> `KafkaFetchRecords`, through `KafkaErrorStrategies.strategies(this, ...)`.
> The built-in reconnect, stop, and bridge strategies are bound to that
> fetcher: each receives the calling `KafkaFetchRecords` and drives its
> reconnect through it. With `consumersCount` \> 1 each fetcher gets its own
> strategy, bound to its own fetcher.
> The only extension point, `KafkaComponent.setPollExceptionStrategy`, is a
> single component-scoped instance. `strategies()` returns it for every
> fetcher, without the fetcher, and never builds the per-fetcher default. So a
> custom strategy cannot drive the consumer, and the same instance is shared by
> all fetcher threads. A strategy that needs the fetcher, or that wants to
> extend the reconnect, stop, or bridge behavior, cannot be plugged in for
> `consumersCount` \> 1.
> *Proposed change*
> Add a `PollExceptionStrategyFactory`, following the existing
> `KafkaClientFactory` and `KafkaManualCommitFactory` pattern. The factory is
> invoked once per fetcher and receives the owning `KafkaFetchRecords`, the
> endpoint, and the consumer, so a custom strategy can be bound to its fetcher
> and can wrap a built-in one. The existing `setPollExceptionStrategy` option
> stays unchanged, so the change is backwards compatible.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
