KAFKA-3112; Warn instead of error on unresolvable bootstrap server

so that unresolvable DNS names are ignored and only throw an error if no other bootstrap servers are resolvable. Author: Jonathan Bond <[email protected]> Reviewers: Ewen Cheslack-Postava <[email protected]>, Grant Henke <[email protected]>, Ismael Juma <[email protected]> Closes apache#792 from bondj/KAFKA-3112
priyank5485 · May 7, 2016 · c4bbf34 · c4bbf34
1 parent 717eea8
commit c4bbf34
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 6 deletions.
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
@@ -45,16 +45,18 @@ public static List<InetSocketAddress> parseAndValidateAddresses(List<String> url
                     throw new ConfigException("Invalid url in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
                 try {
                     InetSocketAddress address = new InetSocketAddress(host, port);
-                    if (address.isUnresolved())
-                        throw new ConfigException("DNS resolution failed for url in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
-                    addresses.add(address);
+                    if (address.isUnresolved()) {
+                        log.warn("Removing server from " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + " as DNS resolution failed: " + url);
+                    } else {
+                        addresses.add(address);
+                    }
                 } catch (NumberFormatException e) {
                     throw new ConfigException("Invalid port in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
                 }
             }
         }
         if (addresses.size() < 1)
-            throw new ConfigException("No bootstrap urls given in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+            throw new ConfigException("No resolvable bootstrap urls given in " + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
         return addresses;
     }
 

diff --git a/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java b/clients/src/test/java/org/apache/kafka/clients/ClientUtilsTest.java
@@ -18,8 +18,11 @@
 
 import org.apache.kafka.common.config.ConfigException;
 import org.junit.Test;
+import static org.junit.Assert.assertEquals;
 
+import java.net.InetSocketAddress;
 import java.util.Arrays;
+import java.util.List;
 
 public class ClientUtilsTest {
 
@@ -29,14 +32,24 @@ public void testParseAndValidateAddresses() {
         check("mydomain.com:8080");
         check("[::1]:8000");
         check("[2001:db8:85a3:8d3:1319:8a2e:370:7348]:1234", "mydomain.com:10000");
+        List<InetSocketAddress> validatedAddresses = check("some.invalid.hostname.foo.bar:9999", "mydomain.com:10000");
+        assertEquals(1, validatedAddresses.size());
+        InetSocketAddress onlyAddress = validatedAddresses.get(0);
+        assertEquals("mydomain.com", onlyAddress.getHostName());
+        assertEquals(10000, onlyAddress.getPort());
     }
 
     @Test(expected = ConfigException.class)
     public void testNoPort() {
         check("127.0.0.1");
     }
+
+    @Test(expected = ConfigException.class)
+    public void testOnlyBadHostname() {
+        check("some.invalid.hostname.foo.bar:9999");
+    }
 
-    private void check(String... url) {
-        ClientUtils.parseAndValidateAddresses(Arrays.asList(url));
+    private List<InetSocketAddress> check(String... url) {
+        return ClientUtils.parseAndValidateAddresses(Arrays.asList(url));
     }
 }