Use Future.zip instead of Future.flatMap(for-loop)

zsxwing · zsxwing · commit 46a2cd927eac · 2014-12-25T09:41:42.000+08:00
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala b/streaming/src/main/scala/org/apache/spark/streaming/receiver/ReceivedBlockHandler.scala
@@ -169,10 +169,7 @@ private[streaming] class WriteAheadLogBasedBlockHandler(
     }
 
     // Combine the futures, wait for both to complete, and return the write ahead log segment
-    val combinedFuture = for {
-      _ <- storeInBlockManagerFuture
-      fileSegment <- storeInWriteAheadLogFuture
-    } yield fileSegment
+    val combinedFuture = storeInBlockManagerFuture.zip(storeInWriteAheadLogFuture).map(_._2)
     val segment = Await.result(combinedFuture, blockStoreTimeout)
     WriteAheadLogBasedStoreResult(blockId, segment)
   }

Original file line number	Diff line number	Diff line change
`@@ -169,10 +169,7 @@ private[streaming] class WriteAheadLogBasedBlockHandler(`
`169`	`169`	`}`
`170`	`170`
`171`	`171`	`// Combine the futures, wait for both to complete, and return the write ahead log segment`
`172`		`- val combinedFuture = for {`
`173`		`- _ <- storeInBlockManagerFuture`
`174`		`- fileSegment <- storeInWriteAheadLogFuture`
`175`		`- } yield fileSegment`
	`172`	`+ val combinedFuture = storeInBlockManagerFuture.zip(storeInWriteAheadLogFuture).map(_._2)`
`176`	`173`	`val segment = Await.result(combinedFuture, blockStoreTimeout)`
`177`	`174`	`WriteAheadLogBasedStoreResult(blockId, segment)`
`178`	`175`	`}`