gnodet commented on code in PR #25732:
URL: https://github.com/apache/camel/pull/25732#discussion_r3880976426
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -585,12 +589,41 @@ private boolean isReady() {
"Health-Check calling
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.hasReadyNode");
ready = nc.hasReadyNodes(System.currentTimeMillis());
}
+ } else if
(consumer.getClass().getName().equals(ASYNC_KAFKA_CONSUMER_CLASS_NAME)) {
Review Comment:
⚠️ **Dead code — this branch is unreachable.**
`DefaultKafkaClientFactory.getConsumer()` always returns `new
KafkaConsumer(kafkaProps)` (the public wrapper class), so `consumer instanceof
KafkaConsumer` on line 578 is always `true`, and this `else if` is never
entered.
In `kafka-clients:4.3.1`, `KafkaConsumer` holds a `private final
ConsumerDelegate delegate` field. The delegate is `ClassicKafkaConsumer`
(classic protocol) or `AsyncKafkaConsumer` (consumer protocol). The type check
must happen on the `delegate` object **inside** the `instanceof KafkaConsumer`
block, not at the outer `consumer` level.
The reflection chain here (`applicationEventHandler → networkThread →
networkClientDelegate → client`) is correct — it just needs to operate on the
delegate rather than the consumer.
--
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]
