Update CheckpointSuite.scala

marmbrus · Aug 28, 2014 · 17be9fb · 17be9fb
1 parent 714bda5
commit 17be9fb
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala b/streaming/src/test/scala/org/apache/spark/streaming/CheckpointSuite.scala
@@ -70,7 +70,7 @@ class CheckpointSuite extends TestSuiteBase {
     val input = (1 to 10).map(_ => Seq("a")).toSeq
     val operation = (st: DStream[String]) => {
       val updateFunc = (values: Seq[Int], state: Option[Int]) => {
-        Some((values.foldLeft(0)(_ + _) + state.getOrElse(0)))
+        Some((values.sum + state.getOrElse(0)))
       }
       st.map(x => (x, 1))
       .updateStateByKey(updateFunc)
@@ -214,7 +214,7 @@ class CheckpointSuite extends TestSuiteBase {
     val output = (1 to 10).map(x => Seq(("a", x))).toSeq
     val operation = (st: DStream[String]) => {
       val updateFunc = (values: Seq[Int], state: Option[Int]) => {
-        Some((values.foldLeft(0)(_ + _) + state.getOrElse(0)))
+        Some((values.sum + state.getOrElse(0)))
       }
       st.map(x => (x, 1))
         .updateStateByKey(updateFunc)