Andreas Mueller created CAMEL-26815:
---------------------------------------
Summary: 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
Fix For: 4.22.3, 4.23.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)
