[KYUUBI #2881] [SUB-TASK][KPIP-4] Rest client supports retry request if catch net exception

### _Why are the changes needed?_

close #2880

The main changes:
- Add a interface IRestClient for the underlying rest client
- Add KyuubiRetryableException to make call side aware whether attempts
- Add RetryableRestClient which delegates the request to the real RestClient and provide a functionality of request attempts
- Make KyuubiRestClient supports multi-hostUrls

### _How was this patch tested?_
- [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2881 from ulysses-you/rest-retry.

Closes #2881

edd3934 [ulysses-you] comment
df94e71 [ulysses-you] fix
5e16343 [ulysses-you] Rest client supports retry request if catch net exception

Authored-by: ulysses-you <ulyssesyou18@gmail.com>
Signed-off-by: ulysses-you <ulyssesyou@apache.org>

Author: ulysses-you

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/BatchRestApi.java

@@ -89,7 +89,7 @@ public CloseBatchResponse deleteBatch(String batchId, String hs2ProxyUser) {
     return this.getClient().delete(path, params, CloseBatchResponse.class, client.getAuthHeader());
   }
 
-  private RestClient getClient() {
+  private IRestClient getClient() {
     return this.client.getHttpClient();
   }
 }

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/HttpClientFactory.java

@@ -0,0 +1,56 @@
+/*
+ * 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.kyuubi.client;
+
+import javax.net.ssl.SSLContext;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.conn.ssl.NoopHostnameVerifier;
+import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
+import org.apache.http.conn.ssl.TrustStrategy;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.ssl.SSLContexts;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class HttpClientFactory {
+  private static final Logger LOG = LoggerFactory.getLogger(HttpClientFactory.class);
+
+  public static CloseableHttpClient createHttpClient(RestClientConf conf) {
+    RequestConfig requestConfig =
+        RequestConfig.custom()
+            .setSocketTimeout(conf.getSocketTimeout())
+            .setConnectTimeout(conf.getConnectTimeout())
+            .build();
+    SSLConnectionSocketFactory sslSocketFactory;
+    try {
+      TrustStrategy acceptingTrustStrategy = (cert, authType) -> true;
+      SSLContext sslContext =
+          SSLContexts.custom().loadTrustMaterial(null, acceptingTrustStrategy).build();
+      sslSocketFactory = new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);
+    } catch (Exception e) {
+      LOG.error("Error: ", e);
+      throw new RuntimeException(e);
+    }
+
+    return HttpClientBuilder.create()
+        .setDefaultRequestConfig(requestConfig)
+        .setSSLSocketFactory(sslSocketFactory)
+        .build();
+  }
+}

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/IRestClient.java

@@ -0,0 +1,35 @@
+/*
+ * 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.kyuubi.client;
+
+import java.util.Map;
+
+/** A underlying http client interface for common rest request. */
+public interface IRestClient extends AutoCloseable {
+  public <T> T get(String path, Map<String, Object> params, Class<T> type, String authHeader);
+
+  public String get(String path, Map<String, Object> params, String authHeader);
+
+  public <T> T post(String path, String body, Class<T> type, String authHeader);
+
+  public String post(String path, String body, String authHeader);
+
+  public <T> T delete(String path, Map<String, Object> params, Class<T> type, String authHeader);
+
+  public String delete(String path, Map<String, Object> params, String authHeader);
+}

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/KyuubiRestClient.java

@@ -18,24 +18,15 @@
 package org.apache.kyuubi.client;
 
 import java.net.URI;
-import javax.net.ssl.SSLContext;
+import java.util.Arrays;
+import java.util.LinkedList;
+import java.util.List;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.http.client.config.RequestConfig;
-import org.apache.http.conn.ssl.NoopHostnameVerifier;
-import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
-import org.apache.http.conn.ssl.TrustStrategy;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClientBuilder;
-import org.apache.http.ssl.SSLContexts;
 import org.apache.kyuubi.client.auth.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class KyuubiRestClient implements AutoCloseable {
 
-  private static final Logger LOG = LoggerFactory.getLogger(KyuubiRestClient.class);
-
-  private RestClient httpClient;
+  private IRestClient httpClient;
 
   private AuthHeaderGenerator authHeaderGenerator;
 
@@ -64,13 +55,21 @@ public void close() throws Exception {
   private KyuubiRestClient() {}
 
   private KyuubiRestClient(Builder builder) {
-    // Remove the trailing "/" from the hostUrl if present
-    String hostUrl = builder.hostUrl.replaceAll("/$", "");
-    String baseUrl = String.format("%s/%s", hostUrl, builder.version.getApiNamespace());
+    List<String> baseUrls = new LinkedList<>();
+    for (String hostUrl : builder.hostUrls) {
+      // Remove the trailing "/" from the hostUrl if present
+      String baseUrl =
+          String.format("%s/%s", hostUrl.replaceAll("/$", ""), builder.version.getApiNamespace());
+      baseUrls.add(baseUrl);
+    }
 
-    CloseableHttpClient httpclient = initHttpClient(builder);
+    RestClientConf conf = new RestClientConf();
+    conf.setConnectTimeout(builder.connectTimeout);
+    conf.setSocketTimeout(builder.socketTimeout);
+    conf.setMaxAttempts(builder.maxAttempts);
+    conf.setAttemptWaitTime(builder.attemptWaitTime);
 
-    this.httpClient = new RestClient(baseUrl, httpclient);
+    this.httpClient = RetryableRestClient.getRestClient(baseUrls, conf);
 
     switch (builder.authHeaderMethod) {
       case BASIC:
@@ -94,42 +93,25 @@ public String getAuthHeader() {
     return authHeaderGenerator.generateAuthHeader();
   }
 
-  public RestClient getHttpClient() {
+  public IRestClient getHttpClient() {
     return httpClient;
   }
 
-  private CloseableHttpClient initHttpClient(Builder builder) {
-    RequestConfig requestConfig =
-        RequestConfig.custom()
-            .setSocketTimeout(builder.socketTimeout)
-            .setConnectTimeout(builder.connectTimeout)
-            .build();
-    SSLConnectionSocketFactory sslSocketFactory;
-    try {
-      TrustStrategy acceptingTrustStrategy = (cert, authType) -> true;
-      SSLContext sslContext =
-          SSLContexts.custom().loadTrustMaterial(null, acceptingTrustStrategy).build();
-      sslSocketFactory = new SSLConnectionSocketFactory(sslContext, NoopHostnameVerifier.INSTANCE);
-    } catch (Exception e) {
-      LOG.error("Error: ", e);
-      throw new RuntimeException(e);
-    }
-
-    CloseableHttpClient httpclient =
-        HttpClientBuilder.create()
-            .setDefaultRequestConfig(requestConfig)
-            .setSSLSocketFactory(sslSocketFactory)
-            .build();
-    return httpclient;
-  }
-
   public static Builder builder(String hostUrl) {
     return new Builder(hostUrl);
   }
 
+  public static Builder builder(String... hostUrls) {
+    return new Builder(Arrays.asList(hostUrls));
+  }
+
+  public static Builder builder(List<String> hostUrls) {
+    return new Builder(hostUrls);
+  }
+
   public static class Builder {
 
-    private String hostUrl;
+    private List<String> hostUrls;
 
     private String spnegoHost;
 
@@ -147,8 +129,23 @@ public static class Builder {
 
     private int connectTimeout = 3000;
 
+    private int maxAttempts = 3;
+
+    private int attemptWaitTime = 3000;
+
     public Builder(String hostUrl) {
-      this.hostUrl = hostUrl;
+      if (StringUtils.isBlank(hostUrl)) {
+        throw new IllegalArgumentException("hostUrl cannot be blank.");
+      }
+      this.hostUrls = new LinkedList<>();
+      this.hostUrls.add(hostUrl);
+    }
+
+    public Builder(List<String> hostUrls) {
+      if (hostUrls.isEmpty()) {
+        throw new IllegalArgumentException("hostUrls cannot be blank.");
+      }
+      this.hostUrls = hostUrls;
     }
 
     public Builder spnegoHost(String host) {
@@ -193,16 +190,27 @@ public Builder connectionTimeout(int connectTimeout) {
       return this;
     }
 
-    public KyuubiRestClient build() {
-      if (StringUtils.isBlank(hostUrl)) {
-        throw new IllegalArgumentException("hostUrl cannot be blank.");
-      }
+    public Builder maxAttempts(int maxAttempts) {
+      this.maxAttempts = maxAttempts;
+      return this;
+    }
+
+    public Builder attemptWaitTime(int attemptWaitTime) {
+      this.attemptWaitTime = attemptWaitTime;
+      return this;
+    }
 
+    public KyuubiRestClient build() {
       if (authHeaderMethod == AuthHeaderMethod.SPNEGO && StringUtils.isBlank(spnegoHost)) {
-        try {
-          this.spnegoHost = new URI(hostUrl).getHost();
-        } catch (Exception e) {
+        if (hostUrls.size() > 1) {
           throw new IllegalArgumentException("spnegoHost is invalid.");
+        } else {
+          // follow the behavior of curl, use host url by default
+          try {
+            this.spnegoHost = new URI(hostUrls.get(0)).getHost();
+          } catch (Exception e) {
+            throw new IllegalArgumentException("spnegoHost is invalid.", e);
+          }
         }
       }
       return new KyuubiRestClient(this);

kyuubi-rest-client/src/main/java/org/apache/kyuubi/client/RestClient.java

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.client;
 
 import java.io.IOException;
+import java.net.ConnectException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
@@ -32,15 +33,17 @@
 import org.apache.http.client.methods.HttpUriRequest;
 import org.apache.http.client.methods.RequestBuilder;
 import org.apache.http.client.utils.URIBuilder;
+import org.apache.http.conn.ConnectTimeoutException;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.util.EntityUtils;
 import org.apache.kyuubi.client.exception.KyuubiRestException;
+import org.apache.kyuubi.client.exception.KyuubiRetryableException;
 import org.apache.kyuubi.client.util.JsonUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class RestClient implements AutoCloseable {
+public class RestClient implements IRestClient {
 
   private static final Logger LOG = LoggerFactory.getLogger(RestClient.class);
 
@@ -60,31 +63,37 @@ public void close() throws Exception {
     }
   }
 
+  @Override
   public <T> T get(String path, Map<String, Object> params, Class<T> type, String authHeader) {
     String responseJson = get(path, params, authHeader);
     return JsonUtil.toObject(responseJson, type);
   }
 
+  @Override
   public String get(String path, Map<String, Object> params, String authHeader) {
     return doRequest(buildURI(path, params), authHeader, RequestBuilder.get());
   }
 
+  @Override
   public <T> T post(String path, String body, Class<T> type, String authHeader) {
     String responseJson = post(path, body, authHeader);
     return JsonUtil.toObject(responseJson, type);
   }
 
+  @Override
   public String post(String path, String body, String authHeader) {
     RequestBuilder postRequestBuilder =
         RequestBuilder.post().setEntity(new StringEntity(body, StandardCharsets.UTF_8));
     return doRequest(buildURI(path), authHeader, postRequestBuilder);
   }
 
+  @Override
   public <T> T delete(String path, Map<String, Object> params, Class<T> type, String authHeader) {
     String responseJson = delete(path, params, authHeader);
     return JsonUtil.toObject(responseJson, type);
   }
 
+  @Override
   public String delete(String path, Map<String, Object> params, String authHeader) {
     return doRequest(buildURI(path, params), authHeader, RequestBuilder.delete());
   }
@@ -118,6 +127,9 @@ private String doRequest(URI uri, String authHeader, RequestBuilder requestBuild
 
       response = httpclient.execute(httpRequest, responseHandler);
       LOG.info("Response: {}", response);
+    } catch (ConnectException | ConnectTimeoutException e) {
+      // net exception can be retried by connecting to other Kyuubi server
+      throw new KyuubiRetryableException("Api request failed for " + uri.toString(), e);
     } catch (Exception e) {
       LOG.error("Error: ", e);
       throw new KyuubiRestException("Api request failed for " + uri.toString(), e);