From 12774c64bebf4df45701880b585f7e8c5daebe4e Mon Sep 17 00:00:00 2001 From: sinberCS <42942020+sinberCS@users.noreply.github.com> Date: Mon, 24 Nov 2025 18:00:38 +0800 Subject: [PATCH] [ISSUE #9779] fix DirectBuffer will cause error at 9+ JDK version. (#9801) * fix(DirectBuffer): issue#9779,fix the promblem of 'DirectBuffer will cause error at 9+ JDK version'. Change-Id: I4657ecc401046a3b0d29b466ee68845f45d34105 * fix(DirectBuffer): issue#9779,fix the promblem of 'DirectBuffer will cause error at 9+ JDK version'. Change-Id: Iff0880f694cccfa86b812b81260dfe09e4763fa9 --- .../java/org/apache/rocketmq/store/CommitLog.java | 7 ++++--- .../org/apache/rocketmq/store/TransientStorePool.java | 11 ++++++----- .../rocketmq/store/logfile/DefaultMappedFile.java | 10 ++++++---- 3 files changed, 16 insertions(+), 12 deletions(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java index 29b09bc649..d6ea017218 100644 --- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java +++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java @@ -36,6 +36,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.function.Supplier; import java.util.stream.Collectors; + +import io.netty.util.internal.PlatformDependent; import org.apache.rocketmq.common.MixAll; import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.SystemClock; @@ -69,7 +71,6 @@ import org.apache.rocketmq.store.queue.CqUnit; import org.apache.rocketmq.store.queue.ReferredIterator; import org.apache.rocketmq.store.util.LibC; import org.rocksdb.RocksDBException; -import sun.nio.ch.DirectBuffer; /** * Store all metadata downtime for recovery, data protection reliability @@ -2433,7 +2434,7 @@ public class CommitLog implements Swappable { private byte[] checkFileInPageCache(MappedFile mappedFile) { long fileSize = mappedFile.getFileSize(); - final long address = ((DirectBuffer) mappedFile.getMappedByteBuffer()).address(); + final long address = PlatformDependent.directBufferAddress(mappedFile.getMappedByteBuffer()); int pageNums = (int) (fileSize + this.pageSize - 1) / this.pageSize; byte[] pageCacheRst = new byte[pageNums]; int mincore = LibC.INSTANCE.mincore(new Pointer(address), new NativeLong(fileSize), pageCacheRst); @@ -2509,7 +2510,7 @@ public class CommitLog implements Swappable { log.error("setFileReadMode mappedFile is null"); return -1; } - final long address = ((DirectBuffer) mappedFile.getMappedByteBuffer()).address(); + final long address = PlatformDependent.directBufferAddress(mappedFile.getMappedByteBuffer()); int madvise = LibC.INSTANCE.madvise(new Pointer(address), new NativeLong(mappedFile.getFileSize()), mode); if (madvise != 0) { log.error("setFileReadMode error fileName: {}, madvise: {}, mode:{}", mappedFile.getFileName(), madvise, mode); diff --git a/store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java b/store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java index 0d42ee69e6..d9ad4f4ed1 100644 --- a/store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java +++ b/store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java @@ -16,16 +16,17 @@ */ package org.apache.rocketmq.store; -import com.sun.jna.NativeLong; -import com.sun.jna.Pointer; import java.nio.ByteBuffer; import java.util.Deque; import java.util.concurrent.ConcurrentLinkedDeque; + +import com.sun.jna.NativeLong; +import com.sun.jna.Pointer; +import io.netty.util.internal.PlatformDependent; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.logging.org.slf4j.Logger; import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; import org.apache.rocketmq.store.util.LibC; -import sun.nio.ch.DirectBuffer; public class TransientStorePool { private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME); @@ -48,7 +49,7 @@ public class TransientStorePool { for (int i = 0; i < poolSize; i++) { ByteBuffer byteBuffer = ByteBuffer.allocateDirect(fileSize); - final long address = ((DirectBuffer) byteBuffer).address(); + final long address = PlatformDependent.directBufferAddress(byteBuffer); Pointer pointer = new Pointer(address); LibC.INSTANCE.mlock(pointer, new NativeLong(fileSize)); @@ -58,7 +59,7 @@ public class TransientStorePool { public void destroy() { for (ByteBuffer byteBuffer : availableBuffers) { - final long address = ((DirectBuffer) byteBuffer).address(); + final long address = PlatformDependent.directBufferAddress(byteBuffer); Pointer pointer = new Pointer(address); LibC.INSTANCE.munlock(pointer, new NativeLong(fileSize)); } diff --git a/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java b/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java index b28b7a8aef..fbfffef3bc 100644 --- a/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java +++ b/store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java @@ -37,6 +37,8 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLong; import java.util.function.Consumer; + +import io.netty.util.internal.PlatformDependent; import org.apache.commons.lang3.SystemUtils; import org.apache.rocketmq.common.UtilAll; import org.apache.rocketmq.common.constant.LoggerName; @@ -57,7 +59,7 @@ import org.apache.rocketmq.store.TransientStorePool; import org.apache.rocketmq.store.config.FlushDiskType; import org.apache.rocketmq.store.util.LibC; import sun.misc.Unsafe; -import sun.nio.ch.DirectBuffer; + public class DefaultMappedFile extends AbstractMappedFile { public static final int OS_PAGE_SIZE = 1024 * 4; @@ -914,7 +916,7 @@ public class DefaultMappedFile extends AbstractMappedFile { @Override public void mlock() { final long beginTime = System.currentTimeMillis(); - final long address = ((DirectBuffer) (this.mappedByteBuffer)).address(); + final long address = PlatformDependent.directBufferAddress(this.mappedByteBuffer); Pointer pointer = new Pointer(address); { int ret = LibC.INSTANCE.mlock(pointer, new NativeLong(this.fileSize)); @@ -930,7 +932,7 @@ public class DefaultMappedFile extends AbstractMappedFile { @Override public void munlock() { final long beginTime = System.currentTimeMillis(); - final long address = ((DirectBuffer) (this.mappedByteBuffer)).address(); + final long address = PlatformDependent.directBufferAddress(this.mappedByteBuffer); Pointer pointer = new Pointer(address); int ret = LibC.INSTANCE.munlock(pointer, new NativeLong(this.fileSize)); log.info("munlock {} {} {} ret = {} time consuming = {}", address, this.fileName, this.fileSize, ret, System.currentTimeMillis() - beginTime); @@ -1049,7 +1051,7 @@ public class DefaultMappedFile extends AbstractMappedFile { return true; } try { - long addr = ((DirectBuffer) mappedByteBuffer).address() + position; + long addr = PlatformDependent.directBufferAddress(mappedByteBuffer) + position; return (boolean) IS_LOADED_METHOD.invoke(mappedByteBuffer, mappingAddr(addr), size, pageCount(size)); } catch (Exception e) { log.info("invoke isLoaded0 of file {} error:", file.getAbsolutePath(), e);