[incubator-hudi] branch master updated: add jssc.stop() (#797)

[vinoth]

This is an automated email from the ASF dual-hosted git repository.

vinoth pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new 9857c4b  add jssc.stop() (#797)
9857c4b is described below

commit 9857c4b21ccb77a97817e1e448c4c54bdd409562
Author: eisig <eisi...@live.com>
AuthorDate: Fri Jul 19 20:01:45 2019 +0800

    add jssc.stop() (#797)
---
 .../java/com/uber/hoodie/utilities/HDFSParquetImporter.java  | 12 +++++++++---
 .../hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java  |  6 +++++-
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git 
a/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/HDFSParquetImporter.java
 
b/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/HDFSParquetImporter.java
index c446afb..661f1a1 100644
--- 
a/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/HDFSParquetImporter.java
+++ 
b/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/HDFSParquetImporter.java
@@ -61,6 +61,7 @@ import scala.Tuple2;
  * Loads data from Parquet Sources
  */
 public class HDFSParquetImporter implements Serializable {
+
   private static volatile Logger log = 
LogManager.getLogger(HDFSParquetImporter.class);
 
   public static final SimpleDateFormat PARTITION_FORMATTER = new 
SimpleDateFormat("yyyy/MM/dd");
@@ -87,9 +88,14 @@ public class HDFSParquetImporter implements Serializable {
       System.exit(1);
     }
     HDFSParquetImporter dataImporter = new HDFSParquetImporter(cfg);
-    dataImporter
-        .dataImport(UtilHelpers.buildSparkContext("data-importer-" + 
cfg.tableName, cfg.sparkMaster, cfg.sparkMemory),
-            cfg.retry);
+    JavaSparkContext jssc = UtilHelpers
+        .buildSparkContext("data-importer-" + cfg.tableName, cfg.sparkMaster, 
cfg.sparkMemory);
+    try {
+      dataImporter.dataImport(jssc, cfg.retry);
+    } finally {
+      jssc.stop();
+    }
+
   }
 
   public int dataImport(JavaSparkContext jsc, int retry) throws Exception {
diff --git 
a/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java
 
b/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java
index 1951546..1d0aca3 100644
--- 
a/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java
+++ 
b/hoodie-utilities/src/main/java/com/uber/hoodie/utilities/deltastreamer/HoodieDeltaStreamer.java
@@ -280,7 +280,11 @@ public class HoodieDeltaStreamer implements Serializable {
     Map<String, String> additionalSparkConfigs = 
SchedulerConfGenerator.getSparkSchedulingConfigs(cfg);
     JavaSparkContext jssc = UtilHelpers.buildSparkContext("delta-streamer-" + 
cfg.targetTableName,
         cfg.sparkMaster, additionalSparkConfigs);
-    new HoodieDeltaStreamer(cfg, jssc).sync();
+    try {
+      new HoodieDeltaStreamer(cfg, jssc).sync();
+    } finally {
+      jssc.stop();
+    }
   }
 
   /**