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

[ISSUE #9773] Fix core dump when shutdown broker twice

Browse Source 
Change-Id: I31338a675c8af1657d3e2675a5783b1afc33d7ab
This commit is contained in:
qianye
2025-10-27 19:08:52 +08:00
committed by GitHub
parent e262890216
commit bea086feb9
1 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/config/v1/RocksDBConsumerOffsetManager.java
+13 -9
View File
@@ -157,17 +157,21 @@ public class RocksDBConsumerOffsetManager extends ConsumerOffsetManager {
@Override
public synchronized void persist() {
try (WriteBatch writeBatch = new WriteBatch()) {
for (Entry<String, ConcurrentMap<Integer, Long>> entry : this.offsetTable.entrySet()) {
putWriteBatch(writeBatch, entry.getKey(), entry.getValue());
if (writeBatch.getDataSize() >= 4 * 1024) {
this.rocksDBConfigManager.batchPutWithWal(writeBatch);
if (!rocksDBConfigManager.isStop) {
try (WriteBatch writeBatch = new WriteBatch()) {
for (Entry<String, ConcurrentMap<Integer, Long>> entry : this.offsetTable.entrySet()) {
putWriteBatch(writeBatch, entry.getKey(), entry.getValue());
if (writeBatch.getDataSize() >= 4 * 1024) {
this.rocksDBConfigManager.batchPutWithWal(writeBatch);
}
}
this.rocksDBConfigManager.batchPutWithWal(writeBatch);
this.rocksDBConfigManager.flushWAL();
} catch (Exception e) {
log.error("consumer offset persist Failed", e);
}
this.rocksDBConfigManager.batchPutWithWal(writeBatch);
this.rocksDBConfigManager.flushWAL();
} catch (Exception e) {
log.error("consumer offset persist Failed", e);
} else {
log.warn("RocksDBConsumerOffsetManager has been stopped, persist fail");
}
}