lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3358152911
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -426,6 +431,11 @@ public class ConsumerConfig extends AbstractConfig {
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+ .define(BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
Review Comment:
should we validate "atLeast(0L)," (like in producer and admin) also they
don't have the same "importance", should we align?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##########
@@ -157,7 +158,12 @@ public void poll(final long timeoutMs, final long
currentTimeMs) {
* @param onClose True when the network thread is closing.
*/
public void poll(final long timeoutMs, final long currentTimeMs, boolean
onClose) {
- trySend(currentTimeMs);
+ try {
+ trySend(currentTimeMs);
+ } catch (BootstrapResolutionException e) {
+ // Bootstrap DNS resolution timeout - propagate to app thread
Review Comment:
this made me notice we don't seem to propagate it for the Producer? or am I
missing it?
We want producer.send/partitionsFor to throw the Bootstrap exception, but
it's the same challenge we solved here: the exception is on the network layer,
Sender thread for producer, we need to propagate to the producer thread.
I imagine the Sender.runOnce call to the client.poll needs to catch the
Bootstrap error and propagate it via the producer metadata fatalError? (so that
the Producer API calls throw it on "metadata.awaitUpdate")
##########
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##########
@@ -4197,6 +4200,46 @@ public void
testConstructorFailsOnNetworkClientConstructorFailure(GroupProtocol
assertEquals("No LoginModule found for
org.example.InvalidLoginModule", cause.getMessage());
}
+ @ParameterizedTest
+ @EnumSource(value = GroupProtocol.class)
+ public void
testConsumerBootstrapResolutionExceptionPropagatedToPoll(GroupProtocol
protocol) throws InterruptedException {
Review Comment:
could we add similar test for the producer.send/partitionsFor? (to be sure
that we're propagating the bootstrap exception, related to comment above). Same
for admin btw, can't find tests to ensure we do throw Bootstrap at the API
level and not swallow it along the way)
##########
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java:
##########
@@ -164,6 +166,11 @@ public boolean tryConnect() {
try (Metrics metrics = new Metrics()) {
try (Selector selector = new
Selector(conf.getLong(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
metrics, Time.SYSTEM, "", channelBuilder, logContext))
{
+ NetworkClient.BootstrapConfiguration
bootstrapConfiguration = NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
ditto
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##########
@@ -148,6 +149,17 @@ public void rebootstrap(long now) {
AdminMetadataManager.this.rebootstrap(now);
}
+ @Override
+ public boolean isBootstrapped() {
+ return bootstrapCluster != null;
+ }
+
+ @Override
+ public void bootstrap(List<InetSocketAddress> addresses) {
+ // Called by NetworkClient.ensureBootstrapped() after DNS
resolution succeeds
+ AdminMetadataManager.this.update(Cluster.bootstrap(addresses), 0);
Review Comment:
is this "0" here intentional? this is supposed to be "now"
##########
tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java:
##########
@@ -664,14 +665,19 @@ private static class ReplicaFetcherBlockingSend {
metrics,
time,
"replica-fetcher",
- new HashMap<String, String>() {{
+ new HashMap<>() {{
put("broker-id", sourceNode.idString());
put("fetcher-id", String.valueOf(fetcherId));
}},
false,
channelBuilder,
logContext
);
+ NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
do we need "enabled" here if this tool uses the ManualMetadataUpdater?
--
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]
