fixup! [FLINK-20663][core] Release unsafe memory instantly on segment…

… freed.

flink-core/src/test/java/org/apache/flink/core/memory/HybridOffHeapUnsafeMemorySegmentTest.java

@@ -22,6 +22,10 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import java.util.concurrent.CompletableFuture;
+
+import static org.junit.Assert.assertTrue;
+
 /** Tests for the {@link HybridMemorySegment} in off-heap mode using unsafe memory. */
 @RunWith(Parameterized.class)
 public class HybridOffHeapUnsafeMemorySegmentTest extends MemorySegmentTestBase {
@@ -45,4 +49,13 @@ MemorySegment createSegment(int size, Object owner) {
     public void testByteBufferWrapping() {
         createSegment(10).wrap(1, 2);
     }
+
+    @Test
+    public void testCallCleanerOnFree() {
+        final CompletableFuture<Void> cleanerFuture = new CompletableFuture<>();
+        MemorySegmentFactory.allocateOffHeapUnsafeMemory(
+                        10, null, () -> cleanerFuture.complete(null))
+                .free();
+        assertTrue(cleanerFuture.isDone());
+    }
 }