From e48d68d1f3777585e8c9ecd298f4c240a4255d39 Mon Sep 17 00:00:00 2001 From: LetLetMe Date: Mon, 9 Sep 2024 16:42:04 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8Dpr=E4=B8=AD=E6=8F=90=E5=87=BA?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../apache/rocketmq/store/RocksDBMessageStore.java | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java b/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java index edb68894853..90df7aed596 100644 --- a/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java +++ b/store/src/main/java/org/apache/rocketmq/store/RocksDBMessageStore.java @@ -25,6 +25,7 @@ import org.apache.rocketmq.common.BrokerConfig; import org.apache.rocketmq.common.TopicConfig; import org.apache.rocketmq.common.UtilAll; +import org.apache.rocketmq.common.sysflag.MessageSysFlag; import org.apache.rocketmq.store.config.MessageStoreConfig; import org.apache.rocketmq.store.config.StorePathConfigHelper; import org.apache.rocketmq.store.metrics.DefaultStoreMetricsManager; @@ -187,7 +188,16 @@ public CommitLogDispatcherBuildRocksdbConsumeQueue getDispatcherBuildRocksdbCons class CommitLogDispatcherBuildRocksdbConsumeQueue implements CommitLogDispatcher { @Override public void dispatch(DispatchRequest request) throws RocksDBException { - putMessagePositionInfo(request); + final int tranType = MessageSysFlag.getTransactionValue(request.getSysFlag()); + switch (tranType) { + case MessageSysFlag.TRANSACTION_NOT_TYPE: + case MessageSysFlag.TRANSACTION_COMMIT_TYPE: + putMessagePositionInfo(request); + break; + case MessageSysFlag.TRANSACTION_PREPARED_TYPE: + case MessageSysFlag.TRANSACTION_ROLLBACK_TYPE: + break; + } } }