Explicitly stop sparkSession on exception (#65)

Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
feast-dev · Apr 28, 2021 · 3929cbd · 3929cbd
1 parent aa68e32
commit 3929cbd
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala b/spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala
@@ -117,7 +117,11 @@ object IngestionJob {
             sparkSession.close()
           case Modes.Online =>
             val sparkSession = BasePipeline.createSparkSession(config)
-            StreamingPipeline.createPipeline(sparkSession, config).get.awaitTermination
+            try {
+              StreamingPipeline.createPipeline(sparkSession, config).get.awaitTermination
+            } finally {
+              sparkSession.close()
+            }
         }
       case None =>
         println("Parameters can't be parsed")