[apache#1953] [apache#1946] [apache#1941] [apache#1920] [apache#1910]…

… improment(trino): make version atomic to keep thread safe (apache#2139)

<!--
1. Title: [#<issue>] <type>(<scope>): <subject>
   Examples:
     - "[apache#123] feat(operator): support xxx"
     - "[apache#233] fix: check null before access result in xxx"
     - "[MINOR] refactor: fix typo in variable name"
     - "[MINOR] docs: fix typo in README"
     - "[apache#255] test: fix flaky test NameOfTheTest"
   Reference: https://www.conventionalcommits.org/en/v1.0.0/
2. If the PR is unfinished, please mark this PR as draft.
-->

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

close apache#1953 apache#1946 apache#1941 apache#1920 apache#1910 

### Why are the changes needed?


Fix: apache#1953 apache#1946 apache#1941 apache#1920 apache#1910 

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

- no

### How was this patch tested?

- ut

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/hive/HiveConnectorAdapter.java

@@ -15,11 +15,12 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /** Transforming Hive connector configuration and components into Gravitino connector. */
 public class HiveConnectorAdapter implements CatalogConnectorAdapter {
 
-  private static int version = 0;
+  private static final AtomicInteger VERSION = new AtomicInteger(0);
 
   private final HasPropertyMeta propertyMetadata;
   private final PropertyConverter catalogConverter;
@@ -33,7 +34,8 @@ public Map<String, Object> buildInternalConnectorConfig(GravitinoCatalog catalog
       throws Exception {
     Map<String, Object> config = new HashMap<>();
     config.put(
-        "catalogHandle", String.format("%s_v%d:normal:default", catalog.getName(), version++));
+        "catalogHandle",
+        String.format("%s_v%d:normal:default", catalog.getName(), VERSION.getAndIncrement()));
     config.put("connectorName", "hive");
 
     Map<String, Object> properties = new HashMap<>();

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/iceberg/IcebergConnectorAdapter.java

@@ -14,11 +14,12 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /** Transforming Iceberg connector configuration and components into Gravitino connector. */
 public class IcebergConnectorAdapter implements CatalogConnectorAdapter {
 
-  private static int version = 0;
+  private static final AtomicInteger VERSION = new AtomicInteger(0);
   private final IcebergPropertyMeta propertyMetadata;
   private final PropertyConverter catalogConverter;
 
@@ -31,7 +32,8 @@ public Map<String, Object> buildInternalConnectorConfig(GravitinoCatalog catalog
       throws Exception {
     Map<String, Object> config = new HashMap<>();
     config.put(
-        "catalogHandle", String.format("%s_v%d:normal:default", catalog.getName(), version++));
+        "catalogHandle",
+        String.format("%s_v%d:normal:default", catalog.getName(), VERSION.getAndIncrement()));
     config.put("connectorName", "iceberg");
 
     Map<String, String> properties =

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/jdbc/mysql/MySQLConnectorAdapter.java

@@ -16,12 +16,13 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /** Transforming MySQL connector configuration and components into Gravitino connector. */
 public class MySQLConnectorAdapter implements CatalogConnectorAdapter {
 
   private final PropertyConverter catalogConverter;
-  private static int version = 0;
+  private static final AtomicInteger VERSION = new AtomicInteger(0);
   private final HasPropertyMeta propertyMetadata;
 
   public MySQLConnectorAdapter() {
@@ -33,7 +34,8 @@ public Map<String, Object> buildInternalConnectorConfig(GravitinoCatalog catalog
       throws Exception {
     Map<String, Object> config = new HashMap<>();
     config.put(
-        "catalogHandle", String.format("%s_v%d:normal:default", catalog.getName(), version++));
+        "catalogHandle",
+        String.format("%s_v%d:normal:default", catalog.getName(), VERSION.getAndIncrement()));
     config.put("connectorName", "mysql");
 
     Map<String, String> properties =

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/jdbc/postgresql/PostgreSQLConnectorAdapter.java

@@ -13,11 +13,12 @@
 import com.datastrato.gravitino.trino.connector.metadata.GravitinoCatalog;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /** Transforming PostgreSQL connector configuration and components into Gravitino connector. */
 public class PostgreSQLConnectorAdapter implements CatalogConnectorAdapter {
   private final PropertyConverter catalogConverter;
-  private static int version = 0;
+  private static final AtomicInteger VERSION = new AtomicInteger(0);
 
   public PostgreSQLConnectorAdapter() {
     this.catalogConverter = new JDBCCatalogPropertyConverter();
@@ -27,7 +28,8 @@ public Map<String, Object> buildInternalConnectorConfig(GravitinoCatalog catalog
       throws Exception {
     Map<String, Object> config = new HashMap<>();
     config.put(
-        "catalogHandle", String.format("%s_v%d:normal:default", catalog.getName(), version++));
+        "catalogHandle",
+        String.format("%s_v%d:normal:default", catalog.getName(), VERSION.getAndIncrement()));
     config.put("connectorName", "postgresql");
 
     Map<String, String> properties =

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/memory/MemoryConnectorAdapter.java

@@ -13,14 +13,15 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Support trino Memory connector for testing. Transforming Memory connector configuration and
  * components into Gravitino connector.
  */
 public class MemoryConnectorAdapter implements CatalogConnectorAdapter {
 
-  private static int version = 0;
+  private static final AtomicInteger VERSION = new AtomicInteger(0);
   private final HasPropertyMeta propertyMetadata;
 
   public MemoryConnectorAdapter() {
@@ -31,7 +32,8 @@ public MemoryConnectorAdapter() {
   public Map<String, Object> buildInternalConnectorConfig(GravitinoCatalog catalog) {
     Map<String, Object> config = new HashMap<>();
     config.put(
-        "catalogHandle", String.format("%s_v%d:normal:default", catalog.getName(), version++));
+        "catalogHandle",
+        String.format("%s_v%d:normal:default", catalog.getName(), VERSION.getAndIncrement()));
     config.put("connectorName", "memory");
 
     Map<String, Object> properties = new HashMap<>();