You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "lizhanhui (via GitHub)" <gi...@apache.org> on 2024/10/18 11:08:34 UTC
[PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
lizhanhui opened a new pull request, #8842:
URL: https://github.com/apache/rocketmq/pull/8842
<!-- Please make sure the target branch is right. In most case, the target branch should be `develop`. -->
### Which Issue(s) This PR Fixes
<!-- Please ensure that the related issue has already been created, and [link this pull request to that issue using keywords](<https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword>) to ensure automatic closure. -->
to #8829
### Brief Description
<!-- Write a brief description for your pull request to help the maintainer understand the reasons behind your changes. -->
### How Did You Test This Change?
<!-- In order to ensure the code quality of Apache RocketMQ, we expect every pull request to have undergone thorough testing. -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810410921
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
@fuyou001 This method does not impose further restriction on topic length.
```
int total = iterator.key(keyBuffer);
if (total > keyBufferCapacity) {
keyBufferCapacity = total;
PlatformDependent.freeDirectBuffer(keyBuffer);
keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
continue;
}
```
This code blocks roughly means, try to read key into the pre-allocated direct buffer, if the key length is greater than the pre-allocated key buffer capacity, free the pre-allocated direct buffer(by default, 256 bytes) and re-allocate a larger one and then retry again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810534484
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
I get it
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
I get it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810428500
##########
store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java:
##########
@@ -175,11 +176,11 @@ public PutMessageResult encodeWithoutProperties(MessageExtBrokerInner msgInner)
public PutMessageResult encode(MessageExtBrokerInner msgInner) {
this.byteBuf.clear();
- if (CommitLog.isMultiDispatchMsg(messageStoreConfig, msgInner)) {
+ if (messageStoreConfig.isEnableLmq() && MixAll.topicAllowsLMQ(msgInner.getTopic())) {
Review Comment:
Exactly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "RongtongJin (via GitHub)" <gi...@apache.org>.
RongtongJin commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810013536
##########
store/src/main/java/org/apache/rocketmq/store/MessageExtEncoder.java:
##########
@@ -175,11 +176,11 @@ public PutMessageResult encodeWithoutProperties(MessageExtBrokerInner msgInner)
public PutMessageResult encode(MessageExtBrokerInner msgInner) {
this.byteBuf.clear();
- if (CommitLog.isMultiDispatchMsg(messageStoreConfig, msgInner)) {
+ if (messageStoreConfig.isEnableLmq() && MixAll.topicAllowsLMQ(msgInner.getTopic())) {
Review Comment:
It should use `msgInner.needDispatchLMQ()` instead of `MixAll.topicAllowsLMQ(msgInner.getTopic())`, otherwise non-lmq messages will enter this branch and cause issues during writing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#issuecomment-2424429303
## [Codecov](https://app.codecov.io/gh/apache/rocketmq/pull/8842?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `40.82278%` with `374 lines` in your changes missing coverage. Please review.
> Project coverage is 47.32%. Comparing base [(`1e0b08f`)](https://app.codecov.io/gh/apache/rocketmq/commit/1e0b08f56ab628c9017536eba8509d8987ec6be5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`2a1720d`)](https://app.codecov.io/gh/apache/rocketmq/commit/2a1720d6e9daf454663cd79ec5fe77adc2b04f8e?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
| [Files with missing lines](https://app.codecov.io/gh/apache/rocketmq/pull/8842?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ocketmq/broker/processor/AdminBrokerProcessor.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fbroker%2Fprocessor%2FAdminBrokerProcessor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvcHJvY2Vzc29yL0FkbWluQnJva2VyUHJvY2Vzc29yLmphdmE=) | 37.90% | [70 Missing and 7 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rocketmq/broker/metrics/ConsumerLagCalculator.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fbroker%2Fmetrics%2FConsumerLagCalculator.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvbWV0cmljcy9Db25zdW1lckxhZ0NhbGN1bGF0b3IuamF2YQ==) | 0.00% | [43 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rocketmq/store/queue/RocksDBConsumeQueueStore.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=store%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fstore%2Fqueue%2FRocksDBConsumeQueueStore.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3N0b3JlL3F1ZXVlL1JvY2tzREJDb25zdW1lUXVldWVTdG9yZS5qYXZh) | 48.14% | [34 Missing and 8 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...tieredstore/metrics/TieredStoreMetricsManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=tieredstore%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Ftieredstore%2Fmetrics%2FTieredStoreMetricsManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-dGllcmVkc3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3RpZXJlZHN0b3JlL21ldHJpY3MvVGllcmVkU3RvcmVNZXRyaWNzTWFuYWdlci5qYXZh) | 0.00% | [39 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...mq/store/queue/RocksDBConsumeQueueOffsetTable.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=store%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fstore%2Fqueue%2FRocksDBConsumeQueueOffsetTable.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3N0b3JlL3F1ZXVlL1JvY2tzREJDb25zdW1lUXVldWVPZmZzZXRUYWJsZS5qYXZh) | 69.41% | [14 Missing and 12 partials :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rocketmq/broker/processor/PopMessageProcessor.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fbroker%2Fprocessor%2FPopMessageProcessor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvcHJvY2Vzc29yL1BvcE1lc3NhZ2VQcm9jZXNzb3IuamF2YQ==) | 16.66% | [20 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...che/rocketmq/broker/metrics/PopMetricsManager.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fbroker%2Fmetrics%2FPopMetricsManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-YnJva2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9yb2NrZXRtcS9icm9rZXIvbWV0cmljcy9Qb3BNZXRyaWNzTWFuYWdlci5qYXZh) | 7.69% | [12 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...main/java/org/apache/rocketmq/store/CommitLog.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=store%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fstore%2FCommitLog.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3N0b3JlL0NvbW1pdExvZy5qYXZh) | 45.00% | [10 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ocketmq/store/exception/ConsumeQueueException.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=store%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fstore%2Fexception%2FConsumeQueueException.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3N0b3JlL2V4Y2VwdGlvbi9Db25zdW1lUXVldWVFeGNlcHRpb24uamF2YQ==) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...pache/rocketmq/store/exception/StoreException.java](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&filepath=store%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Frocketmq%2Fstore%2Fexception%2FStoreException.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-c3RvcmUvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL3JvY2tldG1xL3N0b3JlL2V4Y2VwdGlvbi9TdG9yZUV4Y2VwdGlvbi5qYXZh) | 0.00% | [10 Missing :warning: ](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| ... and [21 more](https://app.codecov.io/gh/apache/rocketmq/pull/8842?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## develop #8842 +/- ##
=============================================
- Coverage 47.46% 47.32% -0.15%
+ Complexity 11617 11609 -8
=============================================
Files 1284 1288 +4
Lines 90025 90268 +243
Branches 11585 11609 +24
=============================================
- Hits 42728 42715 -13
- Misses 42061 42281 +220
- Partials 5236 5272 +36
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/rocketmq/pull/8842?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "RongtongJin (via GitHub)" <gi...@apache.org>.
RongtongJin commented on PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#issuecomment-2425463077
@LetLetMe PTAL～
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810534584
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
I get it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "fuyou001 (via GitHub)" <gi...@apache.org>.
fuyou001 commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1809752406
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
The key length of CQ needs to be reconsidered or should have the same restrictions as the file version on both sides
topic v2 topic's lenght may be greater 300
cqKey lenght only 300 Byte
org.apache.rocketmq.store.queue.RocksDBConsumeQueueTable#getCQByteBufferPair
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on code in PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#discussion_r1810410921
##########
store/src/main/java/org/apache/rocketmq/store/queue/RocksDBConsumeQueueOffsetTable.java:
##########
@@ -111,69 +117,147 @@ public class RocksDBConsumeQueueOffsetTable {
/**
* Although we have already put max(min) consumeQueueOffset and physicalOffset in rocksdb, we still hope to get them
* from heap to avoid accessing rocksdb.
+ *
* @see ConsumeQueue#getMaxPhysicOffset(), maxPhysicOffset --> topicQueueMaxCqOffset
* @see ConsumeQueue#getMinLogicOffset(), minLogicOffset --> topicQueueMinOffset
*/
- private final Map<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
- private final Map<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
+ private final ConcurrentMap<String/* topic-queueId */, PhyAndCQOffset> topicQueueMinOffset;
+ private final ConcurrentMap<String/* topic-queueId */, Long> topicQueueMaxCqOffset;
public RocksDBConsumeQueueOffsetTable(RocksDBConsumeQueueTable rocksDBConsumeQueueTable,
ConsumeQueueRocksDBStorage rocksDBStorage, DefaultMessageStore messageStore) {
this.rocksDBConsumeQueueTable = rocksDBConsumeQueueTable;
this.rocksDBStorage = rocksDBStorage;
this.messageStore = messageStore;
- this.topicQueueMinOffset = new ConcurrentHashMap(1024);
- this.topicQueueMaxCqOffset = new ConcurrentHashMap(1024);
+ this.topicQueueMinOffset = new ConcurrentHashMap<>(1024);
+ this.topicQueueMaxCqOffset = new ConcurrentHashMap<>(1024);
this.maxPhyOffsetBB = ByteBuffer.allocateDirect(8);
}
public void load() {
this.offsetCFH = this.rocksDBStorage.getOffsetCFHandle();
+ loadMaxConsumeQueueOffsets();
}
- public void updateTempTopicQueueMaxOffset(final Pair<ByteBuffer, ByteBuffer> offsetBBPair,
- final byte[] topicBytes, final DispatchRequest request,
- final Map<ByteBuffer, Pair<ByteBuffer, DispatchRequest>> tempTopicQueueMaxOffsetMap) {
- buildOffsetKeyAndValueByteBuffer(offsetBBPair, topicBytes, request);
- ByteBuffer topicQueueId = offsetBBPair.getObject1();
- ByteBuffer maxOffsetBB = offsetBBPair.getObject2();
- Pair<ByteBuffer, DispatchRequest> old = tempTopicQueueMaxOffsetMap.get(topicQueueId);
- if (old == null) {
- tempTopicQueueMaxOffsetMap.put(topicQueueId, new Pair(maxOffsetBB, request));
- } else {
- long oldMaxOffset = old.getObject1().getLong(OFFSET_CQ_OFFSET);
- long maxOffset = maxOffsetBB.getLong(OFFSET_CQ_OFFSET);
- if (maxOffset >= oldMaxOffset) {
- ERROR_LOG.error("cqOffset invalid1. old: {}, now: {}", oldMaxOffset, maxOffset);
+ private void loadMaxConsumeQueueOffsets() {
+ 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);
+ };
+ try {
+ forEach(predicate, fn);
+ } catch (RocksDBException e) {
+ log.error("Failed to maximum consume queue offset", e);
+ }
+ }
+
+ public void forEach(Function<OffsetEntry, Boolean> predicate, Consumer<OffsetEntry> fn) throws RocksDBException {
+ try (RocksIterator iterator = this.rocksDBStorage.seekOffsetCF()) {
+ if (null == iterator) {
+ return;
+ }
+
+ int keyBufferCapacity = 256;
+ iterator.seekToFirst();
+ ByteBuffer keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
+ ByteBuffer valueBuffer = ByteBuffer.allocateDirect(16);
+ while (iterator.isValid()) {
+ // parse key buffer according to key layout
+ keyBuffer.clear(); // clear position and limit before reuse
+ int total = iterator.key(keyBuffer);
+ if (total > keyBufferCapacity) {
Review Comment:
@fuyou001 This method does not impose further restriction on topic length.
`
int total = iterator.key(keyBuffer);
if (total > keyBufferCapacity) {
keyBufferCapacity = total;
PlatformDependent.freeDirectBuffer(keyBuffer);
keyBuffer = ByteBuffer.allocateDirect(keyBufferCapacity);
continue;
}
`
This code blocks roughly means, try to read key into the pre-allocated direct buffer, if the key length is greater than the pre-allocated key buffer capacity, free the pre-allocated direct buffer(by default, 256 bytes) and re-allocate a larger one and then retry again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "tianliuliu (via GitHub)" <gi...@apache.org>.
tianliuliu commented on PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#issuecomment-2439866794
LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui merged PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support LMQ dispatch in case if Consume Queue Store is RocksDB-based [rocketmq]
Posted by "lizhanhui (via GitHub)" <gi...@apache.org>.
lizhanhui commented on PR #8842:
URL: https://github.com/apache/rocketmq/pull/8842#issuecomment-2426109091
@redlsz @leizhiyuan Please also take a look
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org