Added warning

tdas · tdas · commit 75909377cb28 · 2015-08-24T22:55:01.000-07:00
diff --git a/streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala b/streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala
@@ -119,6 +119,11 @@ abstract class ReceiverInputDStream[T: ClassTag](@transient ssc_ : StreamingCont
         val validBlockIds = blockIds.filter { id =>
           ssc.sparkContext.env.blockManager.master.contains(id)
         }
+        if (validBlockIds.size != blockIds.size) {
+          logWarning("Some blocks could not be recovered as they were not found in memory. " +
+            "To prevent such data loss, enabled Write Ahead Log (see programming guide " +
+            "for more details.") 
+        }
         new BlockRDD[T](ssc.sc, validBlockIds)
       }
     } else {

Original file line number	Diff line number	Diff line change
`@@ -119,6 +119,11 @@ abstract class ReceiverInputDStream[T: ClassTag](@transient ssc_ : StreamingCont`
`119`	`119`	`val validBlockIds = blockIds.filter { id =>`
`120`	`120`	`ssc.sparkContext.env.blockManager.master.contains(id)`
`121`	`121`	`}`
	`122`	`+ if (validBlockIds.size != blockIds.size) {`
	`123`	`+ logWarning("Some blocks could not be recovered as they were not found in memory. " +`
	`124`	`+ "To prevent such data loss, enabled Write Ahead Log (see programming guide " +`
	`125`	`+ "for more details.")`
	`126`	`+ }`
`122`	`127`	`new BlockRDD[T](ssc.sc, validBlockIds)`
`123`	`128`	`}`
`124`	`129`	`} else {`