[FLINK-10213] Task managers cache a negative DNS lookup of the blob s…

…erver indefinitely

* Added a test case for unresolved InetSocketAddresses that fails before the fix
* Updated BlobClient to create the socket using the hostname and port

Added a check in the BlobClientTest.testUnresolvedInetSocketAddress() that the client is connected.

This closes apache#6862.

flink-runtime/src/main/java/org/apache/flink/runtime/blob/BlobClient.java

@@ -92,7 +92,13 @@ public BlobClient(InetSocketAddress serverAddress, Configuration clientConfig) t
  socket = new Socket();
  }
 
- socket.connect(serverAddress, clientConfig.getInteger(BlobServerOptions.CONNECT_TIMEOUT));
+ // Establish the socket using the hostname and port. This avoids a potential issue where the
+ // InetSocketAddress can cache a failure in hostname resolution forever.
+ socket.connect(
+ new InetSocketAddress(
+ serverAddress.getHostName(),
+ serverAddress.getPort()),
+ clientConfig.getInteger(BlobServerOptions.CONNECT_TIMEOUT));
  socket.setSoTimeout(clientConfig.getInteger(BlobServerOptions.SO_TIMEOUT));
  }
  catch (Exception e) {
@@ -185,6 +191,10 @@ public boolean isClosed() {
  return this.socket.isClosed();
  }
 
+ public boolean isConnected() {
+ return socket.isConnected();
+ }
+
  // --------------------------------------------------------------------------------------------
  // GET
  // --------------------------------------------------------------------------------------------

flink-runtime/src/test/java/org/apache/flink/runtime/blob/BlobClientTest.java

@@ -54,6 +54,7 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 /**
@@ -491,7 +492,6 @@ private static void uploadJarFile(
  }
  }
 
-
  /**
  * Tests the socket operation timeout.
  */
@@ -519,6 +519,14 @@ public void testSocketTimeout() {
  }
  }
 
+ @Test
+ public void testUnresolvedInetSocketAddress() throws Exception {
+ try (BlobClient client = new BlobClient(
+ InetSocketAddress.createUnresolved("localhost", getBlobServer().getPort()), getBlobClientConfig())) {
+ assertTrue(client.isConnected());
+ }
+ }
+
  static class TestBlobServer extends BlobServer {
 
  private volatile long blockingMillis = 0;