[apache#2866] improve(java-clients) Graviton java client support chec…

…k version and backward compatibility (apache#2867)

### What changes were proposed in this pull request?

Graviton java client  support check version and backward compatibility

### Why are the changes needed?

Fix: apache#2866

### Does this PR introduce _any_ user-facing change?

NO

### How was this patch tested?

UT

.gitignore

@@ -35,7 +35,7 @@ out/**
 *.ipr
 
 distribution
-server/src/main/resources/project.properties
+common/src/main/resources/project.properties
 
 dev/docker/*/packages
 docs/build

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoAdminClient.java

@@ -40,11 +40,16 @@ public class GravitinoAdminClient extends GravitinoClientBase implements Support
    *
    * @param uri The base URI for the Gravitino API.
    * @param authDataProvider The provider of the data which is used for authentication.
+   * @param checkVersion Whether to check the version of the Gravitino server. Gravitino does not
+   *     support the case that the client-side version is higher than the server-side version.
    * @param headers The base header for Gravitino API.
    */
   private GravitinoAdminClient(
-      String uri, AuthDataProvider authDataProvider, Map<String, String> headers) {
-    super(uri, authDataProvider, headers);
+      String uri,
+      AuthDataProvider authDataProvider,
+      boolean checkVersion,
+      Map<String, String> headers) {
+    super(uri, authDataProvider, checkVersion, headers);
   }
 
   /**
@@ -189,8 +194,7 @@ protected AdminClientBuilder(String uri) {
     public GravitinoAdminClient build() {
       Preconditions.checkArgument(
           uri != null && !uri.isEmpty(), "The argument 'uri' must be a valid URI");
-
-      return new GravitinoAdminClient(uri, authDataProvider, headers);
+      return new GravitinoAdminClient(uri, authDataProvider, checkVersion, headers);
     }
   }
 }

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoClient.java

@@ -33,15 +33,18 @@ public class GravitinoClient extends GravitinoClientBase implements SupportsCata
    * @param uri The base URI for the Gravitino API.
    * @param metalakeName The specified metalake name.
    * @param authDataProvider The provider of the data which is used for authentication.
+   * @param checkVersion Whether to check the version of the Gravitino server. Gravitino does not
+   *     support the case that the client-side version is higher than the server-side version.
    * @param headers The base header for Gravitino API.
    * @throws NoSuchMetalakeException if the metalake with specified name does not exist.
    */
   private GravitinoClient(
       String uri,
       String metalakeName,
       AuthDataProvider authDataProvider,
+      boolean checkVersion,
       Map<String, String> headers) {
-    super(uri, authDataProvider, headers);
+    super(uri, authDataProvider, checkVersion, headers);
     this.metalake = loadMetalake(NameIdentifier.of(metalakeName));
   }
 
@@ -143,7 +146,7 @@ public GravitinoClient build() {
           metalakeName != null && !metalakeName.isEmpty(),
           "The argument 'metalakeName' must be a valid name");
 
-      return new GravitinoClient(uri, metalakeName, authDataProvider, headers);
+      return new GravitinoClient(uri, metalakeName, authDataProvider, checkVersion, headers);
     }
   }
 }

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoClientBase.java

@@ -6,9 +6,14 @@
 package com.datastrato.gravitino.client;
 
 import com.datastrato.gravitino.NameIdentifier;
+import com.datastrato.gravitino.Version;
 import com.datastrato.gravitino.dto.responses.MetalakeResponse;
 import com.datastrato.gravitino.dto.responses.VersionResponse;
+import com.datastrato.gravitino.exceptions.GravitinoRuntimeException;
 import com.datastrato.gravitino.exceptions.NoSuchMetalakeException;
+import com.datastrato.gravitino.json.JsonUtils;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableMap;
 import java.io.Closeable;
 import java.net.URI;
