From 28b61a2a85ab58460e25d20aa8968a497796aae1 Mon Sep 17 00:00:00 2001
From: yyki <1721277082@qq.com>
Date: Tue, 23 Jun 2026 14:09:58 +0800
Subject: [PATCH] FLINK-39968 [Optimization] Reduce redundant AtomicBoolean
 allocation in MemorySegment to improve construction performance 1 Replace the
 member variable final AtomicBoolean isFreedAtomic with a primitive boolean
 isFreed; 2 Add the synchronized modifier to the free() method to ensure the
 atomicity and thread safety of the free state judgment and modification;

---
 .../org/apache/flink/core/memory/MemorySegment.java    | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
index 93e81fc5be8152..f207fb3d2c01d9 100644
--- a/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
+++ b/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java
@@ -33,7 +33,6 @@
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.ReadOnlyBufferException;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
@@ -136,7 +135,7 @@ public final class MemorySegment {
      */
     private final boolean allowWrap;
 
-    private final AtomicBoolean isFreedAtomic;
+    private boolean isFreed = false;
 
     /**
      * Creates a new memory segment that represents the memory of the byte array.
@@ -158,7 +157,6 @@ public final class MemorySegment {
         this.owner = owner;
         this.allowWrap = true;
         this.cleaner = null;
-        this.isFreedAtomic = new AtomicBoolean(false);
     }
 
     /**
@@ -204,7 +202,6 @@ public final class MemorySegment {
         this.owner = owner;
         this.allowWrap = allowWrap;
         this.cleaner = cleaner;
-        this.isFreedAtomic = new AtomicBoolean(false);
     }
 
     // ------------------------------------------------------------------------
@@ -237,13 +234,14 @@ public boolean isFreed() {
      * segment and will fail. The actual memory (heap or off-heap) will only be released after this
      * memory segment object has become garbage collected.
      */
-    public void free() {
-        if (isFreedAtomic.getAndSet(true)) {
+    public synchronized void free() {
+        if (isFreed) {
             // the segment has already been freed
             if (checkMultipleFree) {
                 throw new IllegalStateException("MemorySegment can be freed only once!");
             }
         } else {
+            isFreed = true;
             // this ensures we can place no more data and trigger
             // the checks for the freed segment
             address = addressLimit + 1;