atiaomar1978-hub commented on PR #25844: URL: https://github.com/apache/camel/pull/25844#issuecomment-5444654817
## Work summary for CAMEL-24473 _AI-generated comment on behalf of atiaomar1978-hub_ ### Issue [camel-kafka: consumer/producer readiness health check does not detect a downed broker under group.protocol=consumer (AsyncKafkaConsumer)](https://issues.apache.org/jira/browse/CAMEL-24473) With `group.protocol=consumer` (KIP-848), the readiness probe could not detect broker outages because reflection into `ClassicKafkaConsumer.client` failed silently on `AsyncKafkaConsumer`, leaving readiness at UP. ### Root cause `KafkaFetchRecords.isReady()` reflected `KafkaConsumer.delegate.client` expecting a `ConsumerNetworkClient`. `AsyncKafkaConsumer` has no such field; networking runs via `applicationEventHandler → networkThread → networkClientDelegate → client`. The `NoSuchFieldException` was caught and ignored (fail-open), so readiness never transitioned to DOWN. Verified against kafka-clients **4.3.1** (current parent POM version). ### Implementation **New: `KafkaNetworkHealthHelper`** - **Classic path:** `delegate.client` → `ConsumerNetworkClient.hasReadyNodes()` (unchanged behavior for `group.protocol=classic`) - **Async path:** `delegate.applicationEventHandler.networkThread.networkClientDelegate.client` → `KafkaClient.hasReadyNodes()` - **Fail-closed** when a real `KafkaConsumer` delegate cannot be resolved (per JIRA fail-safe guidance) - **Fail-open preserved** for custom `Consumer` / `Producer` implementations and producer reflection errors (no behavior change for those cases) **Updated:** - `KafkaFetchRecords.isReady()` → delegates to helper - `KafkaProducer.isReady()` → delegates to helper (refactor only; producer fail-open semantics unchanged) ### Tests added 1. **`KafkaNetworkHealthHelperTest`** (unit) - Fail-open for null/custom consumer and producer - Real `KafkaConsumer` instances for both `group.protocol=classic` and `group.protocol=consumer` — verifies reflection paths execute without exception - Real `KafkaProducer` layout probe 2. **`KafkaConsumerGroupProtocolHealthCheckIT`** (integration) - Readiness **UP** when broker healthy with `groupProtocol=consumer` - Readiness **DOWN** after broker shutdown - Gated on broker `group.version >= 1` (same assumption as `KafkaConsumerGroupProtocolIT`) Local verification: ```bash ./mvnw -pl components/camel-kafka test -Dtest=KafkaNetworkHealthHelperTest ``` All 6 unit tests pass. ### Automated review | Reviewer | Result | Action taken | |----------|--------|--------------| | **Bugbot** | No bugs found | — | | **Grok** | Several findings | Addressed in commit `af3776c5cdd` | Grok feedback addressed: - Restored fail-open for custom clients/producers (avoid breaking CAMEL-20592 behavior for non-standard clients) - Removed brittle `AsyncKafkaConsumer` simple-name check; probe fields directly - Added null guards on async reflection chain - Added real-client unit tests for classic + async layouts - Increased IT timeout to 60s - Removed unused `interceptorClasses` from new IT ### Branch / commits - Branch: `feature/CAMEL-24473-kafka-async-health-c587` - `f8233a3c239` — initial fix + tests - `af3776c5cdd` — review feedback (fail-open custom clients, null guards, real-client tests) ### Out of scope / not changed - No upgrade-guide entry (bug fix restores correct readiness behavior for async protocol; default `group.protocol=classic` unaffected) - Producer internals unchanged (only extracted to helper; fail-open on errors preserved) Ready for human review when CI is green. -- 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]
