address matei's comment on the replication of offHeap storagelevel

Author: RongGu

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -717,7 +717,7 @@ private[spark] class BlockManager(
 
     // Either we're storing bytes and we asynchronously started replication, or we're storing
     // values and need to serialize and replicate them now:
-    if (level.replication > 1 && !level.useOffHeap) {
+    if (level.replication > 1) {
       data match {
         case ByteBufferValues(bytes) => Await.ready(replicationFuture, Duration.Inf)
         case _ => {
@@ -739,7 +739,7 @@ private[spark] class BlockManager(
 
     BlockManager.dispose(bytesAfterPut)
 
-    if (level.replication > 1 && !level.useOffHeap) {
+    if (level.replication > 1) {
       logDebug("Put for block " + blockId + " with replication took " +
         Utils.getUsedTimeMs(startTimeMs))
     } else {

core/src/main/scala/org/apache/spark/storage/StorageLevel.scala

@@ -50,6 +50,9 @@ class StorageLevel private(
   def replication = replication_
 
   assert(replication < 40, "Replication restricted to be less than 40 for calculating hashcodes")
+  
+  assert(!(useOffHeap && (replication > 1)), 
+    "The replication of useOffHeap mode can not set more than 1")
 
   override def clone(): StorageLevel = new StorageLevel(
     this.useDisk, this.useMemory, this.useOffHeap, this.deserialized, this.replication)