KAFKA-12600: Remove deprecated config value default for client config client.dns.lookup

clients/src/main/java/org/apache/kafka/clients/ClientDnsLookup.java

@@ -19,8 +19,6 @@
 import java.util.Locale;
 
 public enum ClientDnsLookup {
-
-    DEFAULT("default"),
     USE_ALL_DNS_IPS("use_all_dns_ips"),
     RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY("resolve_canonical_bootstrap_servers_only");
 

clients/src/main/java/org/apache/kafka/clients/ClientUtils.java

@@ -32,7 +32,6 @@
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 
 import static org.apache.kafka.common.utils.Utils.getHost;
@@ -106,19 +105,9 @@ public static ChannelBuilder createChannelBuilder(AbstractConfig config, Time ti
                 clientSaslMechanism, time, true, logContext);
     }
 
-    static List<InetAddress> resolve(String host, ClientDnsLookup clientDnsLookup,
-                                     HostResolver hostResolver) throws UnknownHostException {
+    static List<InetAddress> resolve(String host, HostResolver hostResolver) throws UnknownHostException {
         InetAddress[] addresses = hostResolver.resolve(host);
-
-        switch (clientDnsLookup) {
-            case DEFAULT:
-                return Collections.singletonList(addresses[0]);
-            case USE_ALL_DNS_IPS:
-            case RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY:
-                return filterPreferredAddresses(addresses);
-        }
-
-        throw new IllegalStateException("Unhandled ClientDnsLookup instance: " + clientDnsLookup);
+        return filterPreferredAddresses(addresses);
     }
 
     /**

clients/src/main/java/org/apache/kafka/clients/ClusterConnectionStates.java

@@ -140,9 +140,8 @@ public boolean isPreparingConnection(String id) {
      * @param id the id of the connection
      * @param now the current time in ms
      * @param host the host of the connection, to be resolved internally if needed
-     * @param clientDnsLookup the mode of DNS lookup to use when resolving the {@code host}
      */
-    public void connecting(String id, long now, String host, ClientDnsLookup clientDnsLookup) {
+    public void connecting(String id, long now, String host) {
         NodeConnectionState connectionState = nodeState.get(id);
         if (connectionState != null && connectionState.host().equals(host)) {
             connectionState.lastConnectAttemptMs = now;
@@ -158,8 +157,7 @@ public void connecting(String id, long now, String host, ClientDnsLookup clientD
         // Create a new NodeConnectionState if nodeState does not already contain one
         // for the specified id or if the hostname associated with the node id changed.
         nodeState.put(id, new NodeConnectionState(ConnectionState.CONNECTING, now,
-                reconnectBackoff.backoff(0), connectionSetupTimeout.backoff(0), host,
-                clientDnsLookup, hostResolver));
+                reconnectBackoff.backoff(0), connectionSetupTimeout.backoff(0), host, hostResolver));
         connectingNodes.add(id);
     }
 
@@ -477,12 +475,10 @@ private static class NodeConnectionState {
         private List<InetAddress> addresses;
         private int addressIndex;
         private final String host;
-        private final ClientDnsLookup clientDnsLookup;
         private final HostResolver hostResolver;
 
         private NodeConnectionState(ConnectionState state, long lastConnectAttempt, long reconnectBackoffMs,
-                long connectionSetupTimeoutMs, String host, ClientDnsLookup clientDnsLookup,
-                HostResolver hostResolver) {
+                long connectionSetupTimeoutMs, String host, HostResolver hostResolver) {
             this.state = state;
             this.addresses = Collections.emptyList();
             this.addressIndex = -1;
@@ -493,7 +489,6 @@ private NodeConnectionState(ConnectionState state, long lastConnectAttempt, long
             this.connectionSetupTimeoutMs = connectionSetupTimeoutMs;
             this.throttleUntilTimeMs = 0;
             this.host = host;
-            this.clientDnsLookup = clientDnsLookup;
             this.hostResolver = hostResolver;
         }
 
@@ -509,7 +504,7 @@ public String host() {
         private InetAddress currentAddress() throws UnknownHostException {
             if (addresses.isEmpty()) {
                 // (Re-)initialize list
-                addresses = ClientUtils.resolve(host, clientDnsLookup, hostResolver);
+                addresses = ClientUtils.resolve(host, hostResolver);
                 addressIndex = 0;
             }
 

clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java

@@ -50,10 +50,7 @@ public class CommonClientConfigs {
                                                        + "(both the JVM and the OS cache DNS name lookups, however). "
                                                        + "If set to <code>resolve_canonical_bootstrap_servers_only</code>, "
                                                        + "resolve each bootstrap address into a list of canonical names. After "
-                                                       + "the bootstrap phase, this behaves the same as <code>use_all_dns_ips</code>. "
-                                                       + "If set to <code>default</code> (deprecated), attempt to connect to the "
-                                                       + "first IP address returned by the lookup, even if the lookup returns multiple "
-                                                       + "IP addresses.";
+                                                       + "the bootstrap phase, this behaves the same as <code>use_all_dns_ips</code>.";
 
     public static final String METADATA_MAX_AGE_CONFIG = "metadata.max.age.ms";
     public static final String METADATA_MAX_AGE_DOC = "The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions.";
@@ -185,13 +182,4 @@ public static Map<String, Object> postProcessReconnectBackoffConfigs(AbstractCon
         }
         return rval;
     }
-
-    public static void warnIfDeprecatedDnsLookupValue(AbstractConfig config) {
-        String clientDnsLookupValue = config.getString(CLIENT_DNS_LOOKUP_CONFIG);
-        if (clientDnsLookupValue.equals(ClientDnsLookup.DEFAULT.toString()))
-            log.warn("Configuration '{}' with value '{}' is deprecated and will be removed in " +
-                "future version. Please use '{}' or another non-deprecated value.",
-                CLIENT_DNS_LOOKUP_CONFIG, ClientDnsLookup.DEFAULT,
-                ClientDnsLookup.USE_ALL_DNS_IPS);
-    }
 }

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -111,8 +111,6 @@ private enum State {
     /* time in ms to wait before retrying to create connection to a server */
     private final long reconnectBackoffMs;
 
-    private final ClientDnsLookup clientDnsLookup;
-
     private final Time time;
 
     /**
@@ -141,7 +139,6 @@ public NetworkClient(Selectable selector,
                          int defaultRequestTimeoutMs,
                          long connectionSetupTimeoutMs,
                          long connectionSetupTimeoutMaxMs,
-                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -157,7 +154,6 @@ public NetworkClient(Selectable selector,
              defaultRequestTimeoutMs,
              connectionSetupTimeoutMs,
              connectionSetupTimeoutMaxMs,
-             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -176,7 +172,6 @@ public NetworkClient(Selectable selector,
                          int defaultRequestTimeoutMs,
                          long connectionSetupTimeoutMs,
                          long connectionSetupTimeoutMaxMs,
-                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -194,7 +189,6 @@ public NetworkClient(Selectable selector,
              defaultRequestTimeoutMs,
              connectionSetupTimeoutMs,
              connectionSetupTimeoutMaxMs,
-             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -214,7 +208,6 @@ public NetworkClient(Selectable selector,
                          int defaultRequestTimeoutMs,
                          long connectionSetupTimeoutMs,
                          long connectionSetupTimeoutMaxMs,
-                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -231,7 +224,6 @@ public NetworkClient(Selectable selector,
              defaultRequestTimeoutMs,
              connectionSetupTimeoutMs,
              connectionSetupTimeoutMaxMs,
-             clientDnsLookup,
              time,
              discoverBrokerVersions,
              apiVersions,
@@ -252,7 +244,6 @@ public NetworkClient(MetadataUpdater metadataUpdater,
                          int defaultRequestTimeoutMs,
                          long connectionSetupTimeoutMs,
                          long connectionSetupTimeoutMaxMs,
-                         ClientDnsLookup clientDnsLookup,
                          Time time,
                          boolean discoverBrokerVersions,
                          ApiVersions apiVersions,
@@ -287,7 +278,6 @@ public NetworkClient(MetadataUpdater metadataUpdater,
         this.apiVersions = apiVersions;
         this.throttleTimeSensor = throttleTimeSensor;
         this.log = logContext.logger(NetworkClient.class);
-        this.clientDnsLookup = clientDnsLookup;
         this.state = new AtomicReference<>(State.ACTIVE);
     }
 
@@ -981,7 +971,7 @@ private void handleInitiateApiVersionRequests(long now) {
     private void initiateConnect(Node node, long now) {
         String nodeConnectionId = node.idString();
         try {
-            connectionStates.connecting(nodeConnectionId, now, node.host(), clientDnsLookup);
+            connectionStates.connecting(nodeConnectionId, now, node.host());
             InetAddress address = connectionStates.currentAddress(nodeConnectionId);
             log.debug("Initiating connection to node {} using address {}", node, address);
             selector.connect(nodeConnectionId,

clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java

@@ -199,8 +199,7 @@ public class AdminClientConfig extends AbstractConfig {
                                 .define(CLIENT_DNS_LOOKUP_CONFIG,
                                         Type.STRING,
                                         ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
-                                        in(ClientDnsLookup.DEFAULT.toString(),
-                                           ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
+                                        in(ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
                                            ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
                                         Importance.MEDIUM,
                                         CLIENT_DNS_LOOKUP_DOC)
@@ -221,7 +220,6 @@ public class AdminClientConfig extends AbstractConfig {
 
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
-        CommonClientConfigs.warnIfDeprecatedDnsLookupValue(this);
         return CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
     }
 

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -18,7 +18,6 @@
 package org.apache.kafka.clients.admin;
 
 import org.apache.kafka.clients.ApiVersions;
-import org.apache.kafka.clients.ClientDnsLookup;
 import org.apache.kafka.clients.ClientRequest;
 import org.apache.kafka.clients.ClientResponse;
 import org.apache.kafka.clients.ClientUtils;
@@ -532,7 +531,6 @@ static KafkaAdminClient createInternal(AdminClientConfig config, TimeoutProcesso
                 (int) TimeUnit.HOURS.toMillis(1),
                 config.getLong(AdminClientConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),
                 config.getLong(AdminClientConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),
-                ClientDnsLookup.forConfig(config.getString(AdminClientConfig.CLIENT_DNS_LOOKUP_CONFIG)),
                 time,
                 true,
                 apiVersions,

clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java

@@ -322,8 +322,7 @@ public class ConsumerConfig extends AbstractConfig {
                                 .define(CLIENT_DNS_LOOKUP_CONFIG,
                                         Type.STRING,
                                         ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
-                                        in(ClientDnsLookup.DEFAULT.toString(),
-                                           ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
+                                        in(ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
                                            ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
                                         Importance.MEDIUM,
                                         CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
@@ -560,7 +559,6 @@ public class ConsumerConfig extends AbstractConfig {
 
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
-        CommonClientConfigs.warnIfDeprecatedDnsLookupValue(this);
         Map<String, Object> refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
         maybeOverrideClientId(refinedConfigs);
         return refinedConfigs;

clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java

@@ -17,7 +17,6 @@
 package org.apache.kafka.clients.consumer;
 
 import org.apache.kafka.clients.ApiVersions;
-import org.apache.kafka.clients.ClientDnsLookup;
 import org.apache.kafka.clients.ClientUtils;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.GroupRebalanceConfig;
@@ -755,7 +754,6 @@ public KafkaConsumer(Map<String, Object> configs,
                     config.getInt(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG),
                     config.getLong(ConsumerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),
                     config.getLong(ConsumerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),
-                    ClientDnsLookup.forConfig(config.getString(ConsumerConfig.CLIENT_DNS_LOOKUP_CONFIG)),
                     time,
                     true,
                     apiVersions,

clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java

@@ -17,7 +17,6 @@
 package org.apache.kafka.clients.producer;
 
 import org.apache.kafka.clients.ApiVersions;
-import org.apache.kafka.clients.ClientDnsLookup;
 import org.apache.kafka.clients.ClientUtils;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.KafkaClient;
@@ -461,7 +460,6 @@ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadat
                 requestTimeoutMs,
                 producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MS_CONFIG),
                 producerConfig.getLong(ProducerConfig.SOCKET_CONNECTION_SETUP_TIMEOUT_MAX_MS_CONFIG),
-                ClientDnsLookup.forConfig(producerConfig.getString(ProducerConfig.CLIENT_DNS_LOOKUP_CONFIG)),
                 time,
                 true,
                 apiVersions,

clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java

@@ -291,8 +291,7 @@ public class ProducerConfig extends AbstractConfig {
                                 .define(CLIENT_DNS_LOOKUP_CONFIG,
                                         Type.STRING,
                                         ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
-                                        in(ClientDnsLookup.DEFAULT.toString(),
-                                           ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
+                                        in(ClientDnsLookup.USE_ALL_DNS_IPS.toString(),
                                            ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
                                         Importance.MEDIUM,
                                         CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
@@ -434,7 +433,6 @@ public class ProducerConfig extends AbstractConfig {
 
     @Override
     protected Map<String, Object> postProcessParsedConfig(final Map<String, Object> parsedValues) {
-        CommonClientConfigs.warnIfDeprecatedDnsLookupValue(this);
         Map<String, Object> refinedConfigs = CommonClientConfigs.postProcessReconnectBackoffConfigs(this, parsedValues);
         maybeOverrideEnableIdempotence(refinedConfigs);
         maybeOverrideClientId(refinedConfigs);

clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java

@@ -103,25 +103,13 @@ public void testFilterPreferredAddresses() throws UnknownHostException {
     @Test
     public void testResolveUnknownHostException() {
         assertThrows(UnknownHostException.class,
-            () -> ClientUtils.resolve("some.invalid.hostname.foo.bar.local", ClientDnsLookup.USE_ALL_DNS_IPS, hostResolver));
+            () -> ClientUtils.resolve("some.invalid.hostname.foo.bar.local", hostResolver));
     }
 
     @Test
     public void testResolveDnsLookup() throws UnknownHostException {
         // Note that kafka.apache.org resolves to at least 2 IP addresses
-        assertEquals(1, ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.DEFAULT, hostResolver).size());
-    }
-
-    @Test
-    public void testResolveDnsLookupAllIps() throws UnknownHostException {
-        // Note that kafka.apache.org resolves to at least 2 IP addresses
-        assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.USE_ALL_DNS_IPS, hostResolver).size() > 1);
-    }
-
-    @Test
-    public void testResolveDnsLookupResolveCanonicalBootstrapServers() throws UnknownHostException {
-        // Note that kafka.apache.org resolves to at least 2 IP addresses
-        assertTrue(ClientUtils.resolve("kafka.apache.org", ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY, hostResolver).size() > 1);
+        assertTrue(ClientUtils.resolve("kafka.apache.org", hostResolver).size() > 1);
     }
 
     private List<InetSocketAddress> checkWithoutLookup(String... url) {