[FLINK-28913][hive] Fix failed to open HiveCatalog when it's for hive3 (

apache#20573)
liuyongvs · Aug 15, 2022 · cb4ead7 · cb4ead7
1 parent 7661af0
commit cb4ead7
Showing 1 changed file with 17 additions and 6 deletions.
diff --git a/.../src/main/java/org/apache/flink/table/catalog/hive/client/HiveMetastoreClientWrapper.java b/.../src/main/java/org/apache/flink/table/catalog/hive/client/HiveMetastoreClientWrapper.java
@@ -71,9 +71,9 @@ public class HiveMetastoreClientWrapper implements AutoCloseable {
     private static final Logger LOG = LoggerFactory.getLogger(HiveMetastoreClientWrapper.class);
 
     private final IMetaStoreClient client;
-    private final Hive hive;
     private final HiveConf hiveConf;
     private final HiveShim hiveShim;
+    private volatile Hive hive;
 
     public HiveMetastoreClientWrapper(HiveConf hiveConf, String hiveVersion) {
         this(hiveConf, HiveShimLoader.loadHiveShim(hiveVersion));
@@ -87,11 +87,6 @@ public HiveMetastoreClientWrapper(HiveConf hiveConf, HiveShim hiveShim) {
                 HiveCatalog.isEmbeddedMetastore(hiveConf)
                         ? createMetastoreClient()
                         : HiveMetaStoreClient.newSynchronizedClient(createMetastoreClient());
-        try {
-            this.hive = Hive.get(hiveConf);
-        } catch (HiveException e) {
-            throw new FlinkHiveException(e);
-        }
     }
 
     @Override
@@ -349,6 +344,7 @@ public void unlock(long lockid) throws NoSuchLockException, TxnOpenException, TE
 
     public void loadTable(Path loadPath, String tableName, boolean replace, boolean isSrcLocal)
             throws HiveException {
+        initHive();
         hiveShim.loadTable(hive, loadPath, tableName, replace, isSrcLocal);
     }
 
@@ -359,7 +355,22 @@ public void loadPartition(
             boolean isSkewedStoreAsSubdir,
             boolean replace,
             boolean isSrcLocal) {
+        initHive();
         hiveShim.loadPartition(
                 hive, loadPath, tableName, partSpec, isSkewedStoreAsSubdir, replace, isSrcLocal);
     }
+
+    private void initHive() {
+        if (this.hive == null) {
+            synchronized (this) {
+                if (this.hive == null) {
+                    try {
+                        this.hive = Hive.get(hiveConf);
+                    } catch (HiveException e) {
+                        throw new FlinkHiveException(e);
+                    }
+                }
+            }
+        }
+    }
 }