[FLINK-21419][coordination] Remove GC cleaner mechanism for unsafe me…

…mory segments

This closes apache#15058

flink-core/src/main/java/org/apache/flink/core/memory/HybridMemorySegment.java

@@ -60,7 +60,7 @@ public final class HybridMemorySegment extends MemorySegment {
  */
  @Nullable private ByteBuffer offHeapBuffer;
 
- @Nullable private final Runnable cleaner;
+ @Nullable private Runnable cleaner;
 
  /**
  * Wrapping is not allowed when the underlying memory is unsafe. Unsafe memory can be actively
@@ -136,6 +136,7 @@ public void free() {
  cleaner.run();
  }
  offHeapBuffer = null; // to enable GC of unsafe memory
+ cleaner = null;
  }
 
  @Override

flink-core/src/main/java/org/apache/flink/core/memory/MemorySegmentFactory.java

@@ -175,8 +175,7 @@ public static MemorySegment allocateOffHeapUnsafeMemory(
  int size, Object owner, Runnable customCleanupAction) {
  long address = MemoryUtils.allocateUnsafe(size);
  ByteBuffer offHeapBuffer = MemoryUtils.wrapUnsafeMemoryWithByteBuffer(address, size);
- Runnable cleaner =
- MemoryUtils.createMemoryGcCleaner(offHeapBuffer, address, customCleanupAction);
+ Runnable cleaner = MemoryUtils.createMemoryCleaner(address, customCleanupAction);
  return new HybridMemorySegment(offHeapBuffer, owner, false, cleaner);
  }
 

flink-core/src/main/java/org/apache/flink/core/memory/MemoryUtils.java

@@ -19,7 +19,6 @@
 package org.apache.flink.core.memory;
 
 import org.apache.flink.annotation.Internal;
-import org.apache.flink.util.JavaGcCleanerWrapper;
 import org.apache.flink.util.Preconditions;
 
 import java.lang.reflect.Field;
@@ -112,23 +111,17 @@ static long allocateUnsafe(long size) {
  }
 
  /**
- * Creates a cleaner to release the unsafe memory by VM GC.
+ * Creates a cleaner to release the unsafe memory.
  *
- * <p>When memory owner becomes <a href="package-summary.html#reachability">phantom
- * reachable</a>, GC will release the underlying unsafe memory if not released yet.
- *
- * @param owner memory owner which phantom reaching is to monitor by GC and release the unsafe
- * memory
  * @param address address of the unsafe memory to release
+ * @param customCleanup A custom action to clean up GC
  * @return action to run to release the unsafe memory manually
  */
- static Runnable createMemoryGcCleaner(Object owner, long address, Runnable customCleanup) {
- return JavaGcCleanerWrapper.createCleaner(
- owner,
- () -> {
- releaseUnsafe(address);
- customCleanup.run();
- });
+ static Runnable createMemoryCleaner(long address, Runnable customCleanup) {
+ return () -> {
+ releaseUnsafe(address);
+ customCleanup.run();
+ };
  }
 
  private static void releaseUnsafe(long address) {