diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/HeapMemoryAllocator.java b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/HeapMemoryAllocator.java
index 355748238540..cc9cc429643a 100644
--- a/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/HeapMemoryAllocator.java
+++ b/common/unsafe/src/main/java/org/apache/spark/unsafe/memory/HeapMemoryAllocator.java
@@ -56,6 +56,9 @@ public MemoryBlock allocate(long size) throws OutOfMemoryError {
             final MemoryBlock memory = blockReference.get();
             if (memory != null) {
               assert (memory.size() == size);
+              if (MemoryAllocator.MEMORY_DEBUG_FILL_ENABLED) {
+                memory.fill(MemoryAllocator.MEMORY_DEBUG_FILL_CLEAN_VALUE);
+              }
               return memory;
             }
           }
diff --git a/common/unsafe/src/test/java/org/apache/spark/unsafe/PlatformUtilSuite.java b/common/unsafe/src/test/java/org/apache/spark/unsafe/PlatformUtilSuite.java
index 4ae49d82efa2..4b141339ec81 100644
--- a/common/unsafe/src/test/java/org/apache/spark/unsafe/PlatformUtilSuite.java
+++ b/common/unsafe/src/test/java/org/apache/spark/unsafe/PlatformUtilSuite.java
@@ -66,10 +66,21 @@ public void overlappingCopyMemory() {
   public void memoryDebugFillEnabledInTest() {
     Assert.assertTrue(MemoryAllocator.MEMORY_DEBUG_FILL_ENABLED);
     MemoryBlock onheap = MemoryAllocator.HEAP.allocate(1);
-    MemoryBlock offheap = MemoryAllocator.UNSAFE.allocate(1);
     Assert.assertEquals(
       Platform.getByte(onheap.getBaseObject(), onheap.getBaseOffset()),
       MemoryAllocator.MEMORY_DEBUG_FILL_CLEAN_VALUE);
+
+    MemoryBlock onheap1 = MemoryAllocator.HEAP.allocate(1024 * 1024);
+    MemoryAllocator.HEAP.free(onheap1);
+    Assert.assertEquals(
+      Platform.getByte(onheap1.getBaseObject(), onheap1.getBaseOffset()),
+      MemoryAllocator.MEMORY_DEBUG_FILL_FREED_VALUE);
+    MemoryBlock onheap2 = MemoryAllocator.HEAP.allocate(1024 * 1024);
+    Assert.assertEquals(
+      Platform.getByte(onheap2.getBaseObject(), onheap2.getBaseOffset()),
+      MemoryAllocator.MEMORY_DEBUG_FILL_CLEAN_VALUE);
+
+    MemoryBlock offheap = MemoryAllocator.UNSAFE.allocate(1);
     Assert.assertEquals(
       Platform.getByte(offheap.getBaseObject(), offheap.getBaseOffset()),
       MemoryAllocator.MEMORY_DEBUG_FILL_CLEAN_VALUE);