[HiveSink]Fix the risk of resource leakage. (apache#6721)

Co-authored-by: lightzhao <[email protected]>
TongchengOpenSource · Apr 19, 2024 · c23804f · c23804f
1 parent 94afa89
commit c23804f
Showing 1 changed file with 18 additions and 15 deletions.
diff --git a/...va/org/apache/seatunnel/connectors/seatunnel/hive/commit/HiveSinkAggregatedCommitter.java b/...va/org/apache/seatunnel/connectors/seatunnel/hive/commit/HiveSinkAggregatedCommitter.java
@@ -57,26 +57,29 @@ public HiveSinkAggregatedCommitter(
     @Override
     public List<FileAggregatedCommitInfo> commit(
             List<FileAggregatedCommitInfo> aggregatedCommitInfos) throws IOException {
-        HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(pluginConfig);
         List<FileAggregatedCommitInfo> errorCommitInfos = super.commit(aggregatedCommitInfos);
         if (errorCommitInfos.isEmpty()) {
-            for (FileAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) {
-                Map<String, List<String>> partitionDirAndValuesMap =
-                        aggregatedCommitInfo.getPartitionDirAndValuesMap();
-                List<String> partitions =
-                        partitionDirAndValuesMap.keySet().stream()
-                                .map(partition -> partition.replaceAll("\\\\", "/"))
-                                .collect(Collectors.toList());
-                try {
-                    hiveMetaStore.addPartitions(dbName, tableName, partitions);
-                    log.info("Add these partitions {}", partitions);
-                } catch (TException e) {
-                    log.error("Failed to add these partitions {}", partitions, e);
-                    errorCommitInfos.add(aggregatedCommitInfo);
+            HiveMetaStoreProxy hiveMetaStore = HiveMetaStoreProxy.getInstance(pluginConfig);
+            try {
+                for (FileAggregatedCommitInfo aggregatedCommitInfo : aggregatedCommitInfos) {
+                    Map<String, List<String>> partitionDirAndValuesMap =
+                            aggregatedCommitInfo.getPartitionDirAndValuesMap();
+                    List<String> partitions =
+                            partitionDirAndValuesMap.keySet().stream()
+                                    .map(partition -> partition.replaceAll("\\\\", "/"))
+                                    .collect(Collectors.toList());
+                    try {
+                        hiveMetaStore.addPartitions(dbName, tableName, partitions);
+                        log.info("Add these partitions {}", partitions);
+                    } catch (TException e) {
+                        log.error("Failed to add these partitions {}", partitions, e);
+                        errorCommitInfos.add(aggregatedCommitInfo);
+                    }
                 }
+            } finally {
+                hiveMetaStore.close();
             }
         }
-        hiveMetaStore.close();
         return errorCommitInfos;
     }