You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by zs...@apache.org on 2016/02/02 19:57:23 UTC
spark git commit: [SPARK-13121][STREAMING] java mapWithState
mishandles scala Option
Repository: spark
Updated Branches:
refs/heads/branch-1.6 53f518a6e -> 4c28b4c8f
[SPARK-13121][STREAMING] java mapWithState mishandles scala Option
java mapwithstate with Function3 has wrong conversion of java `Optional` to scala `Option`, fixed code uses same conversion used in the mapwithstate call that uses Function4 as an input. `Optional.fromNullable(v.get)` fails if v is `None`, better to use `JavaUtils.optionToOptional(v)` instead.
Author: Gabriele Nizzoli <ma...@nizzoli.net>
Closes #11007 from gabrielenizzoli/branch-1.6.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/4c28b4c8
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/4c28b4c8
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/4c28b4c8
Branch: refs/heads/branch-1.6
Commit: 4c28b4c8f342fde937ff77ab30f898dfe3186c03
Parents: 53f518a
Author: Gabriele Nizzoli <ma...@nizzoli.net>
Authored: Tue Feb 2 10:57:18 2016 -0800
Committer: Shixiong Zhu <sh...@databricks.com>
Committed: Tue Feb 2 10:57:18 2016 -0800
----------------------------------------------------------------------
.../src/main/scala/org/apache/spark/streaming/StateSpec.scala | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/4c28b4c8/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
----------------------------------------------------------------------
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala b/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
index 9f6f952..017d451 100644
--- a/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
+++ b/streaming/src/main/scala/org/apache/spark/streaming/StateSpec.scala
@@ -219,7 +219,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.fromNullable(v.get), s)
+ mappingFunction.call(k, JavaUtils.optionToOptional(v), s)
}
StateSpec.function(wrappedFunc)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org