Duc and Graphex pointed out that there is an error in callback implem…

…entation, now it's fixed. Thanks!
mkuthan · Nov 6, 2017 · 1d2b23f · 1d2b23f
1 parent f616e8d
commit 1d2b23f
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/src/main/scala/org/mkuthan/spark/KafkaDStreamSink.scala b/src/main/scala/org/mkuthan/spark/KafkaDStreamSink.scala
@@ -66,7 +66,8 @@ class KafkaDStreamSinkExceptionHandler extends Callback {
 
   private val lastException = new AtomicReference[Option[Exception]](None)
 
-  override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit = lastException.set(Option(exception))
+  override def onCompletion(metadata: RecordMetadata, exception: Exception): Unit =
+    Option(exception).foreach{ ex => lastException.set(Some(ex))}
 
   def throwExceptionIfAny(): Unit = lastException.getAndSet(None).foreach(ex => throw ex)