feat: close catalog

eric666666 · Sep 5, 2024 · 7419b06 · 7419b06
1 parent a3204fc
commit 7419b06
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 0 deletions.
diff --git a/.../main/java/org/apache/paimon/flink/sink/cdc/StpCdcDynamicTableParsingProcessFunction.java b/.../main/java/org/apache/paimon/flink/sink/cdc/StpCdcDynamicTableParsingProcessFunction.java
@@ -154,4 +154,12 @@ record ->
     private CdcMultiplexRecord wrapRecord(String databaseName, String tableName, CdcRecord record) {
         return CdcMultiplexRecord.fromCdcRecord(databaseName, tableName, record);
     }
+
+    @Override
+    public void close() throws Exception {
+        if (catalog != null) {
+            catalog.close();
+            catalog = null;
+        }
+    }
 }
diff --git a/...va/org/apache/paimon/flink/sink/cdc/StpCdcRecordStoreDynamicBucketMultiWriteOperator.java b/...va/org/apache/paimon/flink/sink/cdc/StpCdcRecordStoreDynamicBucketMultiWriteOperator.java
@@ -198,6 +198,10 @@ public void close() throws Exception {
         if (compactExecutor != null) {
             compactExecutor.shutdownNow();
         }
+        if (catalog != null) {
+            catalog.close();
+            catalog = null;
+        }
     }
 
     @Override

diff --git a/...va/org/apache/paimon/flink/sink/cdc/StpCdcRecordStoreUnawareBucketMultiWriteOperator.java b/...va/org/apache/paimon/flink/sink/cdc/StpCdcRecordStoreUnawareBucketMultiWriteOperator.java
@@ -199,6 +199,10 @@ public void close() throws Exception {
         if (compactExecutor != null) {
             compactExecutor.shutdownNow();
         }
+        if (catalog != null) {
+            catalog.close();
+            catalog = null;
+        }
     }
 
     @Override