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

[ISSUE #8895] Fix NPE when broker shutdown and optimize the log #9094

Browse Source
This commit is contained in:
qianye
2025-01-14 14:50:56 +08:00
committed by GitHub
parent b30afe8184
commit 94d9185f2c
3 changed files with 12 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
store/src/main/java/org/apache/rocketmq/store/DefaultMessageStore.java
+1 -3
View File
@@ -517,11 +517,9 @@ public class DefaultMessageStore implements MessageStore {
if (this.compactionService != null) {
this.compactionService.shutdown();
}
if (messageStoreConfig.isRocksdbCQDoubleWriteEnable()) {
if (messageStoreConfig.isRocksdbCQDoubleWriteEnable() && this.rocksDBMessageStore != null) {
this.rocksDBMessageStore.consumeQueueStore.shutdown();
}
this.flushConsumeQueueService.shutdown();
this.allocateMappedFileService.shutdown();
this.storeCheckpoint.flush();
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java
+1 -1
View File
@@ -144,7 +144,7 @@ public class RocksDBConsumeQueueOffsetTable {
Function<OffsetEntry, Boolean> predicate = entry -> entry.type == OffsetEntryType.MAXIMUM;
Consumer<OffsetEntry> fn = entry -> {
topicQueueMaxCqOffset.putIfAbsent(entry.topic + "-" + entry.queueId, entry.offset);
ROCKSDB_LOG.info("Max {}:{} --> {}|{}", entry.topic, entry.queueId, entry.offset, entry.commitLogOffset);
log.info("LoadMaxConsumeQueueOffsets Max {}:{} --> {}|{}", entry.topic, entry.queueId, entry.offset, entry.commitLogOffset);
};
try {
forEach(predicate, fn);