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 #8591] Preliminary support for key commands of LMQ (#8590)

Browse Source 
* Preliminary support for key commands of LMQ

* Preliminary support for key commands of LMQ

* Optimize some code

* Fix some bugs and UTs for lmq support

* Fix UTs can not pass

* Fix UTs can not pass

* Add some check to prevent NPE
This commit is contained in:
rongtong
2024-08-30 14:15:03 +08:00
committed by GitHub
parent 1a15729ca9
commit e5e38396ba
19 changed files with 349 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+1 -1
View File
@@ -2062,7 +2062,7 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
Map<Integer, Long> queueOffsetMap = new HashMap<>();
// Reset offset for all queues belonging to the specified topic
TopicConfig topicConfig = brokerController.getTopicConfigManager().getTopicConfigTable().get(topic);
TopicConfig topicConfig = brokerController.getTopicConfigManager().selectTopicConfig(topic);
if (null == topicConfig) {
response.setCode(ResponseCode.TOPIC_NOT_EXIST);
response.setRemark("Topic " + topic + " does not exist");
client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
+32 -15
View File
@@ -28,6 +28,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.Validators;
import org.apache.rocketmq.client.exception.MQBrokerException;
@@ -43,6 +44,7 @@ import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageId;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.NetworkUtil;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
@@ -199,7 +201,7 @@ public class MQAdminImpl {
if (brokerAddr != null) {
try {
return this.mQClientFactory.getMQClientAPIImpl().searchOffset(brokerAddr, mq, timestamp,
boundaryType, timeoutMillis);
boundaryType, timeoutMillis);
} catch (Exception e) {
throw new MQClientException("Invoke Broker[" + brokerAddr + "] exception", e);
}
@@ -277,13 +279,20 @@ public class MQAdminImpl {
public QueryResult queryMessage(String topic, String key, int maxNum, long begin,
long end) throws MQClientException,
InterruptedException {
return queryMessage(topic, key, maxNum, begin, end, false);
return queryMessage(null, topic, key, maxNum, begin, end, false);
}
public QueryResult queryMessageByUniqKey(String topic, String uniqKey, int maxNum, long begin, long end)
throws MQClientException, InterruptedException {
return queryMessage(topic, uniqKey, maxNum, begin, end, true);
return queryMessage(null, topic, uniqKey, maxNum, begin, end, true);
}
public QueryResult queryMessageByUniqKey(String clusterName, String topic, String uniqKey, int maxNum, long begin,
long end)
throws MQClientException, InterruptedException {
return queryMessage(clusterName, topic, uniqKey, maxNum, begin, end, true);
}
public MessageExt queryMessageByUniqKey(String topic,
@@ -311,25 +320,29 @@ public class MQAdminImpl {
}
}
protected QueryResult queryMessage(String topic, String key, int maxNum, long begin, long end,
public QueryResult queryMessage(String clusterName, String topic, String key, int maxNum, long begin, long end,
boolean isUniqKey) throws MQClientException,
InterruptedException {
return queryMessage(null, topic, key, maxNum, begin, end, isUniqKey);
}
boolean isLmq = MixAll.isLmq(topic);
protected QueryResult queryMessage(String clusterName, String topic, String key, int maxNum, long begin, long end,
boolean isUniqKey) throws MQClientException,
InterruptedException {
TopicRouteData topicRouteData = this.mQClientFactory.getAnExistTopicRouteData(topic);
String routeTopic = topic;
// if topic is lmq ,then use clusterName as lmq parent topic
// Use clusterName or lmq parent topic to get topic route for lmq or rmq_sys_wheel_timer
if (!StringUtils.isEmpty(topic) && (isLmq || topic.equals(TopicValidator.SYSTEM_TOPIC_PREFIX + "wheel_timer"))
&& !StringUtils.isEmpty(clusterName)) {
routeTopic = clusterName;
}
TopicRouteData topicRouteData = this.mQClientFactory.getAnExistTopicRouteData(routeTopic);
if (null == topicRouteData) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
topicRouteData = this.mQClientFactory.getAnExistTopicRouteData(topic);
this.mQClientFactory.updateTopicRouteInfoFromNameServer(routeTopic);
topicRouteData = this.mQClientFactory.getAnExistTopicRouteData(routeTopic);
}
if (topicRouteData != null) {
List<String> brokerAddrs = new LinkedList<>();
for (BrokerData brokerData : topicRouteData.getBrokerDatas()) {
if (clusterName != null && !clusterName.isEmpty()
if (!isLmq && clusterName != null && !clusterName.isEmpty()
&& !clusterName.equals(brokerData.getCluster())) {
continue;
}
@@ -347,7 +360,11 @@ public class MQAdminImpl {
for (String addr : brokerAddrs) {
try {
QueryMessageRequestHeader requestHeader = new QueryMessageRequestHeader();
requestHeader.setTopic(topic);
if (isLmq) {
requestHeader.setTopic(clusterName);
} else {
requestHeader.setTopic(topic);
}
requestHeader.setKey(key);
requestHeader.setMaxNum(maxNum);
requestHeader.setBeginTimestamp(begin);
@@ -436,7 +453,7 @@ public class MQAdminImpl {
String[] keyArray = keys.split(MessageConst.KEY_SEPARATOR);
for (String k : keyArray) {
// both topic and key must be equal at the same time
if (Objects.equals(key, k) && Objects.equals(topic, msgTopic)) {
if (Objects.equals(key, k) && (isLmq || Objects.equals(topic, msgTopic))) {
matched = true;
break;
}
client/src/test/java/org/apache/rocketmq/client/impl/MQAdminImplTest.java
+1 -1
View File
@@ -165,7 +165,7 @@ public class MQAdminImplTest {
callback.operationSucceed(response);
return null;
}).when(mQClientAPIImpl).queryMessage(anyString(), any(), anyLong(), any(InvokeCallback.class), any());
QueryResult actual = mqAdminImpl.queryMessage(defaultTopic, "keys", 100, 1L, 50L, false);
QueryResult actual = mqAdminImpl.queryMessage(defaultTopic, "keys", 100, 1L, 50L);
assertNotNull(actual);
assertEquals(1, actual.getMessageList().size());
assertEquals(defaultTopic, actual.getMessageList().get(0).getTopic());
example/src/main/java/org/apache/rocketmq/example/simple/LMQProducer.java → example/src/main/java/org/apache/rocketmq/example/lmq/LMQProducer.java
+2 -1
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.example.simple;
package org.apache.rocketmq.example.lmq;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
@@ -47,6 +47,7 @@ public class LMQProducer {
for (int i = 0; i < 128; i++) {
try {
Message msg = new Message(TOPIC, TAG, ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
msg.setKeys("Key" + i);
msg.putUserProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH /* "INNER_MULTI_DISPATCH" */,
String.join(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER, LMQ_TOPIC_1, LMQ_TOPIC_2) /* "%LMQ%123,%LMQ%456" */);
SendResult sendResult = producer.send(msg);
example/src/main/java/org/apache/rocketmq/example/simple/LMQPullConsumer.java → example/src/main/java/org/apache/rocketmq/example/lmq/LMQPullConsumer.java
+1 -1
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.example.simple;
package org.apache.rocketmq.example.lmq;
import java.util.Arrays;
import java.util.HashSet;
example/src/main/java/org/apache/rocketmq/example/simple/LMQPushConsumer.java → example/src/main/java/org/apache/rocketmq/example/lmq/LMQPushConsumer.java
+1 -1
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.example.simple;
package org.apache.rocketmq.example.lmq;
import com.google.common.collect.Lists;
example/src/main/java/org/apache/rocketmq/example/simple/LMQPushPopConsumer.java → example/src/main/java/org/apache/rocketmq/example/lmq/LMQPushPopConsumer.java
+1 -1
View File
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.rocketmq.example.simple;
package org.apache.rocketmq.example.lmq;
import com.google.common.collect.Lists;
import java.util.HashMap;
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExt.java
+68 -17
View File
@@ -153,6 +153,12 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
return defaultMQAdminExtImpl.queryMessage(topic, key, maxNum, begin, end);
}
public QueryResult queryMessage(String clusterName, String topic, String key, int maxNum, long begin, long end)
throws MQClientException, InterruptedException, RemotingException {
return defaultMQAdminExtImpl.queryMessage(clusterName, topic, key, maxNum, begin, end);
}
@Override
public void start() throws MQClientException {
defaultMQAdminExtImpl.start();
@@ -196,7 +202,8 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
public void createAndUpdateTopicConfigList(String addr, List<TopicConfig> topicConfigList) throws InterruptedException, RemotingException, MQClientException {
public void createAndUpdateTopicConfigList(String addr,
List<TopicConfig> topicConfigList) throws InterruptedException, RemotingException, MQClientException {
defaultMQAdminExtImpl.createAndUpdateTopicConfigList(addr, topicConfigList);
}
@@ -300,6 +307,12 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
return examineConsumeStats(consumerGroup, null);
}
@Override
public ConsumeStats examineConsumeStats(String clusterName, String consumerGroup,
String topic) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.examineConsumeStats(clusterName, consumerGroup, topic);
}
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
String topic) throws RemotingException, MQClientException,
@@ -459,16 +472,35 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
return defaultMQAdminExtImpl.resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
}
public List<RollbackStats> resetOffsetByTimestampOld(String clusterName, String consumerGroup, String topic, long timestamp,
boolean force)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestampOld(clusterName, consumerGroup, topic, timestamp, force);
}
@Override
public Map<MessageQueue, Long> resetOffsetByTimestamp(String clusterName, String topic, String group,
long timestamp, boolean isForce) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestamp(clusterName, topic, group, timestamp, isForce);
}
@Override
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return resetOffsetByTimestamp(topic, group, timestamp, isForce, false);
}
public Map<MessageQueue, Long> resetOffsetByTimestamp(String clusterName, String topic, String group,
long timestamp, boolean isForce, boolean isC)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestamp(clusterName, topic, group, timestamp, isForce, isC);
}
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce,
boolean isC)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return defaultMQAdminExtImpl.resetOffsetByTimestamp(topic, group, timestamp, isForce, isC);
return defaultMQAdminExtImpl.resetOffsetByTimestamp(null, topic, group, timestamp, isForce, isC);
}
@Override
@@ -589,10 +621,19 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
final String topic, final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
}
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String clusterName, final String consumerGroup,
final String clientId,
final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return defaultMQAdminExtImpl.consumeMessageDirectly(clusterName, consumerGroup, clientId, topic, msgId);
}
@Override
public List<MessageTrack> messageTrackDetail(
MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
@@ -796,10 +837,10 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
this.defaultMQAdminExtImpl.resetMasterFlushOffset(brokerAddr, masterFlushOffset);
}
public QueryResult queryMessageByUniqKey(String topic, String key, int maxNum, long begin, long end)
public QueryResult queryMessageByUniqKey(String clusterName, String topic, String key, int maxNum, long begin,
long end)
throws MQClientException, InterruptedException {
return defaultMQAdminExtImpl.queryMessageByUniqKey(topic, key, maxNum, begin, end);
return defaultMQAdminExtImpl.queryMessageByUniqKey(clusterName, topic, key, maxNum, begin, end);
}
public DefaultMQAdminExtImpl getDefaultMQAdminExtImpl() {
@@ -831,13 +872,14 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
@Override
public Pair<ElectMasterResponseHeader, BrokerMemberGroup> electMaster(String controllerAddr, String clusterName,
String brokerName, Long brokerId) throws RemotingException, InterruptedException, MQBrokerException {
String brokerName, Long brokerId) throws RemotingException, InterruptedException, MQBrokerException {
return this.defaultMQAdminExtImpl.electMaster(controllerAddr, clusterName, brokerName, brokerId);
}
@Override
public void cleanControllerBrokerData(String controllerAddr, String clusterName, String brokerName,
String brokerControllerIdsToClean, boolean isCleanLivingBroker) throws RemotingException, InterruptedException, MQBrokerException {
String brokerControllerIdsToClean,
boolean isCleanLivingBroker) throws RemotingException, InterruptedException, MQBrokerException {
this.defaultMQAdminExtImpl.cleanControllerBrokerData(controllerAddr, clusterName, brokerName, brokerControllerIdsToClean, isCleanLivingBroker);
}
@@ -876,13 +918,15 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
}
@Override
public void createUser(String brokerAddr, String username, String password, String userType) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void createUser(String brokerAddr, String username, String password,
String userType) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.createUser(brokerAddr, username, password, userType);
}
@Override
public void updateUser(String brokerAddr, String username,
String password, String userType, String userStatus) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
String password, String userType,
String userStatus) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.updateUser(brokerAddr, username, password, userType, userStatus);
}
@@ -912,38 +956,45 @@ public class DefaultMQAdminExt extends ClientConfig implements MQAdminExt {
@Override
public void createAcl(String brokerAddr, String subject, List<String> resources, List<String> actions,
List<String> sourceIps, String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
List<String> sourceIps,
String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.createAcl(brokerAddr, subject, resources, actions, sourceIps, decision);
}
@Override
public void createAcl(String brokerAddr, AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void createAcl(String brokerAddr,
AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.createAcl(brokerAddr, aclInfo);
}
@Override
public void updateAcl(String brokerAddr, String subject, List<String> resources, List<String> actions,
List<String> sourceIps, String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
List<String> sourceIps,
String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.updateAcl(brokerAddr, subject, resources, actions, sourceIps, decision);
}
@Override
public void updateAcl(String brokerAddr, AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void updateAcl(String brokerAddr,
AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.updateAcl(brokerAddr, aclInfo);
}
@Override
public void deleteAcl(String brokerAddr, String subject, String resource) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void deleteAcl(String brokerAddr, String subject,
String resource) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
defaultMQAdminExtImpl.deleteAcl(brokerAddr, subject, resource);
}
@Override
public AclInfo getAcl(String brokerAddr, String subject) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public AclInfo getAcl(String brokerAddr,
String subject) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
return defaultMQAdminExtImpl.getAcl(brokerAddr, subject);
}
@Override
public List<AclInfo> listAcl(String brokerAddr, String subjectFilter, String resourceFilter) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public List<AclInfo> listAcl(String brokerAddr, String subjectFilter,
String resourceFilter) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
return defaultMQAdminExtImpl.listAcl(brokerAddr, subjectFilter, resourceFilter);
}
}
tools/src/main/java/org/apache/rocketmq/tools/admin/DefaultMQAdminExtImpl.java
+116 -41
View File
@@ -424,12 +424,19 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public ConsumeStats examineConsumeStats(
String consumerGroup) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return examineConsumeStats(consumerGroup, null);
String consumerGroup,
String topic) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return examineConsumeStats(null, consumerGroup, topic);
}
@Override
public ConsumeStats examineConsumeStats(String consumerGroup,
public ConsumeStats examineConsumeStats(
String consumerGroup) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
return examineConsumeStats(null, consumerGroup, null);
}
@Override
public ConsumeStats examineConsumeStats(String clusterName, String consumerGroup,
String topic) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
TopicRouteData topicRouteData = null;
List<String> routeTopics = new ArrayList<>();
@@ -438,6 +445,12 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
routeTopics.add(topic);
routeTopics.add(KeyBuilder.buildPopRetryTopic(topic, consumerGroup));
}
// Use clusterName topic to get topic route for lmq or rmq_sys_wheel_timer
if (!StringUtils.isEmpty(topic) && (MixAll.isLmq(topic) || topic.equals(TopicValidator.SYSTEM_TOPIC_PREFIX + "wheel_timer")) && !StringUtils.isEmpty(clusterName)) {
routeTopics.add(clusterName);
}
for (int i = 0; i < routeTopics.size(); i++) {
try {
topicRouteData = this.examineTopicRouteInfo(routeTopics.get(i));
@@ -467,25 +480,33 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
topics.add(messageQueue.getTopic());
}
ConsumeStats staticResult = new ConsumeStats();
staticResult.setConsumeTps(result.getConsumeTps());
// for topic, we put the physical stats, how about group?
// staticResult.getOffsetTable().putAll(result.getOffsetTable());
ConsumeStats staticResult = null;
for (String currentTopic : topics) {
TopicRouteData currentRoute = this.examineTopicRouteInfo(currentTopic);
if (currentRoute.getTopicQueueMappingByBroker() == null
|| currentRoute.getTopicQueueMappingByBroker().isEmpty()) {
//normal topic
for (Map.Entry<MessageQueue, OffsetWrapper> entry : result.getOffsetTable().entrySet()) {
if (entry.getKey().getTopic().equals(currentTopic)) {
staticResult.getOffsetTable().put(entry.getKey(), entry.getValue());
if (StringUtils.isEmpty(clusterName)) {
staticResult = new ConsumeStats();
staticResult.setConsumeTps(result.getConsumeTps());
// for topic, we put the physical stats, how about group?
// staticResult.getOffsetTable().putAll(result.getOffsetTable());
for (String currentTopic : topics) {
TopicRouteData currentRoute = this.examineTopicRouteInfo(currentTopic);
if (currentRoute.getTopicQueueMappingByBroker() == null
|| currentRoute.getTopicQueueMappingByBroker().isEmpty()) {
//normal topic
for (Map.Entry<MessageQueue, OffsetWrapper> entry : result.getOffsetTable().entrySet()) {
if (entry.getKey().getTopic().equals(currentTopic)) {
staticResult.getOffsetTable().put(entry.getKey(), entry.getValue());
}
}
}
Map<String, TopicConfigAndQueueMapping> brokerConfigMap = MQAdminUtils.examineTopicConfigFromRoute(currentTopic, currentRoute, defaultMQAdminExt);
ConsumeStats consumeStats = MQAdminUtils.convertPhysicalConsumeStats(brokerConfigMap, result);
staticResult.getOffsetTable().putAll(consumeStats.getOffsetTable());
}
Map<String, TopicConfigAndQueueMapping> brokerConfigMap = MQAdminUtils.examineTopicConfigFromRoute(currentTopic, currentRoute, defaultMQAdminExt);
ConsumeStats consumeStats = MQAdminUtils.convertPhysicalConsumeStats(brokerConfigMap, result);
staticResult.getOffsetTable().putAll(consumeStats.getOffsetTable());
} else {
staticResult = result;
}
if (staticResult.getOffsetTable().isEmpty()) {
@@ -811,10 +832,16 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
this.mqClientInstance.getMQClientAPIImpl().deleteKVConfigValue(namespace, key, timeoutMillis);
}
@Override
public List<RollbackStats> resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
public List<RollbackStats> resetOffsetByTimestampOld(String clusterName, String consumerGroup, String topic,
long timestamp,
boolean force) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
String routeTopic = topic;
// Use clusterName topic to get topic route for lmq or rmq_sys_wheel_timer
if (!StringUtils.isEmpty(topic) && (MixAll.isLmq(topic) || topic.equals(TopicValidator.SYSTEM_TOPIC_PREFIX + "wheel_timer"))
&& !StringUtils.isEmpty(clusterName)) {
routeTopic = clusterName;
}
TopicRouteData topicRouteData = this.examineTopicRouteInfo(routeTopic);
List<RollbackStats> rollbackStatsList = new ArrayList<>();
Map<String, QueueData> topicRouteMap = new HashMap<>();
for (QueueData queueData : topicRouteData.getQueueDatas()) {
@@ -829,6 +856,12 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return rollbackStatsList;
}
@Override
public List<RollbackStats> resetOffsetByTimestampOld(String consumerGroup, String topic, long timestamp,
boolean force) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return resetOffsetByTimestampOld(null, consumerGroup, topic, timestamp, force);
}
private List<RollbackStats> resetOffsetByTimestampOld(String brokerAddr, QueueData queueData, String consumerGroup,
String topic, long timestamp,
boolean force) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
@@ -864,7 +897,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp,
boolean isForce) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return resetOffsetByTimestamp(topic, group, timestamp, isForce, false);
return resetOffsetByTimestamp(null, topic, group, timestamp, isForce, false);
}
@Override
@@ -951,9 +984,16 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
});
}
public Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce,
public Map<MessageQueue, Long> resetOffsetByTimestamp(String clusterName, String topic, String group,
long timestamp, boolean isForce,
boolean isC) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
TopicRouteData topicRouteData = this.examineTopicRouteInfo(topic);
String routeTopic = topic;
// Use clusterName topic to get topic route for lmq or rmq_sys_wheel_timer
if (!StringUtils.isEmpty(topic) && (MixAll.isLmq(topic) || topic.equals(TopicValidator.SYSTEM_TOPIC_PREFIX + "wheel_timer"))
&& !StringUtils.isEmpty(clusterName)) {
routeTopic = clusterName;
}
TopicRouteData topicRouteData = this.examineTopicRouteInfo(routeTopic);
List<BrokerData> brokerDatas = topicRouteData.getBrokerDatas();
Map<MessageQueue, Long> allOffsetTable = new HashMap<>();
if (brokerDatas != null) {
@@ -1325,7 +1365,8 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String consumerGroup, final String clientId,
final String topic, final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
MessageExt msg = this.viewMessage(topic, msgId);
if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(NetworkUtil.socketAddress2String(msg.getStoreHost()), consumerGroup, clientId, topic, msgId, timeoutMillis);
@@ -1335,6 +1376,20 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
}
@Override
public ConsumeMessageDirectlyResult consumeMessageDirectly(final String clusterName, final String consumerGroup,
final String clientId,
final String topic,
final String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
MessageExt msg = this.queryMessage(clusterName, topic, msgId);
if (msg.getProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX) == null) {
return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(NetworkUtil.socketAddress2String(msg.getStoreHost()), consumerGroup, clientId, topic, msgId, timeoutMillis);
} else {
MessageClientExt msgClient = (MessageClientExt) msg;
return this.mqClientInstance.getMQClientAPIImpl().consumeMessageDirectly(NetworkUtil.socketAddress2String(msg.getStoreHost()), consumerGroup, clientId, topic, msgClient.getOffsetMsgId(), timeoutMillis);
}
}
@Override
public List<MessageTrack> messageTrackDetail(
MessageExt msg) throws RemotingException, MQClientException, InterruptedException, MQBrokerException {
@@ -1664,10 +1719,10 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
while (iterator.hasNext()) {
TopicConfig topicConfig = iterator.next().getValue();
if (topicList.getTopicList().contains(topicConfig.getTopicName())
|| TopicValidator.isSystemTopic(topicConfig.getTopicName())) {
|| TopicValidator.isSystemTopic(topicConfig.getTopicName())) {
iterator.remove();
} else if (!specialTopic && StringUtils.startsWithAny(topicConfig.getTopicName(),
MixAll.RETRY_GROUP_TOPIC_PREFIX, MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
MixAll.RETRY_GROUP_TOPIC_PREFIX, MixAll.DLQ_GROUP_TOPIC_PREFIX)) {
iterator.remove();
} else if (!PermName.isValid(topicConfig.getPerm())) {
iterator.remove();
@@ -1726,6 +1781,11 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return this.mqClientInstance.getMQAdminImpl().queryMessage(topic, key, maxNum, begin, end);
}
public QueryResult queryMessage(String clusterName, String topic, String key, int maxNum, long begin,
long end) throws MQClientException, InterruptedException, RemotingException {
return this.mqClientInstance.getMQAdminImpl().queryMessage(clusterName, topic, key, maxNum, begin, end, false);
}
@Override
public void updateConsumeOffset(String brokerAddr, String consumeGroup, MessageQueue mq,
long offset) throws RemotingException, InterruptedException, MQBrokerException {
@@ -1783,10 +1843,9 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
return this.mqClientInstance.getMQClientAPIImpl().searchOffset(brokerAddr, topicName, queueId, timestamp, timeoutMillis);
}
public QueryResult queryMessageByUniqKey(String topic, String key, int maxNum, long begin,
public QueryResult queryMessageByUniqKey(String clusterName, String topic, String key, int maxNum, long begin,
long end) throws MQClientException, InterruptedException {
return this.mqClientInstance.getMQAdminImpl().queryMessageByUniqKey(topic, key, maxNum, begin, end);
return this.mqClientInstance.getMQAdminImpl().queryMessageByUniqKey(clusterName, topic, key, maxNum, begin, end);
}
@Override
@@ -1812,6 +1871,12 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
}
@Override
public Map<MessageQueue, Long> resetOffsetByTimestamp(String clusterName, String topic, String group,
long timestamp, boolean isForce) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
return resetOffsetByTimestamp(clusterName, topic, group, timestamp, isForce, false);
}
@Override
public HARuntimeInfo getBrokerHAStatus(
String brokerAddr) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
@@ -1844,7 +1909,7 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public Pair<ElectMasterResponseHeader, BrokerMemberGroup> electMaster(String controllerAddr, String clusterName,
String brokerName, Long brokerId) throws RemotingException, InterruptedException, MQBrokerException {
String brokerName, Long brokerId) throws RemotingException, InterruptedException, MQBrokerException {
return this.mqClientInstance.getMQClientAPIImpl().electMaster(controllerAddr, clusterName, brokerName, brokerId);
}
@@ -1930,20 +1995,23 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
}
@Override
public void createUser(String brokerAddr, String username, String password, String userType) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void createUser(String brokerAddr, String username, String password,
String userType) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
UserInfo userInfo = UserInfo.of(username, password, userType);
this.createUser(brokerAddr, userInfo);
}
@Override
public void updateUser(String brokerAddr, String username,
String password, String userType, String userStatus) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
String password, String userType,
String userStatus) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
UserInfo userInfo = UserInfo.of(username, password, userType, userStatus);
this.mqClientInstance.getMQClientAPIImpl().updateUser(brokerAddr, userInfo, timeoutMillis);
}
@Override
public void updateUser(String brokerAddr, UserInfo userInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void updateUser(String brokerAddr,
UserInfo userInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
this.mqClientInstance.getMQClientAPIImpl().updateUser(brokerAddr, userInfo, timeoutMillis);
}
@@ -1967,40 +2035,47 @@ public class DefaultMQAdminExtImpl implements MQAdminExt, MQAdminExtInner {
@Override
public void createAcl(String brokerAddr, String subject, List<String> resources, List<String> actions,
List<String> sourceIps, String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
List<String> sourceIps,
String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
AclInfo aclInfo = AclInfo.of(subject, resources, actions, sourceIps, decision);
this.createAcl(brokerAddr, aclInfo);
}
@Override
public void createAcl(String brokerAddr, AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void createAcl(String brokerAddr,
AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
this.mqClientInstance.getMQClientAPIImpl().createAcl(brokerAddr, aclInfo, timeoutMillis);
}
@Override
public void updateAcl(String brokerAddr, String subject, List<String> resources, List<String> actions,
List<String> sourceIps, String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
List<String> sourceIps,
String decision) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
AclInfo aclInfo = AclInfo.of(subject, resources, actions, sourceIps, decision);
this.updateAcl(brokerAddr, aclInfo);
}
@Override
public void updateAcl(String brokerAddr, AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void updateAcl(String brokerAddr,
AclInfo aclInfo) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
this.mqClientInstance.getMQClientAPIImpl().updateAcl(brokerAddr, aclInfo, timeoutMillis);
}
@Override
public void deleteAcl(String brokerAddr, String subject, String resource) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public void deleteAcl(String brokerAddr, String subject,
String resource) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
this.mqClientInstance.getMQClientAPIImpl().deleteAcl(brokerAddr, subject, resource, timeoutMillis);
}
@Override
public AclInfo getAcl(String brokerAddr, String subject) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public AclInfo getAcl(String brokerAddr,
String subject) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().getAcl(brokerAddr, subject, timeoutMillis);
}
@Override
public List<AclInfo> listAcl(String brokerAddr, String subjectFilter, String resourceFilter) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
public List<AclInfo> listAcl(String brokerAddr, String subjectFilter,
String resourceFilter) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException {
return this.mqClientInstance.getMQClientAPIImpl().listAcl(brokerAddr, subjectFilter, resourceFilter, timeoutMillis);
}
}
tools/src/main/java/org/apache/rocketmq/tools/admin/MQAdminExt.java
+12
View File
@@ -152,6 +152,10 @@ public interface MQAdminExt extends MQAdmin {
final String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException;
ConsumeStats examineConsumeStats(final String clusterName, final String consumerGroup,
final String topic) throws RemotingException, MQClientException,
InterruptedException, MQBrokerException;
ConsumeStats examineConsumeStats(final String brokerAddr, final String consumerGroup, final String topicName,
final long timeoutMillis) throws InterruptedException, RemotingTimeoutException, RemotingSendRequestException,
RemotingConnectException, MQBrokerException;
@@ -232,6 +236,9 @@ public interface MQAdminExt extends MQAdmin {
Map<MessageQueue, Long> resetOffsetByTimestamp(String topic, String group, long timestamp, boolean isForce)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException;
Map<MessageQueue, Long> resetOffsetByTimestamp(String clusterName, String topic, String group, long timestamp, boolean isForce)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException;
void resetOffsetNew(String consumerGroup, String topic, long timestamp) throws RemotingException, MQBrokerException,
InterruptedException, MQClientException;
@@ -293,6 +300,11 @@ public interface MQAdminExt extends MQAdmin {
String topic,
String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException;
ConsumeMessageDirectlyResult consumeMessageDirectly(String clusterName, String consumerGroup,
String clientId,
String topic,
String msgId) throws RemotingException, MQClientException, InterruptedException, MQBrokerException;
List<MessageTrack> messageTrackDetail(
MessageExt msg) throws RemotingException, MQClientException, InterruptedException,
MQBrokerException;
tools/src/main/java/org/apache/rocketmq/tools/command/consumer/ConsumerProgressSubCommand.java
+7 -1
View File
@@ -72,6 +72,10 @@ public class ConsumerProgressSubCommand implements SubCommand {
optionShowClientIP.setRequired(false);
options.addOption(optionShowClientIP);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -109,6 +113,8 @@ public class ConsumerProgressSubCommand implements SubCommand {
boolean showClientIP = commandLine.hasOption('s')
&& "true".equalsIgnoreCase(commandLine.getOptionValue('s'));
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
if (commandLine.hasOption('g')) {
String consumerGroup = commandLine.getOptionValue('g').trim();
String topicName = commandLine.hasOption('t') ? commandLine.getOptionValue('t').trim() : null;
@@ -116,7 +122,7 @@ public class ConsumerProgressSubCommand implements SubCommand {
if (topicName == null) {
consumeStats = defaultMQAdminExt.examineConsumeStats(consumerGroup);
} else {
consumeStats = defaultMQAdminExt.examineConsumeStats(consumerGroup, topicName);
consumeStats = defaultMQAdminExt.examineConsumeStats(clusterName, consumerGroup, topicName);
}
List<MessageQueue> mqList = new LinkedList<>(consumeStats.getOffsetTable().keySet());
Collections.sort(mqList);
tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByIdSubCommand.java
+19 -10
View File
@@ -44,9 +44,10 @@ import org.apache.rocketmq.tools.command.SubCommand;
import org.apache.rocketmq.tools.command.SubCommandException;
public class QueryMsgByIdSubCommand implements SubCommand {
public static void queryById(final DefaultMQAdminExt admin, final String topic, final String msgId, final Charset msgBodyCharset) throws MQClientException,
public static void queryById(final DefaultMQAdminExt admin, final String clusterName, final String topic,
final String msgId, final Charset msgBodyCharset) throws MQClientException,
RemotingException, MQBrokerException, InterruptedException, IOException {
MessageExt msg = admin.viewMessage(topic, msgId);
MessageExt msg = admin.queryMessage(clusterName, topic, msgId);
printMsg(admin, msg, msgBodyCharset);
}
@@ -55,7 +56,8 @@ public class QueryMsgByIdSubCommand implements SubCommand {
printMsg(admin, msg, null);
}
public static void printMsg(final DefaultMQAdminExt admin, final MessageExt msg, final Charset msgBodyCharset) throws IOException {
public static void printMsg(final DefaultMQAdminExt admin, final MessageExt msg,
final Charset msgBodyCharset) throws IOException {
if (msg == null) {
System.out.printf("%nMessage not found!");
return;
@@ -219,6 +221,10 @@ public class QueryMsgByIdSubCommand implements SubCommand {
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -244,13 +250,14 @@ public class QueryMsgByIdSubCommand implements SubCommand {
final String msgIds = commandLine.getOptionValue('i').trim();
final String[] msgIdArr = StringUtils.split(msgIds, ",");
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
if (commandLine.hasOption('g') && commandLine.hasOption('d')) {
final String consumerGroup = commandLine.getOptionValue('g').trim();
final String clientId = commandLine.getOptionValue('d').trim();
for (String msgId : msgIdArr) {
if (StringUtils.isNotBlank(msgId)) {
pushMsg(defaultMQAdminExt, consumerGroup, clientId, topic, msgId.trim());
pushMsg(defaultMQAdminExt, clusterName, consumerGroup, clientId, topic, msgId.trim());
}
}
} else if (commandLine.hasOption('s')) {
@@ -258,7 +265,7 @@ public class QueryMsgByIdSubCommand implements SubCommand {
if (resend) {
for (String msgId : msgIdArr) {
if (StringUtils.isNotBlank(msgId)) {
sendMsg(defaultMQAdminExt, defaultMQProducer, topic, msgId.trim());
sendMsg(defaultMQAdminExt, clusterName, defaultMQProducer, topic, msgId.trim());
}
}
}
@@ -269,7 +276,7 @@ public class QueryMsgByIdSubCommand implements SubCommand {
}
for (String msgId : msgIdArr) {
if (StringUtils.isNotBlank(msgId)) {
queryById(defaultMQAdminExt, topic, msgId.trim(), msgBodyCharset);
queryById(defaultMQAdminExt, clusterName, topic, msgId.trim(), msgBodyCharset);
}
}
@@ -282,13 +289,14 @@ public class QueryMsgByIdSubCommand implements SubCommand {
}
}
private void pushMsg(final DefaultMQAdminExt defaultMQAdminExt, final String consumerGroup, final String clientId,
private void pushMsg(final DefaultMQAdminExt defaultMQAdminExt, final String clusterName,
final String consumerGroup, final String clientId,
final String topic, final String msgId) {
try {
ConsumerRunningInfo consumerRunningInfo = defaultMQAdminExt.getConsumerRunningInfo(consumerGroup, clientId, false, false);
if (consumerRunningInfo != null && ConsumerRunningInfo.isPushType(consumerRunningInfo)) {
ConsumeMessageDirectlyResult result =
defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
defaultMQAdminExt.consumeMessageDirectly(clusterName, consumerGroup, clientId, topic, msgId);
System.out.printf("%s", result);
} else {
System.out.printf("this %s client is not push consumer ,not support direct push \n", clientId);
@@ -298,10 +306,11 @@ public class QueryMsgByIdSubCommand implements SubCommand {
}
}
private void sendMsg(final DefaultMQAdminExt defaultMQAdminExt, final DefaultMQProducer defaultMQProducer,
private void sendMsg(final DefaultMQAdminExt defaultMQAdminExt, final String clusterName,
final DefaultMQProducer defaultMQProducer,
final String topic, final String msgId) {
try {
MessageExt msg = defaultMQAdminExt.viewMessage(topic, msgId);
MessageExt msg = defaultMQAdminExt.queryMessage(clusterName, topic, msgId);
if (msg != null) {
// resend msg by id
System.out.printf("prepare resend msg. originalMsgId=%s", msgId);
tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByKeySubCommand.java
+19 -8
View File
@@ -23,6 +23,8 @@ import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommand;
import org.apache.rocketmq.tools.command.SubCommandException;
@@ -41,7 +43,7 @@ public class QueryMsgByKeySubCommand implements SubCommand {
@Override
public Options buildCommandlineOptions(Options options) {
Option opt = new Option("t", "topic", true, "topic name");
Option opt = new Option("t", "topic", true, "Topic name");
opt.setRequired(true);
options.addOption(opt);
@@ -57,7 +59,11 @@ public class QueryMsgByKeySubCommand implements SubCommand {
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "maxNum", true, "The maximum number of messages returned by the query, default:64");
opt = new Option("m", "maxNum", true, "The maximum number of messages returned by the query, default:64");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
@@ -77,16 +83,20 @@ public class QueryMsgByKeySubCommand implements SubCommand {
long beginTimestamp = 0;
long endTimestamp = Long.MAX_VALUE;
int maxNum = 64;
String clusterName = null;
if (commandLine.hasOption("b")) {
beginTimestamp = Long.parseLong(commandLine.getOptionValue("b").trim());
}
if (commandLine.hasOption("e")) {
endTimestamp = Long.parseLong(commandLine.getOptionValue("e").trim());
}
if (commandLine.hasOption("c")) {
maxNum = Integer.parseInt(commandLine.getOptionValue("c").trim());
if (commandLine.hasOption("m")) {
maxNum = Integer.parseInt(commandLine.getOptionValue("m").trim());
}
this.queryByKey(defaultMQAdminExt, topic, key, maxNum, beginTimestamp, endTimestamp);
if (commandLine.hasOption("c")) {
clusterName = commandLine.getOptionValue("c").trim();
}
this.queryByKey(defaultMQAdminExt, clusterName, topic, key, maxNum, beginTimestamp, endTimestamp);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
} finally {
@@ -94,12 +104,13 @@ public class QueryMsgByKeySubCommand implements SubCommand {
}
}
private void queryByKey(final DefaultMQAdminExt admin, final String topic, final String key, int maxNum, long begin,
private void queryByKey(final DefaultMQAdminExt admin, final String cluster, final String topic, final String key, int maxNum, long begin,
long end)
throws MQClientException, InterruptedException {
throws MQClientException, InterruptedException, RemotingException {
admin.start();
QueryResult queryResult = admin.queryMessage(topic, key, maxNum, begin, end);
QueryResult queryResult = admin.queryMessage(cluster, topic, key, maxNum, begin, end);
System.out.printf("%-50s %4s %40s%n",
"#Message ID",
"#QID",
tools/src/main/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommand.java
+15 -13
View File
@@ -25,13 +25,11 @@ import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.rocketmq.client.QueryResult;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
@@ -51,19 +49,18 @@ public class QueryMsgByUniqueKeySubCommand implements SubCommand {
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
defaultMQAdminExt.start();
}
catch (Exception e) {
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
}
return defaultMQAdminExt;
}
}
public static void queryById(final DefaultMQAdminExt admin, final String topic, final String msgId,
final boolean showAll) throws MQClientException,
RemotingException, MQBrokerException, InterruptedException, IOException {
public static void queryById(final DefaultMQAdminExt admin, final String clusterName, final String topic,
final String msgId,
final boolean showAll) throws MQClientException, InterruptedException, IOException {
QueryResult queryResult = admin.queryMessageByUniqKey(topic, msgId, 32, 0, Long.MAX_VALUE);
QueryResult queryResult = admin.queryMessageByUniqKey(clusterName, topic, msgId, 32, 0, Long.MAX_VALUE);
assert queryResult != null;
List<MessageExt> list = queryResult.getMessageList();
if (list == null || list.size() == 0) {
@@ -94,7 +91,7 @@ public class QueryMsgByUniqueKeySubCommand implements SubCommand {
System.out.printf(strFormat, "Store Host:", RemotingHelper.parseSocketAddressAddr(msg.getStoreHost()));
System.out.printf(intFormat, "System Flag:", msg.getSysFlag());
System.out.printf(strFormat, "Properties:",
msg.getProperties() != null ? msg.getProperties().toString() : "");
msg.getProperties() != null ? msg.getProperties().toString() : "");
System.out.printf(strFormat, "Message Body Path:", bodyTmpFilePath);
try {
@@ -166,6 +163,10 @@ public class QueryMsgByUniqueKeySubCommand implements SubCommand {
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -173,10 +174,11 @@ public class QueryMsgByUniqueKeySubCommand implements SubCommand {
public void execute(CommandLine commandLine, Options options, RPCHook rpcHook) throws SubCommandException {
try {
defaultMQAdminExt = createMQAdminExt(rpcHook);
defaultMQAdminExt = createMQAdminExt(rpcHook);
final String msgId = commandLine.getOptionValue('i').trim();
final String topic = commandLine.getOptionValue('t').trim();
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
final boolean showAll = commandLine.hasOption('a');
if (commandLine.hasOption('g') && commandLine.hasOption('d')) {
final String consumerGroup = commandLine.getOptionValue('g').trim();
@@ -189,14 +191,14 @@ public class QueryMsgByUniqueKeySubCommand implements SubCommand {
}
if (consumerRunningInfo != null && ConsumerRunningInfo.isPushType(consumerRunningInfo)) {
ConsumeMessageDirectlyResult result =
defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
defaultMQAdminExt.consumeMessageDirectly(consumerGroup, clientId, topic, msgId);
System.out.printf("%s", result);
} else {
System.out.printf("get consumer info failed or this %s client is not push consumer ,not support direct push \n", clientId);
System.out.printf("get consumer info failed or this %s client is not push consumer, not support direct push \n", clientId);
}
} else {
queryById(defaultMQAdminExt, topic, msgId, showAll);
queryById(defaultMQAdminExt, clusterName, topic, msgId, showAll);
}
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeCommand.java
+9 -4
View File
@@ -77,6 +77,10 @@ public class ResetOffsetByTimeCommand implements SubCommand {
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -88,6 +92,7 @@ public class ResetOffsetByTimeCommand implements SubCommand {
String group = commandLine.getOptionValue("g").trim();
String topic = commandLine.getOptionValue("t").trim();
String timeStampStr = commandLine.getOptionValue("s").trim();
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
long timestamp = "now".equals(timeStampStr) ? System.currentTimeMillis() : 0;
try {
@@ -129,7 +134,7 @@ public class ResetOffsetByTimeCommand implements SubCommand {
if (brokerAddr != null && queueId >= 0) {
System.out.printf("start reset consumer offset by specified, " +
"group[%s], topic[%s], queueId[%s], broker[%s], timestamp(string)[%s], timestamp(long)[%s]%n",
group, topic, queueId, brokerAddr, timeStampStr, timestamp);
group, topic, queueId, brokerAddr, timeStampStr, timestamp);
long resetOffset = null != offset ? offset :
defaultMQAdminExt.searchOffset(brokerAddr, topic, queueId, timestamp, 3000);
@@ -143,11 +148,11 @@ public class ResetOffsetByTimeCommand implements SubCommand {
Map<MessageQueue, Long> offsetTable;
try {
offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force, isC);
offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(clusterName, topic, group, timestamp, force, isC);
} catch (MQClientException e) {
// if consumer not online, use old command to reset reset
// if consumer not online, use old command to reset
if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) {
ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, group, topic, timestamp, force, timeStampStr);
ResetOffsetByTimeOldCommand.resetOffset(defaultMQAdminExt, clusterName, group, topic, timestamp, force, timeStampStr);
return;
}
throw e;
tools/src/main/java/org/apache/rocketmq/tools/command/offset/ResetOffsetByTimeOldCommand.java
+10 -3
View File
@@ -34,12 +34,13 @@ import org.apache.rocketmq.tools.command.SubCommandException;
public class ResetOffsetByTimeOldCommand implements SubCommand {
public static void resetOffset(DefaultMQAdminExt defaultMQAdminExt, String consumerGroup, String topic,
public static void resetOffset(DefaultMQAdminExt defaultMQAdminExt, String clusterName, String consumerGroup,
String topic,
long timestamp, boolean force, String timeStampStr)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
List<RollbackStats> rollbackStatsList =
defaultMQAdminExt.resetOffsetByTimestampOld(consumerGroup, topic, timestamp, force);
defaultMQAdminExt.resetOffsetByTimestampOld(clusterName, consumerGroup, topic, timestamp, force);
System.out.printf("reset consumer offset by specified " +
"consumerGroup[%s], topic[%s], force[%s], timestamp(string)[%s], timestamp(long)[%s]%n",
@@ -93,6 +94,11 @@ public class ResetOffsetByTimeOldCommand implements SubCommand {
opt = new Option("f", "force", true, "set the force rollback by timestamp switch[true|false]");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -104,6 +110,7 @@ public class ResetOffsetByTimeOldCommand implements SubCommand {
String consumerGroup = commandLine.getOptionValue("g").trim();
String topic = commandLine.getOptionValue("t").trim();
String timeStampStr = commandLine.getOptionValue("s").trim();
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
long timestamp = 0;
try {
timestamp = Long.parseLong(timeStampStr);
@@ -123,7 +130,7 @@ public class ResetOffsetByTimeOldCommand implements SubCommand {
force = Boolean.parseBoolean(commandLine.getOptionValue("f").trim());
}
defaultMQAdminExt.start();
resetOffset(defaultMQAdminExt, consumerGroup, topic, timestamp, force, timeStampStr);
resetOffset(defaultMQAdminExt, clusterName, consumerGroup, topic, timestamp, force, timeStampStr);
} catch (Exception e) {
throw new SubCommandException(this.getClass().getSimpleName() + " command failed", e);
tools/src/main/java/org/apache/rocketmq/tools/command/offset/SkipAccumulationSubCommand.java
+6 -1
View File
@@ -57,6 +57,10 @@ public class SkipAccumulationSubCommand implements SubCommand {
opt = new Option("f", "force", true, "set the force rollback by timestamp switch[true|false]");
opt.setRequired(false);
options.addOption(opt);
opt = new Option("c", "cluster", true, "Cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -68,6 +72,7 @@ public class SkipAccumulationSubCommand implements SubCommand {
try {
String group = commandLine.getOptionValue("g").trim();
String topic = commandLine.getOptionValue("t").trim();
String clusterName = commandLine.hasOption('c') ? commandLine.getOptionValue('c').trim() : null;
boolean force = true;
if (commandLine.hasOption('f')) {
force = Boolean.valueOf(commandLine.getOptionValue("f").trim());
@@ -76,7 +81,7 @@ public class SkipAccumulationSubCommand implements SubCommand {
defaultMQAdminExt.start();
Map<MessageQueue, Long> offsetTable;
try {
offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(topic, group, timestamp, force);
offsetTable = defaultMQAdminExt.resetOffsetByTimestamp(clusterName, topic, group, timestamp, force);
} catch (MQClientException e) {
if (ResponseCode.CONSUMER_NOT_ONLINE == e.getResponseCode()) {
List<RollbackStats> rollbackStatsList = defaultMQAdminExt.resetOffsetByTimestampOld(group, topic, timestamp, force);
tools/src/main/java/org/apache/rocketmq/tools/command/topic/TopicStatusSubCommand.java
+23 -1
View File
@@ -27,6 +27,8 @@ import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.apache.rocketmq.tools.command.SubCommand;
import org.apache.rocketmq.tools.command.SubCommandException;
@@ -48,6 +50,10 @@ public class TopicStatusSubCommand implements SubCommand {
Option opt = new Option("t", "topic", true, "topic name");
opt.setRequired(true);
options.addOption(opt);
opt = new Option("c", "cluster", true, "cluster name or lmq parent topic, lmq is used to find the route.");
opt.setRequired(false);
options.addOption(opt);
return options;
}
@@ -58,10 +64,26 @@ public class TopicStatusSubCommand implements SubCommand {
defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
try {
TopicStatsTable topicStatsTable = new TopicStatsTable();
defaultMQAdminExt.start();
String topic = commandLine.getOptionValue('t').trim();
TopicStatsTable topicStatsTable = defaultMQAdminExt.examineTopicStats(topic);
if (commandLine.hasOption('c')) {
String cluster = commandLine.getOptionValue('c').trim();
TopicRouteData topicRouteData = defaultMQAdminExt.examineTopicRouteInfo(cluster);
for (BrokerData bd : topicRouteData.getBrokerDatas()) {
String addr = bd.selectBrokerAddr();
if (addr != null) {
TopicStatsTable tst = defaultMQAdminExt.examineTopicStats(addr, topic);
topicStatsTable.getOffsetTable().putAll(tst.getOffsetTable());
}
}
} else {
topicStatsTable = defaultMQAdminExt.examineTopicStats(topic);
}
List<MessageQueue> mqList = new LinkedList<>();
mqList.addAll(topicStatsTable.getOffsetTable().keySet());
tools/src/test/java/org/apache/rocketmq/tools/command/message/QueryMsgByUniqueKeySubCommandTest.java
+6 -6
View File
@@ -127,7 +127,7 @@ public class QueryMsgByUniqueKeySubCommandTest {
when(mQAdminImpl.queryMessageByUniqKey(anyString(), anyString())).thenReturn(retMsgExt);
QueryResult queryResult = new QueryResult(0, Lists.newArrayList(retMsgExt));
when(defaultMQAdminExtImpl.queryMessageByUniqKey(anyString(), anyString(), anyInt(), anyLong(), anyLong())).thenReturn(queryResult);
when(mQAdminImpl.queryMessageByUniqKey(anyString(), anyString(), anyString(), anyInt(), anyLong(), anyLong())).thenReturn(queryResult);
TopicRouteData topicRouteData = new TopicRouteData();
List<BrokerData> brokerDataList = new ArrayList<>();
@@ -194,7 +194,7 @@ public class QueryMsgByUniqueKeySubCommandTest {
Options options = ServerUtil.buildCommandlineOptions(new Options());
String[] args = new String[] {"-t myTopicTest", "-i msgId"};
String[] args = new String[] {"-t myTopicTest", "-i msgId", "-c DefaultCluster"};
CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin ", args,
cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
@@ -218,7 +218,7 @@ public class QueryMsgByUniqueKeySubCommandTest {
Options options = ServerUtil.buildCommandlineOptions(new Options());
String[] args = new String[] {"-t myTopicTest", "-i 7F000001000004D20000000000000066"};
String[] args = new String[] {"-t myTopicTest", "-i 7F000001000004D20000000000000066", "-c DefaultCluster"};
CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin ", args,
cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
@@ -230,7 +230,7 @@ public class QueryMsgByUniqueKeySubCommandTest {
Options options = ServerUtil.buildCommandlineOptions(new Options());
String[] args = new String[] {"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000", "-g producerGroupName", "-d clientId"};
String[] args = new String[] {"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000", "-g producerGroupName", "-d clientId", "-c DefaultCluster"};
CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin ", args,
cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
@@ -241,13 +241,13 @@ public class QueryMsgByUniqueKeySubCommandTest {
System.setProperty("rocketmq.namesrv.addr", "127.0.0.1:9876");
String[] args = new String[]{"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000"};
String[] args = new String[]{"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000", "-c DefaultCluster"};
Options options = ServerUtil.buildCommandlineOptions(new Options());
CommandLine commandLine = ServerUtil.parseCmdLine("mqadmin ", args,
cmd.buildCommandlineOptions(options), new DefaultParser());
cmd.execute(commandLine, options, null);
args = new String[] {"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000", "-g producerGroupName", "-d clientId"};
args = new String[] {"-t myTopicTest", "-i 0A3A54F7BF7D18B4AAC28A3FA2CF0000", "-g producerGroupName", "-d clientId", "-c DefaultCluster"};
commandLine = ServerUtil.parseCmdLine("mqadmin ", args, cmd.buildCommandlineOptions(options),
new DefaultParser());
cmd.execute(commandLine, options, null);