[WIP][SPARK-30694][SHUFFLE]If exception occured while fetching blocks by ExternalBlockClient, fail early when External Shuffle Service is not alive

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockStoreClient.java

@@ -103,14 +103,26 @@ public void fetchBlocks(
     try {
       RetryingBlockFetcher.BlockFetchStarter blockFetchStarter =
           (blockIds1, listener1) -> {
+
             // Unless this client is closed.
             if (clientFactory != null) {
-              TransportClient client = clientFactory.createClient(host, port);
+              TransportClient client = null;
+              try {
+                client = clientFactory.createClient(host, port);
+              } catch (Exception e) {
+                // throw ExternalShuffleServiceLostException exception then we won't retry to connect
+                // un-connected External Shuffle Service.
+                String msg = "The relative remote external shuffle service (host: " + host + "," +
+                    "port: " + port + "), which maintains the block data can't been connected.";
+                logger.info(msg);
+                throw new ExternalShuffleServiceLostException(msg);
+              }
               new OneForOneBlockFetcher(client, appId, execId,
-                blockIds1, listener1, conf, downloadFileManager).start();
+                  blockIds1, listener1, conf, downloadFileManager).start();
             } else {
               logger.info("This clientFactory was closed. Skipping further block fetch retries.");
             }
+
           };
 
       int maxRetries = conf.maxIORetries();

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleServiceLostException.java

@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.network.shuffle;
+
+/**
+ * Exception throw when we can't connect to external shuffle service to sto unnecessary
+ * retry in ExternalShuffleStoreClient.fetchBlocks()
+ */
+public class ExternalShuffleServiceLostException extends Exception {
+
+  ExternalShuffleServiceLostException(String message) {
+    super(message);
+  }
+}

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RetryingBlockFetcher.java

@@ -58,7 +58,7 @@ public interface BlockFetchStarter {
      * issues.
      */
     void createAndStart(String[] blockIds, BlockFetchingListener listener)
-         throws IOException, InterruptedException;
+        throws IOException, InterruptedException, ExternalShuffleServiceLostException;
   }
 
   /** Shared executor service used for waiting and retrying. */

common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RetryingBlockFetcherSuite.java

@@ -291,7 +291,11 @@ private static void performInteractions(List<? extends Map<String, Object>> inte
     }
 
     assertNotNull(stub);
-    stub.when(fetchStarter).createAndStart(any(), any());
+    try {
+      stub.when(fetchStarter).createAndStart(any(), any());
+    } catch (ExternalShuffleServiceLostException e) {
+      throw new IOException(e);
+    }
     String[] blockIdArray = blockIds.toArray(new String[blockIds.size()]);
     new RetryingBlockFetcher(conf, fetchStarter, blockIdArray, listener).start();
   }