davsclaus commented on code in PR #25732:
URL: https://github.com/apache/camel/pull/25732#discussion_r3869536078
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -570,7 +570,7 @@ private boolean isReady() {
return false;
}
- boolean ready = true;
+ boolean ready = false;
Review Comment:
This default flip affects the existing, working classic-`KafkaConsumer`
reflection path too, not just the new `AsyncKafkaConsumer` branch below: any
future reflection failure (Kafka client renames `delegate`/`client` internals,
or a stricter JDK module system blocks the access) now flips a healthy classic
consumer to "not ready" instead of previously staying "ready". That may well be
the right call for a readiness probe (fail closed on uncertainty), but please
call it out explicitly as its own decision in the PR description, and confirm
the existing health-check IT suite still passes with this stricter default.
##########
components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaFetchRecords.java:
##########
@@ -585,12 +585,17 @@ private boolean isReady() {
"Health-Check calling
org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.hasReadyNode");
ready = nc.hasReadyNodes(System.currentTimeMillis());
}
+ } else {
+ // fail-closed: unknown consumer type (e.g. AsyncKafkaConsumer
with group.protocol=consumer) -> not ready
+ // alternative would be to reflectively check
AsyncKafkaConsumer.applicationEventHandler
+ ready = false;
Review Comment:
Rather than hardcoding `ready = false` here, `AsyncKafkaConsumer`'s actual
connectivity is reachable via reflection, mirroring the classic-consumer path
above:
```java
} else if (consumer.getClass().getName().equals(
"org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer")) {
// group.protocol=consumer (KIP-848) uses AsyncKafkaConsumer. Its
network client
// lives on a background thread, reachable via:
// applicationEventHandler -> networkThread -> networkClientDelegate ->
client
Object handler = ReflectionHelper.getField(
consumer.getClass().getDeclaredField("applicationEventHandler"),
consumer);
if (handler != null) {
Object networkThread = ReflectionHelper.getField(
handler.getClass().getDeclaredField("networkThread"),
handler);
if (networkThread != null) {
// networkClientDelegate is set once, on the consumer's
background thread,
// inside initializeResources(). It is not volatile, so there is
a brief
// window right after consumer startup where this reads null --
treat that
// as "not yet ready" rather than an error.
Object delegate = ReflectionHelper.getField(
networkThread.getClass().getDeclaredField("networkClientDelegate"),
networkThread);
if (delegate != null) {
Object client = ReflectionHelper.getField(
delegate.getClass().getDeclaredField("client"),
delegate);
if (client instanceof org.apache.kafka.clients.KafkaClient
kc) {
LOG.trace(
"Health-Check calling
org.apache.kafka.clients.KafkaClient.hasReadyNodes on AsyncKafkaConsumer's
NetworkClientDelegate");
ready = kc.hasReadyNodes(System.currentTimeMillis());
}
}
}
}
} else {
// fail-closed: genuinely unknown consumer type -> not ready
ready = false;
}
```
Verified reachable against `kafka-clients:4.3.1` (the version pinned in this
repo's `parent/pom.xml`) via `javap`/sources. Using
`getClass().getName().equals(...)` rather than `instanceof` against the
internals class literal avoids a hard compile-time dependency on a class with
no compatibility guarantee.
--
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]