@@ -39,16 +44,61 @@ public abstract class GravitinoClientBase implements Closeable {
    *
    * @param uri The base URI for the Gravitino API.
    * @param authDataProvider The provider of the data which is used for authentication.
+   * @param checkVersion Whether to check the version of the Gravitino server.
    * @param headers The base header of the Gravitino API.
    */
   protected GravitinoClientBase(
-      String uri, AuthDataProvider authDataProvider, Map<String, String> headers) {
-    this.restClient =
-        HTTPClient.builder(Collections.emptyMap())
-            .uri(uri)
-            .withAuthDataProvider(authDataProvider)
-            .withHeaders(headers)
-            .build();
+      String uri,
+      AuthDataProvider authDataProvider,
+      boolean checkVersion,
+      Map<String, String> headers) {
+    ObjectMapper mapper = JsonUtils.objectMapper();
+    mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+
+    if (checkVersion) {
+      this.restClient =
+          HTTPClient.builder(Collections.emptyMap())
+              .uri(uri)
+              .withAuthDataProvider(authDataProvider)
+              .withObjectMapper(mapper)
+              .withPreConnectHandler(this::checkVersion)
+              .withHeaders(headers)
+              .build();
+
+    } else {
+      this.restClient =
+          HTTPClient.builder(Collections.emptyMap())
+              .uri(uri)
+              .withAuthDataProvider(authDataProvider)
+              .withObjectMapper(mapper)
+              .withHeaders(headers)
+              .build();
+    }
+  }
+
+  /**
+   * Check the compatibility of the client with the target server.
+   *
+   * @throws GravitinoRuntimeException If the client version is greater than the server version.
+   */
+  public void checkVersion() {
+    GravitinoVersion serverVersion = serverVersion();
+    GravitinoVersion clientVersion = clientVersion();
+    if (clientVersion.compareTo(serverVersion) > 0) {
+      throw new GravitinoRuntimeException(
+          "Gravitino does not support the case that the client-side version is higher than the server-side version."
+              + "The client version is %s, and the server version %s",
+          clientVersion.version(), serverVersion.version());
+    }
+  }
+
+  /**
+   * Retrieves the version of the Gravitino client.
+   *
+   * @return A GravitinoVersion instance representing the version of the Gravitino client.
+   */
+  public GravitinoVersion clientVersion() {
+    return new GravitinoVersion(Version.getCurrentVersionDTO());
   }
 
   /**
@@ -77,7 +127,17 @@ public GravitinoMetalake loadMetalake(NameIdentifier ident) throws NoSuchMetalak
    *
    * @return A GravitinoVersion instance representing the version of the Gravitino API.
    */
+  @Deprecated
   public GravitinoVersion getVersion() {
+    return serverVersion();
+  }
+
+  /**
+   * Retrieves the server version of the Gravitino server.
+   *
+   * @return A GravitinoVersion instance representing the server version of the Gravitino API.
+   */
+  public GravitinoVersion serverVersion() {
     VersionResponse resp =
         restClient.get(
             "api/version",
@@ -108,6 +168,8 @@ public abstract static class Builder<T> {
     protected String uri;
     /** The authentication provider. */
     protected AuthDataProvider authDataProvider;
+    /** The check version flag. */
+    protected boolean checkVersion = true;
     /** The request base header for the Gravitino API. */
     protected Map<String, String> headers = ImmutableMap.of();
 
@@ -130,6 +192,16 @@ public Builder<T> withSimpleAuth() {
       return this;
     }
 
+    /**
+     * Optional, set a flag to verify the client is supported to connector the server
+     *
+     * @return This Builder instance for method chaining.
+     */
+    public Builder<T> withVersionCheckDisabled() {
+      this.checkVersion = false;
+      return this;
+    }
+
     /**
      * Sets OAuth2TokenProvider for Gravitino.
      *

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoVersion.java

@@ -5,10 +5,55 @@
 package com.datastrato.gravitino.client;
 
 import com.datastrato.gravitino.dto.VersionDTO;
+import com.datastrato.gravitino.exceptions.GravitinoRuntimeException;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /** Gravitino version information. */
-public class GravitinoVersion extends VersionDTO {
+public class GravitinoVersion extends VersionDTO implements Comparable {
+
+  private static final int VERSION_PART_NUMBER = 3;
+
+  @VisibleForTesting
+  GravitinoVersion(String version, String compoileDate, String gitCommit) {
+    super(version, compoileDate, gitCommit);
+  }
+
   GravitinoVersion(VersionDTO versionDTO) {
     super(versionDTO.version(), versionDTO.compileDate(), versionDTO.gitCommit());
   }
+
+  @VisibleForTesting
+  /** @return parse the version number for a version string */
+  int[] getVersionNumber() {
+    Pattern pattern = Pattern.compile("(\\d+)\\.(\\d+)\\.(\\d+)(-\\w+){0,1}");
+    Matcher matcher = pattern.matcher(version());
+    if (matcher.matches()) {
+      int[] versionNumbers = new int[VERSION_PART_NUMBER];
+      for (int i = 0; i < VERSION_PART_NUMBER; i++) {
+        versionNumbers[i] = Integer.parseInt(matcher.group(i + 1));
+      }
+      return versionNumbers;
+    }
+    throw new GravitinoRuntimeException("Invalid version string " + version());
+  }
+
+  @Override
+  public int compareTo(Object o) {
+    if (!(o instanceof GravitinoVersion)) {
+      return 1;
+    }
+    GravitinoVersion other = (GravitinoVersion) o;
+
+    int[] left = getVersionNumber();
+    int[] right = other.getVersionNumber();
+    for (int i = 0; i < VERSION_PART_NUMBER; i++) {
+      int v = left[i] - right[i];
+      if (v != 0) {
+        return v;
+      }
+    }
+    return 0;
+  }
 }

clients/client-java/src/main/java/com/datastrato/gravitino/client/HTTPClient.java

@@ -79,19 +79,36 @@ public class HTTPClient implements RESTClient {
   private final ObjectMapper mapper;
   private final AuthDataProvider authDataProvider;
 
+  // Handler to be executed before connecting to the server.
+  private final Runnable beforeConnectHandler;
+  // Handler status
+  enum HandlerStatus {
+    // The handler has not been executed yet.
+    Start,
+    // The handler has been executed successfully.
+    Finished,
+    // The handler is currently running.
+    Running,
+  }
+
+  // The status of the handler.
+  private volatile HandlerStatus handlerStatus = HandlerStatus.Start;
+
   /**
    * Constructs an instance of HTTPClient with the provided information.
    *
    * @param uri The base URI of the REST API.
    * @param baseHeaders A map of base headers to be included in all HTTP requests.
    * @param objectMapper The ObjectMapper used for JSON serialization and deserialization.
    * @param authDataProvider The provider of authentication data.
+   * @param beforeConnectHandler The function to be executed before connecting to the server.
    */
   private HTTPClient(
       String uri,
       Map<String, String> baseHeaders,
       ObjectMapper objectMapper,
-      AuthDataProvider authDataProvider) {
+      AuthDataProvider authDataProvider,
+      Runnable beforeConnectHandler) {
     this.uri = uri;
     this.mapper = objectMapper;
 
@@ -106,6 +123,11 @@ private HTTPClient(
 
     this.httpClient = clientBuilder.build();
     this.authDataProvider = authDataProvider;
+
+    if (beforeConnectHandler == null) {
+      handlerStatus = HandlerStatus.Finished;
+    }
+    this.beforeConnectHandler = beforeConnectHandler;
   }
 
   /**
@@ -314,6 +336,11 @@ private <T> T execute(
       Map<String, String> headers,
       Consumer<ErrorResponse> errorHandler,
       Consumer<Map<String, String>> responseHeaders) {
+
+    if (handlerStatus != HandlerStatus.Finished) {
+      performPreConnectHandler();
+    }
+
     if (path.startsWith("/")) {
       throw new RESTException(
           "Received a malformed path for a REST request: %s. Paths should not start with /", path);
@@ -383,6 +410,21 @@ private <T> T execute(
     }
   }
 
+  private synchronized void performPreConnectHandler() {
+    // beforeConnectHandler is a pre-connection handler that needs to be executed before the first
+    // HTTP request. if the handler execute fails, we set the status to Start to retry the handler.
+    if (handlerStatus == HandlerStatus.Start) {
+      handlerStatus = HandlerStatus.Running;
+      try {
+        beforeConnectHandler.run();
+        handlerStatus = HandlerStatus.Finished;
+      } catch (Exception e) {
+        handlerStatus = HandlerStatus.Start;
+        throw e;
+      }
+    }
+  }
+
   /**
    * Sends an HTTP HEAD request to the specified path and processes the response.
    *
@@ -655,6 +697,7 @@ public static class Builder {
     private String uri;
     private ObjectMapper mapper = JsonUtils.objectMapper();
     private AuthDataProvider authDataProvider;
+    private Runnable beforeConnectHandler;
 
     private Builder(Map<String, String> properties) {
       this.properties = properties;
@@ -707,6 +750,17 @@ public Builder withObjectMapper(ObjectMapper objectMapper) {
       return this;
     }
 
+    /**
+     * Sets the preConnect handle for the HTTP client.
+     *
+     * @param beforeConnectHandler The handle run before connect to the server .
+     * @return This Builder instance for method chaining.
+     */
+    public Builder withPreConnectHandler(Runnable beforeConnectHandler) {
+      this.beforeConnectHandler = beforeConnectHandler;
+      return this;
+    }
+
     /**
      * Sets the AuthDataProvider for the HTTP client.
      *
@@ -725,7 +779,7 @@ public Builder withAuthDataProvider(AuthDataProvider authDataProvider) {
      */
     public HTTPClient build() {
 
-      return new HTTPClient(uri, baseHeaders, mapper, authDataProvider);
+      return new HTTPClient(uri, baseHeaders, mapper, authDataProvider, beforeConnectHandler);
     }
   }
 

clients/client-java/src/test/java/com/datastrato/gravitino/client/TestBase.java

@@ -22,7 +22,7 @@
 
 public abstract class TestBase {
 
-  private static final ObjectMapper MAPPER = JsonUtils.objectMapper();
+  protected static final ObjectMapper MAPPER = JsonUtils.objectMapper();
 
   protected static ClientAndServer mockServer;
 
@@ -32,7 +32,8 @@ public abstract class TestBase {
   public static void setUp() throws Exception {
     mockServer = ClientAndServer.startClientAndServer(0);
     int port = mockServer.getLocalPort();
-    client = GravitinoAdminClient.builder("http://127.0.0.1:" + port).build();
+    client =
+        GravitinoAdminClient.builder("http://127.0.0.1:" + port).withVersionCheckDisabled().build();
   }
 
   @AfterAll