commit 4a8f60f03f7caeb5421c7c88fa34a3ba419b61ee
author Shimin Huang <40719512+collabH@users.noreply.github.com> Wed May 31 00:37:14 2023 +0800
committer GitHub <noreply@github.com> Tue May 30 18:37:14 2023 +0200
tree 4baa165921891b066d6f6477e07769a5756e6435
parent 3744457437ef74696747911897bb746dc2dde6c9

[BAHIR-324] Closing KuduReader at JobManager

flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/format/AbstractKuduInputFormat.java

diff --git a/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/format/AbstractKuduInputFormat.java b/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/format/AbstractKuduInputFormat.java
index 0cdf570..4976241 100644
--- a/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/format/AbstractKuduInputFormat.java
+++ b/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/format/AbstractKuduInputFormat.java
@@ -104,19 +104,23 @@
         }
     }
 
+    private void closeKuduReader() throws IOException {
+        if (kuduReader != null) {
+            kuduReader.close();
+            kuduReader = null;
+        }
+    }
+
     @Override
     public void close() throws IOException {
         if (resultIterator != null) {
             try {
                 resultIterator.close();
             } catch (KuduException e) {
-                e.printStackTrace();
+                log.error("Error while closing reader iterator.", e);
             }
         }
-        if (kuduReader != null) {
-            kuduReader.close();
-            kuduReader = null;
-        }
+        closeKuduReader();
     }
 
     @Override
@@ -131,8 +135,12 @@
 
     @Override
     public KuduInputSplit[] createInputSplits(int minNumSplits) throws IOException {
-        startKuduReader();
-        return kuduReader.createInputSplits(minNumSplits);
+        try {
+            startKuduReader();
+            return kuduReader.createInputSplits(minNumSplits);
+        } finally {
+            closeKuduReader();
+        }
     }
 
     @Override

flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/function/lookup/KuduRowDataLookupFunction.java

diff --git a/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/function/lookup/KuduRowDataLookupFunction.java b/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/function/lookup/KuduRowDataLookupFunction.java
index d54f23a..d29588f 100644
--- a/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/function/lookup/KuduRowDataLookupFunction.java
+++ b/flink-connector-kudu/src/main/java/org/apache/flink/connectors/kudu/table/function/lookup/KuduRowDataLookupFunction.java
@@ -107,7 +107,7 @@
                 ArrayList<RowData> rows = new ArrayList<>();
                 for (KuduInputSplit inputSplit : inputSplits) {
                     KuduReaderIterator<RowData> scanner = kuduReader.scanner(inputSplit.getScanToken());
-                    // 没有启用cache
+                    // not use cache
                     if (cache == null) {
                         while (scanner.hasNext()) {
                             collect(scanner.next());