opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-17 17:17:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #9779] fix DirectBuffer will cause error at 9+ JDK version. (#9801)

Browse Source 
* 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
This commit is contained in:
sinberCS
2025-11-24 18:00:38 +08:00
committed by GitHub
parent 736f027c9c
commit 12774c64be
3 changed files with 16 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+4 -3
View File
@@ -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);
store/src/main/java/org/apache/rocketmq/store/TransientStorePool.java
+6 -5
View File
@@ -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));
}
store/src/main/java/org/apache/rocketmq/store/logfile/DefaultMappedFile.java
+6 -4
View File
@@ -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);