andr-mueller opened a new pull request, #29618: URL: https://github.com/apache/camel/pull/29618
# 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 thus gets its own strategy, bound to its own fetcher, which is what lets those strategies act on the correct consumer. 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. This is fine for a stateless strategy such as the retry one, but a strategy that needs the fetcher, or that wants to extend the reconnect, stop, or bridge behaviour, cannot be plugged in for consumersCount > 1. This adds 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. strategies() now prefers the factory, then the existing single strategy, then the default factory. DefaultPollExceptionStrategyFactory holds the former pollOnError switch and exposes it through a protected method so a subclass can decorate the built-in strategy. The existing setPollExceptionStrategy option is unchanged and keeps its current semantics, so this is backwards compatible. The new option is autowired and generated the same way as the other component factories. KafkaErrorStrategiesTest covers the three precedence levels and that the factory is invoked once per fetcher, so each fetcher gets its own strategy instance. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [x] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. -- 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]
