[SPARK-13121][STREAMING] java mapWithState mishandles scala Option

Already merged into 1.6 branch, this PR is to commit to master the same change Author: Gabriele Nizzoli <mail@nizzoli.net> Closes #11028 from gabrielenizzoli/patch-1.
apache · Feb 2, 2016 · d0df2ca · d0df2ca
1 parent be5dd88
commit d0df2ca
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala b/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
@@ -221,7 +221,7 @@ object StateSpec {
       mappingFunction: JFunction3[KeyType, Optional[ValueType], State[StateType], MappedType]):
     StateSpec[KeyType, ValueType, StateType, MappedType] = {
     val wrappedFunc = (k: KeyType, v: Option[ValueType], s: State[StateType]) => {
-      mappingFunction.call(k, Optional.ofNullable(v.get), s)
+      mappingFunction.call(k, JavaUtils.optionToOptional(v), s)
     }
     StateSpec.function(wrappedFunc)
   }