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

[ISSUE #9891] Optimize pop orderly implementation to facilitate expansion (#9892)

Browse Source 
Change-Id: Ie2faff9942027a510c4550a9f2bf2df8c6283137
This commit is contained in:
qianye
2025-12-02 16:30:15 +08:00
committed by GitHub
parent 63d20eb92a
commit bca6393261
12 changed files with 296 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
+26 -26
View File
@@ -18,6 +18,29 @@ package org.apache.rocketmq.broker;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import org.apache.rocketmq.auth.authentication.factory.AuthenticationFactory;
import org.apache.rocketmq.auth.authentication.manager.AuthenticationMetadataManager;
import org.apache.rocketmq.auth.authorization.factory.AuthorizationFactory;
@@ -58,11 +81,12 @@ import org.apache.rocketmq.broker.mqtrace.ConsumeMessageHook;
import org.apache.rocketmq.broker.mqtrace.SendMessageHook;
import org.apache.rocketmq.broker.offset.BroadcastOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.offset.LmqConsumerOffsetManager;
import org.apache.rocketmq.broker.out.BrokerOuterAPI;
import org.apache.rocketmq.broker.plugin.BrokerAttachedPlugin;
import org.apache.rocketmq.broker.pop.PopConsumerService;
import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.pop.orderly.QueueLevelConsumerManager;
import org.apache.rocketmq.broker.processor.AckMessageProcessor;
import org.apache.rocketmq.broker.processor.AdminBrokerProcessor;
import org.apache.rocketmq.broker.processor.ChangeInvisibleTimeProcessor;
@@ -156,30 +180,6 @@ import org.apache.rocketmq.store.timer.TimerCheckpoint;
import org.apache.rocketmq.store.timer.TimerMessageStore;
import org.apache.rocketmq.store.timer.TimerMetrics;
import java.net.InetSocketAddress;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
public class BrokerController {
protected static final Logger LOG = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final Logger LOG_PROTECTION = LoggerFactory.getLogger(LoggerName.PROTECTION_LOGGER_NAME);
@@ -387,7 +387,7 @@ public class BrokerController {
this.consumerManager = new ConsumerManager(this.consumerIdsChangeListener, this.brokerStatsManager, this.brokerConfig);
this.producerManager = new ProducerManager(this.brokerStatsManager);
this.consumerFilterManager = new ConsumerFilterManager(this);
this.consumerOrderInfoManager = new ConsumerOrderInfoManager(this);
this.consumerOrderInfoManager = new QueueLevelConsumerManager(this);
this.popInflightMessageCounter = new PopInflightMessageCounter(this);
this.popConsumerService = brokerConfig.isPopConsumerKVServiceInit() ? new PopConsumerService(this) : null;
this.clientHousekeepingService = new ClientHousekeepingService(this);
broker/src/main/java/org/apache/rocketmq/broker/pop/PopConsumerService.java
+3 -3
View File
@@ -167,7 +167,7 @@ public class PopConsumerService extends ServiceThread {
if (GetMessageStatus.FOUND.equals(result.getStatus()) && !result.getMessageQueueOffset().isEmpty()) {
if (context.isFifo()) {
this.setFifoBlocked(context, context.getGroupId(), topicId, queueId, result.getMessageQueueOffset());
this.setFifoBlocked(context, context.getGroupId(), topicId, queueId, result.getMessageQueueOffset(), result);
}
// build response header here
context.addGetMessageResult(result, topicId, queueId, retryType, offset);
@@ -275,10 +275,10 @@ public class PopConsumerService extends ServiceThread {
* Fifo message does not have retry feature in broker
*/
public void setFifoBlocked(PopConsumerContext context,
String groupId, String topicId, int queueId, List<Long> queueOffsetList) {
String groupId, String topicId, int queueId, List<Long> queueOffsetList, GetMessageResult getMessageResult) {
brokerController.getConsumerOrderInfoManager().update(
context.getAttemptId(), false, topicId, groupId, queueId,
context.getPopTime(), context.getInvisibleTime(), queueOffsetList, context.getOrderCountInfoBuilder());
context.getPopTime(), context.getInvisibleTime(), queueOffsetList, context.getOrderCountInfoBuilder(), getMessageResult);
}
public boolean isFifoBlocked(PopConsumerContext context, String groupId, String topicId, int queueId) {
broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManager.java
+142
View File
@@ -0,0 +1,142 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.broker.pop.orderly;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.rocketmq.common.OrderedConsumptionLevel;
import org.apache.rocketmq.store.GetMessageResult;
/**
*
* Ordered Consumption Controller Interface
* This is the top-level interface that encapsulates complete ordered consumption management functionality,
* supporting different concurrency strategy implementations
* <p>
* Design Goals:
* 1. Support queue-level ordered consumption (existing implementation)
* 2. Support message group-level ordered consumption (improve concurrency)
* 3. Support custom ordered consumption strategies
* </p>
*/
public interface ConsumerOrderInfoManager {
/**
* Update the reception status of message list
* Called by handleGetMessageResult when consumer POPs messages, used to record message status and build consumption information
*
* @param attemptId Distinguish different pop requests
* @param isRetry Whether it is a retry topic
* @param topic Topic name
* @param group Consumer group name
* @param queueId Queue ID
* @param popTime Time when messages are popped
* @param invisibleTime Message invisible time
* @param msgQueueOffsetList List of message queue offsets
* @param orderInfoBuilder String builder for constructing order information
* @param getMessageResult Return new result
*/
void update(String attemptId, boolean isRetry, String topic, String group, int queueId,
long popTime, long invisibleTime, List<Long> msgQueueOffsetList,
StringBuilder orderInfoBuilder, GetMessageResult getMessageResult);
/**
* Check whether the current POP request needs to be blocked
* Used to ensure ordered consumption of ordered messages
* Called when consumer POPs messages
*
* @param attemptId Attempt ID
* @param topic Topic name
* @param group Consumer group name
* @param queueId Queue ID
* @param invisibleTime Invisible time
* @return true indicates blocking is needed, false indicates can proceed
*/
boolean checkBlock(String attemptId, String topic, String group, int queueId, long invisibleTime);
/**
* Commit message and calculate next consumption offset
* Called when consumer ACKs messages
*
* @param topic Topic name
* @param group Consumer group name
* @param queueId Queue ID
* @param queueOffset Message queue offset
* @param popTime Pop time, used for validation
* @return -1: invalid, -2: no need to commit, >=0: offset that needs to be committed (indicates messages below this offset have been consumed)
*/
long commitAndNext(String topic, String group, int queueId, long queueOffset, long popTime);
/**
* Update the next visible time of message
* Used for delayed message re-consumption
*
* @param topic Topic name
* @param group Consumer group name
* @param queueId Queue ID
* @param queueOffset Message offset
* @param popTime Pop time, used for validation
* @param nextVisibleTime Next visible time
*/
void updateNextVisibleTime(String topic, String group, int queueId, long queueOffset,
long popTime, long nextVisibleTime);
/**
* Clear the blocking status of specified queue
* Usually called during consumer rebalancing or queue reassignment
*
* @param topic Topic name
* @param group Consumer group name
* @param queueId Queue ID
*/
void clearBlock(String topic, String group, int queueId);
/**
* Get ordered consumption level
* Used to distinguish different implementation strategies
*
* @return Ordered consumption level, such as: QUEUE, MESSAGE_GROUP, etc.
*/
OrderedConsumptionLevel getOrderedConsumptionLevel();
/**
* Start the controller
* Initialize necessary resources, such as timers, thread pools, etc.
*/
void start();
/**
* Shutdown the controller
* Release resources, clean up scheduled tasks, etc.
*/
void shutdown();
/**
* Persist the controller
* Persist the controller's data
*/
void persist();
boolean load();
/**
* Get available message result
* Used to retrieve messages from cache
*/
CompletableFuture<GetMessageResult> getAvailableMessageResult(String attemptId, long popTime, long invisibleTime, String groupId,
String topicId, int queueId, int batchSize, StringBuilder orderCountInfoBuilder);
}
broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOrderInfoManager.java → broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerManager.java
+54 -20
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.broker.offset;
package org.apache.rocketmq.broker.pop.orderly;
import com.alibaba.fastjson.annotation.JSONField;
import com.google.common.annotations.VisibleForTesting;
@@ -26,18 +26,21 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.BrokerPathConfigHelper;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.OrderedConsumptionLevel;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.store.GetMessageResult;
public class ConsumerOrderInfoManager extends ConfigManager {
public class QueueLevelConsumerManager extends ConfigManager implements ConsumerOrderInfoManager {
private static final Logger log = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);
private static final String TOPIC_GROUP_SEPARATOR = "@";
@@ -46,15 +49,15 @@ public class ConsumerOrderInfoManager extends ConfigManager {
private ConcurrentHashMap<String/* topic@group*/, ConcurrentHashMap<Integer/*queueId*/, OrderInfo>> table =
new ConcurrentHashMap<>(128);
private transient ConsumerOrderInfoLockManager consumerOrderInfoLockManager;
private transient QueueLevelConsumerOrderInfoLockManager queueLevelConsumerOrderInfoLockManager;
private transient BrokerController brokerController;
public ConsumerOrderInfoManager() {
public QueueLevelConsumerManager() {
}
public ConsumerOrderInfoManager(BrokerController brokerController) {
public QueueLevelConsumerManager(BrokerController brokerController) {
this.brokerController = brokerController;
this.consumerOrderInfoLockManager = new ConsumerOrderInfoLockManager(brokerController);
this.queueLevelConsumerOrderInfoLockManager = new QueueLevelConsumerOrderInfoLockManager(brokerController);
}
public ConcurrentHashMap<String, ConcurrentHashMap<Integer, OrderInfo>> getTable() {
@@ -74,8 +77,8 @@ public class ConsumerOrderInfoManager extends ConfigManager {
}
private void updateLockFreeTimestamp(String topic, String group, int queueId, OrderInfo orderInfo) {
if (consumerOrderInfoLockManager != null) {
consumerOrderInfoLockManager.updateLockFreeTimestamp(topic, group, queueId, orderInfo);
if (queueLevelConsumerOrderInfoLockManager != null) {
queueLevelConsumerOrderInfoLockManager.updateLockFreeTimestamp(topic, group, queueId, orderInfo);
}
}
@@ -91,7 +94,8 @@ public class ConsumerOrderInfoManager extends ConfigManager {
* @param msgQueueOffsetList the queue offsets of messages
* @param orderInfoBuilder will append order info to this builder
*/
public void update(String attemptId, boolean isRetry, String topic, String group, int queueId, long popTime, long invisibleTime,
public void update(String attemptId, boolean isRetry, String topic, String group, int queueId, long popTime,
long invisibleTime,
List<Long> msgQueueOffsetList, StringBuilder orderInfoBuilder) {
String key = buildKey(topic, group);
ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);
@@ -140,6 +144,14 @@ public class ConsumerOrderInfoManager extends ConfigManager {
updateLockFreeTimestamp(topic, group, queueId, orderInfo);
}
@Override
public void update(String attemptId, boolean isRetry, String topic, String group, int queueId, long popTime,
long invisibleTime,
List<Long> msgQueueOffsetList, StringBuilder orderInfoBuilder, GetMessageResult getMessageResult) {
update(attemptId, isRetry, topic, group, queueId, popTime, invisibleTime, msgQueueOffsetList, orderInfoBuilder);
}
@Override
public boolean checkBlock(String attemptId, String topic, String group, int queueId, long invisibleTime) {
String key = buildKey(topic, group);
ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);
@@ -159,6 +171,7 @@ public class ConsumerOrderInfoManager extends ConfigManager {
return orderInfo.needBlock(attemptId, invisibleTime);
}
@Override
public void clearBlock(String topic, String group, int queueId) {
table.computeIfPresent(buildKey(topic, group), (key, val) -> {
val.remove(queueId);
@@ -166,6 +179,15 @@ public class ConsumerOrderInfoManager extends ConfigManager {
});
}
@Override
public OrderedConsumptionLevel getOrderedConsumptionLevel() {
return OrderedConsumptionLevel.QUEUE;
}
@Override
public void start() {
}
/**
* mark message is consumed finished. return the consumer offset
*
@@ -175,6 +197,7 @@ public class ConsumerOrderInfoManager extends ConfigManager {
* @param queueOffset queue offset of message
* @return -1 : illegal, -2 : no need commit, >= 0 : commit
*/
@Override
public long commitAndNext(String topic, String group, int queueId, long queueOffset, long popTime) {
String key = buildKey(topic, group);
ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);
@@ -234,7 +257,9 @@ public class ConsumerOrderInfoManager extends ConfigManager {
* @param queueOffset queue offset of message
* @param nextVisibleTime nex visible time
*/
public void updateNextVisibleTime(String topic, String group, int queueId, long queueOffset, long popTime, long nextVisibleTime) {
@Override
public void updateNextVisibleTime(String topic, String group, int queueId, long queueOffset, long popTime,
long nextVisibleTime) {
String key = buildKey(topic, group);
ConcurrentHashMap<Integer/*queueId*/, OrderInfo> qs = table.get(key);
@@ -256,6 +281,7 @@ public class ConsumerOrderInfoManager extends ConfigManager {
updateLockFreeTimestamp(topic, group, queueId, orderInfo);
}
@VisibleForTesting
protected void autoClean() {
if (brokerController == null) {
return;
@@ -328,11 +354,11 @@ public class ConsumerOrderInfoManager extends ConfigManager {
@Override
public void decode(String jsonString) {
if (jsonString != null) {
ConsumerOrderInfoManager obj = RemotingSerializable.fromJson(jsonString, ConsumerOrderInfoManager.class);
QueueLevelConsumerManager obj = RemotingSerializable.fromJson(jsonString, QueueLevelConsumerManager.class);
if (obj != null) {
this.table = obj.table;
if (this.consumerOrderInfoLockManager != null) {
this.consumerOrderInfoLockManager.recover(this.table);
if (this.queueLevelConsumerOrderInfoLockManager != null) {
this.queueLevelConsumerOrderInfoLockManager.recover(this.table);
}
}
}
@@ -345,14 +371,20 @@ public class ConsumerOrderInfoManager extends ConfigManager {
}
public void shutdown() {
if (this.consumerOrderInfoLockManager != null) {
this.consumerOrderInfoLockManager.shutdown();
if (this.queueLevelConsumerOrderInfoLockManager != null) {
this.queueLevelConsumerOrderInfoLockManager.shutdown();
}
}
@Override
public CompletableFuture<GetMessageResult> getAvailableMessageResult(String attemptId, long popTime, long invisibleTime,
String groupId, String topicId, int queueId, int batchSize, StringBuilder orderCountInfoBuilder) {
return CompletableFuture.completedFuture(null);
}
@VisibleForTesting
protected ConsumerOrderInfoLockManager getConsumerOrderInfoLockManager() {
return consumerOrderInfoLockManager;
QueueLevelConsumerOrderInfoLockManager getConsumerOrderInfoLockManager() {
return queueLevelConsumerOrderInfoLockManager;
}
public static class OrderInfo {
@@ -397,7 +429,8 @@ public class ConsumerOrderInfoManager extends ConfigManager {
public OrderInfo() {
}
public OrderInfo(String attemptId, long popTime, long invisibleTime, List<Long> queueOffsetList, long lastConsumeTimestamp,
public OrderInfo(String attemptId, long popTime, long invisibleTime, List<Long> queueOffsetList,
long lastConsumeTimestamp,
long commitOffsetBit) {
this.popTime = popTime;
this.invisibleTime = invisibleTime;
@@ -600,7 +633,8 @@ public class ConsumerOrderInfoManager extends ConfigManager {
* @param prevOffsetConsumedCount the offset list of message
*/
@JSONField(serialize = false, deserialize = false)
public void mergeOffsetConsumedCount(String preAttemptId, List<Long> preOffsetList, Map<Long, Integer> prevOffsetConsumedCount) {
public void mergeOffsetConsumedCount(String preAttemptId, List<Long> preOffsetList,
Map<Long, Integer> prevOffsetConsumedCount) {
Map<Long, Integer> offsetConsumedCount = new HashMap<>();
if (prevOffsetConsumedCount == null) {
prevOffsetConsumedCount = new HashMap<>();
@@ -641,4 +675,4 @@ public class ConsumerOrderInfoManager extends ConfigManager {
.toString();
}
}
}
}
broker/src/main/java/org/apache/rocketmq/broker/offset/ConsumerOrderInfoLockManager.java → broker/src/main/java/org/apache/rocketmq/broker/pop/orderly/QueueLevelConsumerOrderInfoLockManager.java
+12 -10
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.broker.offset;
package org.apache.rocketmq.broker.pop.orderly;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
@@ -32,14 +32,16 @@ import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
public class ConsumerOrderInfoLockManager {
public class QueueLevelConsumerOrderInfoLockManager {
private static final Logger POP_LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LOGGER_NAME);
private ConsumerOrderInfoManager consumerOrderInfoManager;
private final BrokerController brokerController;
private final Map<Key, Timeout> timeoutMap = new ConcurrentHashMap<>();
private final Timer timer;
private static final int TIMER_TICK_MS = 100;
public ConsumerOrderInfoLockManager(BrokerController brokerController) {
public QueueLevelConsumerOrderInfoLockManager(BrokerController brokerController) {
this.brokerController = brokerController;
this.timer = new HashedWheelTimer(
new ThreadFactoryImpl("ConsumerOrderInfoLockManager_"),
@@ -47,22 +49,22 @@ public class ConsumerOrderInfoLockManager {
}
/**
* when ConsumerOrderInfoManager load from disk, recover data
* when QueueLevelConsumerManager load from disk, recover data
*/
public void recover(Map<String/* topic@group*/, ConcurrentHashMap<Integer/*queueId*/, ConsumerOrderInfoManager.OrderInfo>> table) {
public void recover(Map<String/* topic@group*/, ConcurrentHashMap<Integer/*queueId*/, QueueLevelConsumerManager.OrderInfo>> table) {
if (!this.brokerController.getBrokerConfig().isEnableNotifyAfterPopOrderLockRelease()) {
return;
}
for (Map.Entry<String, ConcurrentHashMap<Integer, ConsumerOrderInfoManager.OrderInfo>> entry : table.entrySet()) {
for (Map.Entry<String, ConcurrentHashMap<Integer, QueueLevelConsumerManager.OrderInfo>> entry : table.entrySet()) {
String topicAtGroup = entry.getKey();
ConcurrentHashMap<Integer/*queueId*/, ConsumerOrderInfoManager.OrderInfo> qs = entry.getValue();
String[] arrays = ConsumerOrderInfoManager.decodeKey(topicAtGroup);
ConcurrentHashMap<Integer/*queueId*/, QueueLevelConsumerManager.OrderInfo> qs = entry.getValue();
String[] arrays = QueueLevelConsumerManager.decodeKey(topicAtGroup);
if (arrays.length != 2) {
continue;
}
String topic = arrays[0];
String group = arrays[1];
for (Map.Entry<Integer, ConsumerOrderInfoManager.OrderInfo> qsEntry : qs.entrySet()) {
for (Map.Entry<Integer, QueueLevelConsumerManager.OrderInfo> qsEntry : qs.entrySet()) {
Long lockFreeTimestamp = qsEntry.getValue().getLockFreeTimestamp();
if (lockFreeTimestamp == null || lockFreeTimestamp <= System.currentTimeMillis()) {
continue;
@@ -72,7 +74,7 @@ public class ConsumerOrderInfoLockManager {
}
}
public void updateLockFreeTimestamp(String topic, String group, int queueId, ConsumerOrderInfoManager.OrderInfo orderInfo) {
public void updateLockFreeTimestamp(String topic, String group, int queueId, QueueLevelConsumerManager.OrderInfo orderInfo) {
this.updateLockFreeTimestamp(topic, group, queueId, orderInfo.getLockFreeTimestamp());
}
broker/src/main/java/org/apache/rocketmq/broker/processor/AckMessageProcessor.java
+2 -2
View File
@@ -19,12 +19,12 @@ package org.apache.rocketmq.broker.processor;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.util.BitSet;
import java.nio.charset.StandardCharsets;
import java.util.BitSet;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.pop.PopConsumerLockService;
import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
import org.apache.rocketmq.common.KeyBuilder;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.TopicConfig;
broker/src/main/java/org/apache/rocketmq/broker/processor/ChangeInvisibleTimeProcessor.java
+3 -3
View File
@@ -19,13 +19,13 @@ package org.apache.rocketmq.broker.processor;
import com.alibaba.fastjson.JSON;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.nio.charset.StandardCharsets;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.pop.PopConsumerLockService;
import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
import org.apache.rocketmq.common.PopAckConstants;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
@@ -37,11 +37,11 @@ import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader;
import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
import org.apache.rocketmq.store.PutMessageStatus;
broker/src/main/java/org/apache/rocketmq/broker/processor/PopMessageProcessor.java
+1 -2
View File
@@ -44,7 +44,6 @@ import org.apache.rocketmq.broker.longpolling.PollingHeader;
import org.apache.rocketmq.broker.longpolling.PollingResult;
import org.apache.rocketmq.broker.longpolling.PopLongPollingService;
import org.apache.rocketmq.broker.longpolling.PopRequest;
import org.apache.rocketmq.broker.pagecache.ManyMessageTransfer;
import org.apache.rocketmq.broker.pop.PopConsumerContext;
import org.apache.rocketmq.common.BrokerConfig;
@@ -806,7 +805,7 @@ public class PopMessageProcessor implements NettyRequestProcessor {
this.brokerController.getConsumerOrderInfoManager().update(requestHeader.getAttemptId(), isRetry, topic,
requestHeader.getConsumerGroup(),
queueId, popTime, requestHeader.getInvisibleTime(), result.getMessageQueueOffset(),
orderCountInfo);
orderCountInfo, result);
this.brokerController.getConsumerOffsetManager().commitOffset(channel.remoteAddress().toString(),
requestHeader.getConsumerGroup(), topic, queueId, finalOffset);
} else {
broker/src/test/java/org/apache/rocketmq/broker/pop/PopConsumerServiceTest.java
+2 -2
View File
@@ -35,7 +35,7 @@ import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.failover.EscapeBridge;
import org.apache.rocketmq.broker.longpolling.PopLongPollingService;
import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
import org.apache.rocketmq.broker.offset.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.processor.PopMessageProcessor;
import org.apache.rocketmq.broker.topic.TopicConfigManager;
import org.apache.rocketmq.common.BrokerConfig;
@@ -233,7 +233,7 @@ public class PopConsumerServiceTest {
// fifo block
PopConsumerContext context = new PopConsumerContext(
clientHost, System.currentTimeMillis(), 20000, groupId, false, ConsumeInitMode.MIN, attemptId);
consumerService.setFifoBlocked(context, groupId, topicId, queueId, Collections.singletonList(100L));
consumerService.setFifoBlocked(context, groupId, topicId, queueId, Collections.singletonList(100L), resetGetMessageResult);
Mockito.when(brokerController.getConsumerOrderInfoManager()
.checkBlock(anyString(), anyString(), anyString(), anyInt(), anyLong())).thenReturn(true);
Assert.assertTrue(consumerService.isFifoBlocked(context, groupId, topicId, queueId));
broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOrderInfoManagerLockFreeNotifyTest.java → broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerLockFreeNotifyTest.java
+4 -4
View File
@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.broker.offset;
package org.apache.rocketmq.broker.pop.orderly;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -43,7 +43,7 @@ public class ConsumerOrderInfoManagerLockFreeNotifyTest {
private static final int QUEUE_ID_0 = 0;
private long popTime;
private ConsumerOrderInfoManager consumerOrderInfoManager;
private QueueLevelConsumerManager consumerOrderInfoManager;
private AtomicBoolean notified;
private final BrokerConfig brokerConfig = new BrokerConfig();
@@ -61,7 +61,7 @@ public class ConsumerOrderInfoManagerLockFreeNotifyTest {
return null;
}).when(popMessageProcessor).notifyLongPollingRequestIfNeed(anyString(), anyString(), anyInt());
consumerOrderInfoManager = new ConsumerOrderInfoManager(brokerController);
consumerOrderInfoManager = new QueueLevelConsumerManager(brokerController);
popTime = System.currentTimeMillis();
}
@@ -158,7 +158,7 @@ public class ConsumerOrderInfoManagerLockFreeNotifyTest {
@Test
public void testRecover() {
ConsumerOrderInfoManager savedConsumerOrderInfoManager = new ConsumerOrderInfoManager();
QueueLevelConsumerManager savedConsumerOrderInfoManager = new QueueLevelConsumerManager();
savedConsumerOrderInfoManager.update(
null,
false,
broker/src/test/java/org/apache/rocketmq/broker/offset/ConsumerOrderInfoManagerTest.java → broker/src/test/java/org/apache/rocketmq/broker/pop/orderly/ConsumerOrderInfoManagerTest.java
+8 -8
View File
@@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.rocketmq.broker.offset;
package org.apache.rocketmq.broker.pop.orderly;
import java.time.Duration;
import java.util.Map;
@@ -51,11 +51,11 @@ public class ConsumerOrderInfoManagerTest {
private static final int QUEUE_ID_1 = 1;
private long popTime;
private ConsumerOrderInfoManager consumerOrderInfoManager;
private QueueLevelConsumerManager consumerOrderInfoManager;
@Before
public void before() {
consumerOrderInfoManager = new ConsumerOrderInfoManager();
consumerOrderInfoManager = new QueueLevelConsumerManager();
popTime = System.currentTimeMillis();
}
@@ -387,7 +387,7 @@ public class ConsumerOrderInfoManagerTest {
TopicConfig topicConfig = new TopicConfig(TOPIC);
when(topicConfigManager.selectTopicConfig(eq(TOPIC))).thenReturn(topicConfig);
ConsumerOrderInfoManager consumerOrderInfoManager = new ConsumerOrderInfoManager(brokerController);
QueueLevelConsumerManager consumerOrderInfoManager = new QueueLevelConsumerManager(brokerController);
{
consumerOrderInfoManager.update(null, false,
@@ -444,7 +444,7 @@ public class ConsumerOrderInfoManagerTest {
consumerOrderInfoManager.autoClean();
assertEquals(1, consumerOrderInfoManager.getTable().size());
for (ConcurrentHashMap<Integer, ConsumerOrderInfoManager.OrderInfo> orderInfoMap : consumerOrderInfoManager.getTable().values()) {
for (ConcurrentHashMap<Integer, QueueLevelConsumerManager.OrderInfo> orderInfoMap : consumerOrderInfoManager.getTable().values()) {
assertEquals(1, orderInfoMap.size());
assertNotNull(orderInfoMap.get(QUEUE_ID_0));
break;
@@ -453,13 +453,13 @@ public class ConsumerOrderInfoManagerTest {
}
private void assertEncodeAndDecode() {
ConsumerOrderInfoManager.OrderInfo prevOrderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
QueueLevelConsumerManager.OrderInfo prevOrderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
.get().get(QUEUE_ID_0);
String dataEncoded = consumerOrderInfoManager.encode();
consumerOrderInfoManager.decode(dataEncoded);
ConsumerOrderInfoManager.OrderInfo newOrderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
QueueLevelConsumerManager.OrderInfo newOrderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
.get().get(QUEUE_ID_0);
assertNotSame(prevOrderInfo, newOrderInfo);
@@ -482,7 +482,7 @@ public class ConsumerOrderInfoManagerTest {
1,
Lists.newArrayList(2L, 3L, 4L),
new StringBuilder());
ConsumerOrderInfoManager.OrderInfo orderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
QueueLevelConsumerManager.OrderInfo orderInfo = consumerOrderInfoManager.getTable().values().stream().findFirst()
.get().get(QUEUE_ID_0);
orderInfo.setInvisibleTime(null);
common/src/main/java/org/apache/rocketmq/common/OrderedConsumptionLevel.java
+39
View File
@@ -0,0 +1,39 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.common;
public enum OrderedConsumptionLevel {
QUEUE(0),
SHARDING_KEY(1);
private final int value;
OrderedConsumptionLevel(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static OrderedConsumptionLevel valueOf(int value) {
if (value == 1) {
return SHARDING_KEY;
}
return QUEUE;
}
}