fuyou001 commented on code in PR #10814:
URL: https://github.com/apache/rocketmq/pull/10814#discussion_r3725311622
##########
store/src/main/java/org/apache/rocketmq/store/LmqDispatch.java:
##########
@@ -28,25 +27,53 @@ public class LmqDispatch {
public static void wrapLmqDispatch(MessageStore messageStore, final
MessageExtBrokerInner msg)
throws ConsumeQueueException {
- String lmqNames =
msg.getProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH);
- String[] queueNames = lmqNames.split(MixAll.LMQ_DISPATCH_SEPARATOR);
- Long[] queueOffsets = new Long[queueNames.length];
- if (messageStore.getMessageStoreConfig().isEnableLmq()) {
- for (int i = 0; i < queueNames.length; i++) {
- if (MixAll.isLmq(queueNames[i])) {
- queueOffsets[i] =
messageStore.getQueueStore().getLmqQueueOffset(queueNames[i],
MixAll.LMQ_QUEUE_ID);
- }
+ populateLmqOffsets(messageStore, msg);
+ msg.removeWaitStorePropertyString();
+ }
+
+ static String[] prepareLmqDispatch(MessageStore messageStore, final
MessageExtBrokerInner msg)
+ throws ConsumeQueueException {
+ return populateLmqOffsets(messageStore, msg);
+ }
+
+ static void reinsertWaitStorePropertyForLegacySerialization(final
MessageExtBrokerInner msg) {
+ // Reproduce the legacy remove/reinsert mutation without the discarded
serialization.
+ if
(msg.getProperties().containsKey(MessageConst.PROPERTY_WAIT_STORE_MSG_OK)) {
+ String waitStoreMsgOKValue =
msg.getProperties().remove(MessageConst.PROPERTY_WAIT_STORE_MSG_OK);
+ msg.getProperties().put(MessageConst.PROPERTY_WAIT_STORE_MSG_OK,
waitStoreMsgOKValue);
+ }
+ }
+
+ private static String[] populateLmqOffsets(MessageStore messageStore,
final MessageExtBrokerInner msg)
+ throws ConsumeQueueException {
+ String[] queueNames = parseLmqQueueNames(msg);
+ StringBuilder queueOffsets = new StringBuilder();
+ boolean enableLmq = messageStore.getMessageStoreConfig().isEnableLmq();
+ for (int i = 0; i < queueNames.length; i++) {
+ if (i > 0) {
+ queueOffsets.append(MixAll.LMQ_DISPATCH_SEPARATOR);
+ }
+ if (enableLmq && MixAll.isLmq(queueNames[i])) {
+
queueOffsets.append(messageStore.getQueueStore().getLmqQueueOffset(queueNames[i],
+ MixAll.LMQ_QUEUE_ID));
}
}
- MessageAccessor.putProperty(msg,
MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET,
- StringUtils.join(queueOffsets, MixAll.LMQ_DISPATCH_SEPARATOR));
- msg.removeWaitStorePropertyString();
+ MessageAccessor.putProperty(msg,
MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, queueOffsets.toString());
+ return queueNames;
+ }
+
+ static String[] parseLmqQueueNames(final MessageExtBrokerInner msg) {
Review Comment:
Suggestion (non-blocking): `parseLmqQueueNames` is only used within
`LmqDispatch`. Consider making it `private` to minimize its visibility and
avoid exposing an unnecessary package-private API.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
