[CELEBORN-2162] Supress client side connection error logging

common/src/main/java/org/apache/celeborn/common/client/MasterClient.java

@@ -272,7 +272,7 @@ private RpcEndpointRef setupEndpointRef(String endpoint) {
       // Catch all exceptions. Because we don't care whether this exception is IOException or
       // TimeoutException or other exceptions, so we just try to connect to host:port, if fail,
       // we try next address.
-      LOG.warn("Connect to {} failed.", endpoint, e);
+      LOG.info("Connect to {} failed.", endpoint);
     }
     return endpointRef;
   }

common/src/main/java/org/apache/celeborn/common/network/client/TransportClientFactory.java

@@ -41,8 +41,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.celeborn.common.client.MasterNotLeaderException;
 import org.apache.celeborn.common.exception.CelebornIOException;
 import org.apache.celeborn.common.network.TransportContext;
+import org.apache.celeborn.common.network.sasl.registration.RegistrationClientBootstrap;
 import org.apache.celeborn.common.network.server.TransportChannelHandler;
 import org.apache.celeborn.common.network.util.*;
 import org.apache.celeborn.common.util.JavaUtils;
@@ -350,19 +352,34 @@ public void operationComplete(final Future<Channel> handshakeFuture) {
     // Execute any client bootstraps synchronously before marking the Client as successful.
     long preBootstrap = System.nanoTime();
     logger.debug("Running bootstraps for {} ...", address);
-    try {
-      for (TransportClientBootstrap clientBootstrap : clientBootstraps) {
+    for (TransportClientBootstrap clientBootstrap : clientBootstraps) {
+      try {
         clientBootstrap.doBootstrap(client);
+      } catch (
+          Exception e) { // catch non-RuntimeExceptions too as bootstrap may be written in Scala
+        long bootstrapTime = System.nanoTime() - preBootstrap;
+        if (clientBootstrap instanceof RegistrationClientBootstrap) {
+          Exception processed = RegistrationClientBootstrap.processMasterNotLeaderException(e);
+          String message =
+              (processed instanceof MasterNotLeaderException)
+                  ? String.format(
+                      "Suggested leader is %s",
+                      ((MasterNotLeaderException) processed).getSuggestedLeaderAddress())
+                  : e.getMessage();
+          logger.warn(
+              "Attempted to register with a Master that is not the leader after {}: {}",
+              Utils.nanoDurationToString(bootstrapTime),
+              message);
+        } else {
+          logger.error(
+              "Exception while bootstrapping client after {}",
+              Utils.nanoDurationToString(bootstrapTime),
+              e);
+        }
+        client.close();
+        Throwables.throwIfUnchecked(e);
+        throw new RuntimeException(e);
       }
-    } catch (Exception e) { // catch non-RuntimeExceptions too as bootstrap may be written in Scala
-      long bootstrapTime = System.nanoTime() - preBootstrap;
-      logger.error(
-          "Exception while bootstrapping client after {}",
-          Utils.nanoDurationToString(bootstrapTime),
-          e);
-      client.close();
-      Throwables.throwIfUnchecked(e);
-      throw new RuntimeException(e);
     }
     long postBootstrap = System.nanoTime();
     logger.debug(

common/src/main/java/org/apache/celeborn/common/network/sasl/registration/RegistrationClientBootstrap.java

@@ -276,7 +276,7 @@ private static Map<PbAuthType, Set<String>> findSupportedSaslMechs(
           Pattern.MULTILINE | Pattern.DOTALL);
 
   @SuppressWarnings("NonInclusiveLanguage")
-  static Exception processMasterNotLeaderException(Exception ex) {
+  public static Exception processMasterNotLeaderException(Exception ex) {
     String stringified = Throwables.getStackTraceAsString(ex);
 
     Matcher matcher = MASTER_NOT_LEADER_EXCEPTION_PATTERN.matcher(stringified);