Fix NPE when streaming query is started but not yet executed.

Co-authored-by: Szabolcs Vasas <[email protected]>
HeartSaVioR · Dec 15, 2020 · 6ea9024 · 6ea9024
1 parent ad478ac
commit 6ea9024
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/src/main/scala/net/heartsavior/spark/KafkaOffsetCommitterListener.scala b/src/main/scala/net/heartsavior/spark/KafkaOffsetCommitterListener.scala
@@ -38,8 +38,8 @@ class KafkaOffsetCommitterListener extends StreamingQueryListener with Logging {
     val query = SparkSession.active.streams.get(event.progress.id)
     if (query != null) {
       val exec = query match {
-        case query: StreamingQueryWrapper => Some(query.streamingQuery.lastExecution)
-        case query: StreamExecution => Some(query.lastExecution)
+        case query: StreamingQueryWrapper => Option(query.streamingQuery.lastExecution)
+        case query: StreamExecution => Option(query.lastExecution)
         case _ =>
           logWarning(s"Unexpected type of streaming query: ${query.getClass}")
           None