subscription) {
/**
* Send message back to broker which will be re-delivered in future.
- *
+ *
* This method will be removed or it's visibility will be changed in a certain version after April 5, 2020, so
* please do not use this method.
*
- * @param msg Message to send back.
+ * @param msg Message to send back.
* @param delayLevel delay level.
- * @throws RemotingException if there is any network-tier error.
- * @throws MQBrokerException if there is any broker error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws MQBrokerException if there is any broker error.
* @throws InterruptedException if the thread is interrupted.
- * @throws MQClientException if there is any client error.
+ * @throws MQClientException if there is any client error.
*/
@Deprecated
@Override
public void sendMessageBack(MessageExt msg, int delayLevel)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
msg.setTopic(withNamespace(msg.getTopic()));
- this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, (String) null);
+ this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, msg.getBrokerName());
}
/**
* Send message back to the broker whose name is brokerName and the message will be re-delivered in
* future.
- *
+ *
* This method will be removed or it's visibility will be changed in a certain version after April 5, 2020, so
* please do not use this method.
*
- * @param msg Message to send back.
+ * @param msg Message to send back.
* @param delayLevel delay level.
* @param brokerName broker name.
- * @throws RemotingException if there is any network-tier error.
- * @throws MQBrokerException if there is any broker error.
+ * @throws RemotingException if there is any network-tier error.
+ * @throws MQBrokerException if there is any broker error.
* @throws InterruptedException if the thread is interrupted.
- * @throws MQClientException if there is any client error.
+ * @throws MQClientException if there is any client error.
*/
@Deprecated
@Override
@@ -735,7 +730,7 @@ public void start() throws MQClientException {
this.defaultMQPushConsumerImpl.start();
if (enableTrace) {
try {
- AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(consumerGroup, TraceDispatcher.Type.CONSUME, traceTopic, rpcHook);
+ AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(consumerGroup, TraceDispatcher.Type.CONSUME, getTraceMsgBatchNum(), traceTopic, rpcHook);
dispatcher.setHostConsumer(this.defaultMQPushConsumerImpl);
dispatcher.setNamespaceV2(namespaceV2);
traceDispatcher = dispatcher;
@@ -799,9 +794,9 @@ public void registerMessageListener(MessageListenerOrderly messageListener) {
/**
* Subscribe a topic to consuming subscription.
*
- * @param topic topic to subscribe.
+ * @param topic topic to subscribe.
* @param subExpression subscription expression.it only support or operation such as "tag1 || tag2 || tag3"
- * if null or * expression,meaning subscribe all
+ * if null or * expression,meaning subscribe all
* @throws MQClientException if there is any client error.
*/
@Override
@@ -812,8 +807,8 @@ public void subscribe(String topic, String subExpression) throws MQClientExcepti
/**
* Subscribe a topic to consuming subscription.
*
- * @param topic topic to consume.
- * @param fullClassName full class name,must extend org.apache.rocketmq.common.filter. MessageFilter
+ * @param topic topic to consume.
+ * @param fullClassName full class name,must extend org.apache.rocketmq.common.filter. MessageFilter
* @param filterClassSource class source code,used UTF-8 file encoding,must be responsible for your code safety
*/
@Override
@@ -824,7 +819,7 @@ public void subscribe(String topic, String fullClassName, String filterClassSour
/**
* Subscribe a topic by message selector.
*
- * @param topic topic to consume.
+ * @param topic topic to consume.
* @param messageSelector {@link org.apache.rocketmq.client.consumer.MessageSelector}
* @see org.apache.rocketmq.client.consumer.MessageSelector#bySql
* @see org.apache.rocketmq.client.consumer.MessageSelector#byTag
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
index bcfe29bd4f6..c1e3ee33dc1 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQAdminImpl.java
@@ -28,6 +28,7 @@
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.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 long searchOffset(MessageQueue mq, long timestamp, BoundaryType boundaryT
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 MessageExt viewMessage(String topic, String msgId)
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 MessageExt queryMessageByUniqKey(String clusterName, String topic,
}
}
- 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);
+
+ 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;
+ }
- 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);
+ 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 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 @@ protected QueryResult queryMessage(String clusterName, String topic, String key,
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 void operationFail(Throwable throwable) {
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;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
index f3d7e7c70f9..b539b8f098a 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/MQClientAPIImpl.java
@@ -78,6 +78,7 @@
import org.apache.rocketmq.common.namesrv.TopAddressing;
import org.apache.rocketmq.common.sysflag.PullSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
+import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.ChannelEventListener;
@@ -137,6 +138,7 @@
import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
import org.apache.rocketmq.remoting.protocol.body.SetMessageRequestModeRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupList;
import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
@@ -183,9 +185,9 @@
import org.apache.rocketmq.remoting.protocol.header.GetTopicStatsInfoRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetTopicsByClusterRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.GetUserRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.HeartbeatRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ListAclsRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ListUsersRequestHeader;
-import org.apache.rocketmq.remoting.protocol.header.HeartbeatRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.LockBatchMqRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageResponseHeader;
@@ -246,7 +248,7 @@
import static org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode.SUCCESS;
-public class MQClientAPIImpl implements NameServerUpdateCallback {
+public class MQClientAPIImpl implements NameServerUpdateCallback, StartAndShutdown {
private final static Logger log = LoggerFactory.getLogger(MQClientAPIImpl.class);
private static boolean sendSmartMsg =
Boolean.parseBoolean(System.getProperty("org.apache.rocketmq.client.sendSmartMsg", "true"));
@@ -400,6 +402,22 @@ public void createSubscriptionGroup(final String addr, final SubscriptionGroupCo
}
+ public void createSubscriptionGroupList(final String address, final List configs,
+ final long timeoutMillis) throws RemotingException, InterruptedException, MQClientException {
+ RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_AND_CREATE_SUBSCRIPTIONGROUP_LIST, null);
+ SubscriptionGroupList requestBody = new SubscriptionGroupList(configs);
+ request.setBody(requestBody.encode());
+
+ RemotingCommand response = this.remotingClient.invokeSync(
+ MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), address), request, timeoutMillis);
+ assert response != null;
+ if (response.getCode() == ResponseCode.SUCCESS) {
+ return;
+ }
+
+ throw new MQClientException(response.getCode(), response.getRemark());
+ }
+
public void createTopic(final String addr, final String defaultTopic, final TopicConfig topicConfig,
final long timeoutMillis)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java
index 3713d1aba4d..d5191871106 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopConcurrentlyService.java
@@ -471,7 +471,7 @@ public void run() {
processQueue.decFoundMsg(-msgs.size());
}
- log.warn("processQueue invalid. isDropped={}, isPopTimeout={}, messageQueue={}, msgs={}",
+ log.warn("processQueue invalid or popTimeout. isDropped={}, isPopTimeout={}, messageQueue={}, msgs={}",
processQueue.isDropped(), isPopTimeout(), messageQueue, msgs);
}
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
index a3276cd7823..3f90b67ec99 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultLitePullConsumerImpl.java
@@ -164,10 +164,6 @@ private enum SubscriptionType {
public DefaultLitePullConsumerImpl(final DefaultLitePullConsumer defaultLitePullConsumer, final RPCHook rpcHook) {
this.defaultLitePullConsumer = defaultLitePullConsumer;
this.rpcHook = rpcHook;
- this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
- this.defaultLitePullConsumer.getPullThreadNums(),
- new ThreadFactoryImpl("PullMsgThread-" + this.defaultLitePullConsumer.getConsumerGroup())
- );
this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("MonitorMessageQueueChangeThread"));
this.pullTimeDelayMillsWhenException = defaultLitePullConsumer.getPullTimeDelayMillsWhenException();
}
@@ -293,6 +289,8 @@ public synchronized void start() throws MQClientException {
this.defaultLitePullConsumer.changeInstanceNameToPID();
}
+ initScheduledThreadPoolExecutor();
+
initMQClientFactory();
initRebalanceImpl();
@@ -324,6 +322,13 @@ public synchronized void start() throws MQClientException {
}
}
+ private void initScheduledThreadPoolExecutor() {
+ this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(
+ this.defaultLitePullConsumer.getPullThreadNums(),
+ new ThreadFactoryImpl("PullMsgThread-" + this.defaultLitePullConsumer.getConsumerGroup())
+ );
+ }
+
private void initMQClientFactory() throws MQClientException {
this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultLitePullConsumer, this.rpcHook);
boolean registerOK = mQClientFactory.registerConsumer(this.defaultLitePullConsumer.getConsumerGroup(), this);
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
index 3e832e5a9a3..c92cadf5057 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPushConsumerImpl.java
@@ -621,10 +621,9 @@ public void onException(Throwable e) {
private PopResult processPopResult(final PopResult popResult, final SubscriptionData subscriptionData) {
if (PopStatus.FOUND == popResult.getPopStatus()) {
List msgFoundList = popResult.getMsgFoundList();
- List msgListFilterAgain = msgFoundList;
+ List msgListFilterAgain = new ArrayList<>(popResult.getMsgFoundList().size());
if (!subscriptionData.getTagsSet().isEmpty() && !subscriptionData.isClassFilterMode()
&& popResult.getMsgFoundList().size() > 0) {
- msgListFilterAgain = new ArrayList<>(popResult.getMsgFoundList().size());
for (MessageExt msg : popResult.getMsgFoundList()) {
if (msg.getTags() != null) {
if (subscriptionData.getTagsSet().contains(msg.getTags())) {
@@ -632,6 +631,8 @@ private PopResult processPopResult(final PopResult popResult, final Subscription
}
}
}
+ } else {
+ msgListFilterAgain.addAll(msgFoundList);
}
if (!this.filterMessageHookList.isEmpty()) {
@@ -649,6 +650,15 @@ private PopResult processPopResult(final PopResult popResult, final Subscription
}
}
+ Iterator iterator = msgListFilterAgain.iterator();
+ while (iterator.hasNext()) {
+ MessageExt msg = iterator.next();
+ if (msg.getReconsumeTimes() > getMaxReconsumeTimes()) {
+ iterator.remove();
+ log.info("Reconsume times has reached {}, so ack msg={}", msg.getReconsumeTimes(), msg);
+ }
+ }
+
if (msgFoundList.size() != msgListFilterAgain.size()) {
for (MessageExt msg : msgFoundList) {
if (!msgListFilterAgain.contains(msg)) {
@@ -742,7 +752,7 @@ public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerN
public void sendMessageBack(MessageExt msg, int delayLevel, final MessageQueue mq)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
- sendMessageBack(msg, delayLevel, null, mq);
+ sendMessageBack(msg, delayLevel, msg.getBrokerName(), mq);
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
index b4ebf692736..c9fd3c83e04 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/factory/MQClientInstance.java
@@ -152,6 +152,7 @@ public MQClientInstance(ClientConfig clientConfig, int instanceIndex, String cli
this.nettyClientConfig.setClientCallbackExecutorThreads(clientConfig.getClientCallbackExecutorThreads());
this.nettyClientConfig.setUseTLS(clientConfig.isUseTLS());
this.nettyClientConfig.setSocksProxyConfig(clientConfig.getSocksProxyConfig());
+ this.nettyClientConfig.setScanAvailableNameSrv(false);
ClientRemotingProcessor clientRemotingProcessor = new ClientRemotingProcessor(this);
ChannelEventListener channelEventListener;
if (clientConfig.isEnableHeartbeatChannelEventListener()) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java
index b97e00c577f..0e2092b8a0f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIExt.java
@@ -400,6 +400,33 @@ public CompletableFuture updateConsumerOffsetOneWay(
return future;
}
+ public CompletableFuture updateConsumerOffsetAsync(
+ String brokerAddr,
+ UpdateConsumerOffsetRequestHeader header,
+ long timeoutMillis
+ ) {
+ RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UPDATE_CONSUMER_OFFSET, header);
+ CompletableFuture future = new CompletableFuture<>();
+ invoke(brokerAddr, request, timeoutMillis).whenComplete((response, t) -> {
+ if (t != null) {
+ log.error("updateConsumerOffsetAsync failed, brokerAddr={}, requestHeader={}", brokerAddr, header, t);
+ future.completeExceptionally(t);
+ return;
+ }
+ switch (response.getCode()) {
+ case ResponseCode.SUCCESS: {
+ future.complete(null);
+ }
+ case ResponseCode.SYSTEM_ERROR:
+ case ResponseCode.SUBSCRIPTION_GROUP_NOT_EXIST:
+ case ResponseCode.TOPIC_NOT_EXIST: {
+ future.completeExceptionally(new MQBrokerException(response.getCode(), response.getRemark()));
+ }
+ }
+ });
+ return future;
+ }
+
public CompletableFuture> getConsumerListByGroupAsync(
String brokerAddr,
GetConsumerListByGroupRequestHeader requestHeader,
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
index c68859b2889..0fa31b66406 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/mqclient/MQClientAPIFactory.java
@@ -26,6 +26,7 @@
import org.apache.rocketmq.client.common.NameserverAccessConfig;
import org.apache.rocketmq.client.impl.ClientRemotingProcessor;
import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.utils.AsyncShutdownHelper;
import org.apache.rocketmq.common.utils.StartAndShutdown;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
@@ -85,9 +86,11 @@ public void start() throws Exception {
@Override
public void shutdown() throws Exception {
+ AsyncShutdownHelper helper = new AsyncShutdownHelper();
for (int i = 0; i < this.clientNum; i++) {
- clients[i].shutdown();
+ helper.addTarget(clients[i]);
}
+ helper.shutdown().await(Integer.MAX_VALUE, TimeUnit.SECONDS);
}
protected MQClientAPIExt createAndStart(String instanceName) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
index 7ef34025137..74a2516174a 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/producer/DefaultMQProducerImpl.java
@@ -194,6 +194,14 @@ public void setSemaphoreAsyncSendSize(int size) {
semaphoreAsyncSendSize = new Semaphore(size, true);
}
+ public int getSemaphoreAsyncSendNumAvailablePermits() {
+ return semaphoreAsyncSendNum == null ? 0 : semaphoreAsyncSendNum.availablePermits();
+ }
+
+ public int getSemaphoreAsyncSendSizeAvailablePermits() {
+ return semaphoreAsyncSendSize == null ? 0 : semaphoreAsyncSendSize.availablePermits();
+ }
+
public void initTransactionEnv() {
TransactionMQProducer producer = (TransactionMQProducer) this.defaultMQProducer;
if (producer.getExecutorService() != null) {
@@ -563,7 +571,7 @@ public void run() {
class BackpressureSendCallBack implements SendCallback {
public boolean isSemaphoreAsyncSizeAcquired = false;
- public boolean isSemaphoreAsyncNumbAcquired = false;
+ public boolean isSemaphoreAsyncNumAcquired = false;
public int msgLen;
private final SendCallback sendCallback;
@@ -573,24 +581,49 @@ public BackpressureSendCallBack(final SendCallback sendCallback) {
@Override
public void onSuccess(SendResult sendResult) {
- if (isSemaphoreAsyncSizeAcquired) {
- semaphoreAsyncSendSize.release(msgLen);
- }
- if (isSemaphoreAsyncNumbAcquired) {
- semaphoreAsyncSendNum.release();
- }
+ semaphoreProcessor();
sendCallback.onSuccess(sendResult);
}
@Override
public void onException(Throwable e) {
+ semaphoreProcessor();
+ sendCallback.onException(e);
+ }
+
+ public void semaphoreProcessor() {
if (isSemaphoreAsyncSizeAcquired) {
+ defaultMQProducer.acquireBackPressureForAsyncSendSizeLock();
semaphoreAsyncSendSize.release(msgLen);
+ defaultMQProducer.releaseBackPressureForAsyncSendSizeLock();
}
- if (isSemaphoreAsyncNumbAcquired) {
+ if (isSemaphoreAsyncNumAcquired) {
+ defaultMQProducer.acquireBackPressureForAsyncSendNumLock();
semaphoreAsyncSendNum.release();
+ defaultMQProducer.releaseBackPressureForAsyncSendNumLock();
}
- sendCallback.onException(e);
+ }
+
+ public void semaphoreAsyncAdjust(int semaphoreAsyncNum, int semaphoreAsyncSize) throws InterruptedException {
+ defaultMQProducer.acquireBackPressureForAsyncSendNumLock();
+ if (semaphoreAsyncNum > 0) {
+ semaphoreAsyncSendNum.release(semaphoreAsyncNum);
+ } else {
+ semaphoreAsyncSendNum.acquire(- semaphoreAsyncNum);
+ }
+ defaultMQProducer.setBackPressureForAsyncSendNumInsideAdjust(defaultMQProducer.getBackPressureForAsyncSendNum()
+ + semaphoreAsyncNum);
+ defaultMQProducer.releaseBackPressureForAsyncSendNumLock();
+
+ defaultMQProducer.acquireBackPressureForAsyncSendSizeLock();
+ if (semaphoreAsyncSize > 0) {
+ semaphoreAsyncSendSize.release(semaphoreAsyncSize);
+ } else {
+ semaphoreAsyncSendSize.acquire(- semaphoreAsyncSize);
+ }
+ defaultMQProducer.setBackPressureForAsyncSendSizeInsideAdjust(defaultMQProducer.getBackPressureForAsyncSendSize()
+ + semaphoreAsyncSize);
+ defaultMQProducer.releaseBackPressureForAsyncSendSizeLock();
}
}
@@ -599,32 +632,40 @@ public void executeAsyncMessageSend(Runnable runnable, final Message msg, final
throws MQClientException, InterruptedException {
ExecutorService executor = this.getAsyncSenderExecutor();
boolean isEnableBackpressureForAsyncMode = this.getDefaultMQProducer().isEnableBackpressureForAsyncMode();
- boolean isSemaphoreAsyncNumbAcquired = false;
+ boolean isSemaphoreAsyncNumAcquired = false;
boolean isSemaphoreAsyncSizeAcquired = false;
int msgLen = msg.getBody() == null ? 1 : msg.getBody().length;
+ sendCallback.msgLen = msgLen;
try {
if (isEnableBackpressureForAsyncMode) {
+ defaultMQProducer.acquireBackPressureForAsyncSendNumLock();
long costTime = System.currentTimeMillis() - beginStartTime;
- isSemaphoreAsyncNumbAcquired = timeout - costTime > 0
+
+ isSemaphoreAsyncNumAcquired = timeout - costTime > 0
&& semaphoreAsyncSendNum.tryAcquire(timeout - costTime, TimeUnit.MILLISECONDS);
- if (!isSemaphoreAsyncNumbAcquired) {
+ sendCallback.isSemaphoreAsyncNumAcquired = isSemaphoreAsyncNumAcquired;
+ defaultMQProducer.releaseBackPressureForAsyncSendNumLock();
+ if (!isSemaphoreAsyncNumAcquired) {
sendCallback.onException(
new RemotingTooMuchRequestException("send message tryAcquire semaphoreAsyncNum timeout"));
return;
}
+
+ defaultMQProducer.acquireBackPressureForAsyncSendSizeLock();
costTime = System.currentTimeMillis() - beginStartTime;
+
isSemaphoreAsyncSizeAcquired = timeout - costTime > 0
&& semaphoreAsyncSendSize.tryAcquire(msgLen, timeout - costTime, TimeUnit.MILLISECONDS);
+ sendCallback.isSemaphoreAsyncSizeAcquired = isSemaphoreAsyncSizeAcquired;
+ defaultMQProducer.releaseBackPressureForAsyncSendSizeLock();
if (!isSemaphoreAsyncSizeAcquired) {
sendCallback.onException(
new RemotingTooMuchRequestException("send message tryAcquire semaphoreAsyncSize timeout"));
return;
}
}
- sendCallback.isSemaphoreAsyncSizeAcquired = isSemaphoreAsyncSizeAcquired;
- sendCallback.isSemaphoreAsyncNumbAcquired = isSemaphoreAsyncNumbAcquired;
- sendCallback.msgLen = msgLen;
+
executor.submit(runnable);
} catch (RejectedExecutionException e) {
if (isEnableBackpressureForAsyncMode) {
@@ -969,7 +1010,7 @@ private SendResult sendKernelImpl(final Message msg,
boolean messageCloned = false;
if (msgBodyCompressed) {
//If msg body was compressed, msgbody should be reset using prevBody.
- //Clone new message using commpressed message body and recover origin massage.
+ //Clone new message using compressed message body and recover origin massage.
//Fix bug:https://github.com/apache/rocketmq-externals/issues/66
tmpMessage = MessageAccessor.cloneMessage(msg);
messageCloned = true;
@@ -1538,6 +1579,7 @@ public Message request(final Message msg,
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
+ requestResponseFuture.acquireCountDownLatch();
}
@Override
@@ -1595,6 +1637,7 @@ public Message request(final Message msg, final MessageQueueSelector selector, f
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
+ requestResponseFuture.acquireCountDownLatch();
}
@Override
@@ -1652,6 +1695,7 @@ public Message request(final Message msg, final MessageQueue mq, final long time
@Override
public void onSuccess(SendResult sendResult) {
requestResponseFuture.setSendRequestOk(true);
+ requestResponseFuture.acquireCountDownLatch();
}
@Override
diff --git a/client/src/main/java/org/apache/rocketmq/client/lock/ReadWriteCASLock.java b/client/src/main/java/org/apache/rocketmq/client/lock/ReadWriteCASLock.java
new file mode 100644
index 00000000000..3d157313715
--- /dev/null
+++ b/client/src/main/java/org/apache/rocketmq/client/lock/ReadWriteCASLock.java
@@ -0,0 +1,63 @@
+/*
+ * 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.client.lock;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ReadWriteCASLock {
+ //true : can lock ; false : not lock
+ private final AtomicBoolean writeLock = new AtomicBoolean(true);
+
+ private final AtomicInteger readLock = new AtomicInteger(0);
+
+ public void acquireWriteLock() {
+ boolean isLock = false;
+ do {
+ isLock = writeLock.compareAndSet(true, false);
+ } while (!isLock);
+
+ do {
+ isLock = readLock.get() == 0;
+ } while (!isLock);
+ }
+
+ public void releaseWriteLock() {
+ this.writeLock.compareAndSet(false, true);
+ }
+
+ public void acquireReadLock() {
+ boolean isLock = false;
+ do {
+ isLock = writeLock.get();
+ } while (!isLock);
+ readLock.getAndIncrement();
+ }
+
+ public void releaseReadLock() {
+ this.readLock.getAndDecrement();
+ }
+
+ public boolean getWriteLock() {
+ return this.writeLock.get() && this.readLock.get() == 0;
+ }
+
+ public boolean getReadLock() {
+ return this.writeLock.get();
+ }
+
+}
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
index 4fd038663b5..f0842de8ba7 100644
--- a/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/DefaultMQProducer.java
@@ -24,6 +24,7 @@
import org.apache.rocketmq.client.exception.RequestTimeoutException;
import org.apache.rocketmq.client.impl.MQClientManager;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.lock.ReadWriteCASLock;
import org.apache.rocketmq.client.trace.AsyncTraceDispatcher;
import org.apache.rocketmq.client.trace.TraceDispatcher;
import org.apache.rocketmq.client.trace.hook.EndTransactionTraceHookImpl;
@@ -79,7 +80,8 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
ResponseCode.SYSTEM_BUSY,
ResponseCode.NO_PERMISSION,
ResponseCode.NO_BUYER_ID,
- ResponseCode.NOT_IN_CURRENT_UNIT
+ ResponseCode.NOT_IN_CURRENT_UNIT,
+ ResponseCode.GO_AWAY
));
/**
@@ -174,6 +176,16 @@ public class DefaultMQProducer extends ClientConfig implements MQProducer {
private RPCHook rpcHook = null;
+ /**
+ * backPressureForAsyncSendNum is guaranteed to be modified at runtime and no new requests are allowed
+ */
+ private final ReadWriteCASLock backPressureForAsyncSendNumLock = new ReadWriteCASLock();
+
+ /**
+ * backPressureForAsyncSendSize is guaranteed to be modified at runtime and no new requests are allowed
+ */
+ private final ReadWriteCASLock backPressureForAsyncSendSizeLock = new ReadWriteCASLock();
+
/**
* Compress level of compress algorithm.
*/
@@ -348,7 +360,7 @@ public void start() throws MQClientException {
}
if (enableTrace) {
try {
- AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(producerGroup, TraceDispatcher.Type.PRODUCE, traceTopic, rpcHook);
+ AsyncTraceDispatcher dispatcher = new AsyncTraceDispatcher(producerGroup, TraceDispatcher.Type.PRODUCE, getTraceMsgBatchNum(), traceTopic, rpcHook);
dispatcher.setHostProducer(this.defaultMQProducerImpl);
dispatcher.setNamespaceV2(this.namespaceV2);
traceDispatcher = dispatcher;
@@ -1333,18 +1345,64 @@ public int getBackPressureForAsyncSendNum() {
return backPressureForAsyncSendNum;
}
+ /**
+ * For user modify backPressureForAsyncSendNum at runtime
+ */
public void setBackPressureForAsyncSendNum(int backPressureForAsyncSendNum) {
+ this.backPressureForAsyncSendNumLock.acquireWriteLock();
+ backPressureForAsyncSendNum = Math.max(backPressureForAsyncSendNum, 10);
+ int acquiredBackPressureForAsyncSendNum = this.backPressureForAsyncSendNum
+ - defaultMQProducerImpl.getSemaphoreAsyncSendNumAvailablePermits();
this.backPressureForAsyncSendNum = backPressureForAsyncSendNum;
- defaultMQProducerImpl.setSemaphoreAsyncSendNum(backPressureForAsyncSendNum);
+ defaultMQProducerImpl.setSemaphoreAsyncSendNum(backPressureForAsyncSendNum - acquiredBackPressureForAsyncSendNum);
+ this.backPressureForAsyncSendNumLock.releaseWriteLock();
}
public int getBackPressureForAsyncSendSize() {
return backPressureForAsyncSendSize;
}
+ /**
+ * For user modify backPressureForAsyncSendSize at runtime
+ */
public void setBackPressureForAsyncSendSize(int backPressureForAsyncSendSize) {
+ this.backPressureForAsyncSendSizeLock.acquireWriteLock();
+ backPressureForAsyncSendSize = Math.max(backPressureForAsyncSendSize, 1024 * 1024);
+ int acquiredBackPressureForAsyncSendSize = this.backPressureForAsyncSendSize
+ - defaultMQProducerImpl.getSemaphoreAsyncSendSizeAvailablePermits();
+ this.backPressureForAsyncSendSize = backPressureForAsyncSendSize;
+ defaultMQProducerImpl.setSemaphoreAsyncSendSize(backPressureForAsyncSendSize - acquiredBackPressureForAsyncSendSize);
+ this.backPressureForAsyncSendSizeLock.releaseWriteLock();
+ }
+
+ /**
+ * Used for system internal adjust backPressureForAsyncSendSize
+ */
+ public void setBackPressureForAsyncSendSizeInsideAdjust(int backPressureForAsyncSendSize) {
this.backPressureForAsyncSendSize = backPressureForAsyncSendSize;
- defaultMQProducerImpl.setSemaphoreAsyncSendSize(backPressureForAsyncSendSize);
+ }
+
+ /**
+ * Used for system internal adjust backPressureForAsyncSendNum
+ */
+ public void setBackPressureForAsyncSendNumInsideAdjust(int backPressureForAsyncSendNum) {
+ this.backPressureForAsyncSendNum = backPressureForAsyncSendNum;
+ }
+
+ public void acquireBackPressureForAsyncSendSizeLock() {
+ this.backPressureForAsyncSendSizeLock.acquireReadLock();
+ }
+
+ public void releaseBackPressureForAsyncSendSizeLock() {
+ this.backPressureForAsyncSendSizeLock.releaseReadLock();
+ }
+
+ public void acquireBackPressureForAsyncSendNumLock() {
+ this.backPressureForAsyncSendNumLock.acquireReadLock();
+ }
+
+ public void releaseBackPressureForAsyncSendNumLock() {
+ this.backPressureForAsyncSendNumLock.releaseReadLock();
}
public List getTopics() {
diff --git a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
index e66c08fdc53..203f92907a4 100644
--- a/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
+++ b/client/src/main/java/org/apache/rocketmq/client/producer/RequestResponseFuture.java
@@ -107,6 +107,10 @@ public void setSendRequestOk(boolean sendRequestOk) {
this.sendRequestOk = sendRequestOk;
}
+ public void acquireCountDownLatch() {
+ this.countDownLatch.countDown();
+ }
+
public Message getRequestMsg() {
return requestMsg;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
index 1fe19773a5a..e321e1583d2 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/AsyncTraceDispatcher.java
@@ -16,19 +16,6 @@
*/
package org.apache.rocketmq.client.trace;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.common.ThreadLocalIndex;
import org.apache.rocketmq.client.exception.MQClientException;
@@ -44,68 +31,75 @@
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.topic.TopicValidator;
-import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
+import org.apache.rocketmq.remoting.RPCHook;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import static org.apache.rocketmq.client.trace.TraceConstants.TRACE_INSTANCE_NAME;
public class AsyncTraceDispatcher implements TraceDispatcher {
- private final static Logger log = LoggerFactory.getLogger(AsyncTraceDispatcher.class);
- private final static AtomicInteger COUNTER = new AtomicInteger();
- private final static short MAX_MSG_KEY_SIZE = Short.MAX_VALUE - 10000;
- private final int queueSize;
- private final int batchSize;
+ private static final Logger log = LoggerFactory.getLogger(AsyncTraceDispatcher.class);
+ private static final AtomicInteger COUNTER = new AtomicInteger();
+ private static final AtomicInteger INSTANCE_NUM = new AtomicInteger(0);
+ private volatile boolean stopped = false;
+ private final int traceInstanceId = INSTANCE_NUM.getAndIncrement();
+ private final int batchNum;
private final int maxMsgSize;
- private final long pollingTimeMil;
- private final long waitTimeThresholdMil;
private final DefaultMQProducer traceProducer;
- private final ThreadPoolExecutor traceExecutor;
- // The last discard number of log
private AtomicLong discardCount;
private Thread worker;
+ private final ThreadPoolExecutor traceExecutor;
private final ArrayBlockingQueue traceContextQueue;
- private final HashMap taskQueueByTopic;
- private ArrayBlockingQueue appenderQueue;
+ private final ArrayBlockingQueue appenderQueue;
private volatile Thread shutDownHook;
- private volatile boolean stopped = false;
+
private DefaultMQProducerImpl hostProducer;
private DefaultMQPushConsumerImpl hostConsumer;
private volatile ThreadLocalIndex sendWhichQueue = new ThreadLocalIndex();
- private String dispatcherId = UUID.randomUUID().toString();
private volatile String traceTopicName;
private AtomicBoolean isStarted = new AtomicBoolean(false);
private volatile AccessChannel accessChannel = AccessChannel.LOCAL;
private String group;
private Type type;
private String namespaceV2;
+ private final int flushTraceInterval = 5000;
+
+ private long lastFlushTime = System.currentTimeMillis();
- public AsyncTraceDispatcher(String group, Type type, String traceTopicName, RPCHook rpcHook) {
- // queueSize is greater than or equal to the n power of 2 of value
- this.queueSize = 2048;
- this.batchSize = 100;
+ public AsyncTraceDispatcher(String group, Type type, int batchNum, String traceTopicName, RPCHook rpcHook) {
+ this.batchNum = Math.min(batchNum, 20);/* max value 20*/
this.maxMsgSize = 128000;
- this.pollingTimeMil = 100;
- this.waitTimeThresholdMil = 500;
this.discardCount = new AtomicLong(0L);
- this.traceContextQueue = new ArrayBlockingQueue<>(1024);
- this.taskQueueByTopic = new HashMap();
+ this.traceContextQueue = new ArrayBlockingQueue<>(2048);
this.group = group;
this.type = type;
-
- this.appenderQueue = new ArrayBlockingQueue<>(queueSize);
+ this.appenderQueue = new ArrayBlockingQueue<>(2048);
if (!UtilAll.isBlank(traceTopicName)) {
this.traceTopicName = traceTopicName;
} else {
this.traceTopicName = TopicValidator.RMQ_SYS_TRACE_TOPIC;
}
this.traceExecutor = new ThreadPoolExecutor(//
- 10, //
- 20, //
+ 2, //
+ 4, //
1000 * 60, //
TimeUnit.MILLISECONDS, //
this.appenderQueue, //
- new ThreadFactoryImpl("MQTraceSendThread_"));
+ new ThreadFactoryImpl("MQTraceSendThread_" + traceInstanceId + "_"));
traceProducer = getAndCreateTraceProducer(rpcHook);
}
@@ -153,7 +147,6 @@ public void setNamespaceV2(String namespaceV2) {
this.namespaceV2 = namespaceV2;
}
- @Override
public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClientException {
if (isStarted.compareAndSet(false, true)) {
traceProducer.setNamesrvAddr(nameSrvAddr);
@@ -163,7 +156,8 @@ public void start(String nameSrvAddr, AccessChannel accessChannel) throws MQClie
traceProducer.start();
}
this.accessChannel = accessChannel;
- this.worker = new Thread(new AsyncRunnable(), "MQ-AsyncTraceDispatcher-Thread-" + dispatcherId);
+ this.worker = new ThreadFactoryImpl("MQ-AsyncArrayDispatcher-Thread" + traceInstanceId, true)
+ .newThread(new AsyncRunnable());
this.worker.setDaemon(true);
this.worker.start();
this.registerShutDownHook();
@@ -197,37 +191,28 @@ public boolean append(final Object ctx) {
@Override
public void flush() {
- // The maximum waiting time for refresh,avoid being written all the time, resulting in failure to return.
long end = System.currentTimeMillis() + 500;
- while (System.currentTimeMillis() <= end) {
- synchronized (taskQueueByTopic) {
- for (TraceDataSegment taskInfo : taskQueueByTopic.values()) {
- taskInfo.sendAllData();
- }
- }
- synchronized (traceContextQueue) {
- if (traceContextQueue.size() == 0 && appenderQueue.size() == 0) {
- break;
- }
- }
+ while (traceContextQueue.size() > 0 || appenderQueue.size() > 0 && System.currentTimeMillis() <= end) {
try {
- Thread.sleep(1);
- } catch (InterruptedException e) {
- break;
+ flushTraceContext(true);
+ } catch (Throwable throwable) {
+ log.error("flushTraceContext error", throwable);
}
}
- log.info("------end trace send " + traceContextQueue.size() + " " + appenderQueue.size());
+ if (appenderQueue.size() > 0) {
+ log.error("There are still some traces that haven't been sent " + traceContextQueue.size() + " " + appenderQueue.size());
+ }
}
@Override
public void shutdown() {
- this.stopped = true;
flush();
this.traceExecutor.shutdown();
if (isStarted.get()) {
traceProducer.shutdown();
}
this.removeShutdownHook();
+ stopped = true;
}
public void registerShutDownHook() {
@@ -259,152 +244,121 @@ public void removeShutdownHook() {
}
class AsyncRunnable implements Runnable {
- private boolean stopped;
+ private volatile boolean stopped = false;
@Override
public void run() {
while (!stopped) {
- synchronized (traceContextQueue) {
- long endTime = System.currentTimeMillis() + pollingTimeMil;
- while (System.currentTimeMillis() < endTime) {
- try {
- TraceContext traceContext = traceContextQueue.poll(
- endTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS
- );
-
- if (traceContext != null && !traceContext.getTraceBeans().isEmpty()) {
- // get the topic which the trace message will send to
- String traceTopicName = this.getTraceTopicName(traceContext.getRegionId());
-
- // get the traceDataSegment which will save this trace message, create if null
- TraceDataSegment traceDataSegment = taskQueueByTopic.get(traceTopicName);
- if (traceDataSegment == null) {
- traceDataSegment = new TraceDataSegment(traceTopicName, traceContext.getRegionId());
- taskQueueByTopic.put(traceTopicName, traceDataSegment);
- }
-
- // encode traceContext and save it into traceDataSegment
- // NOTE if data size in traceDataSegment more than maxMsgSize,
- // a AsyncDataSendTask will be created and submitted
- TraceTransferBean traceTransferBean = TraceDataEncoder.encoderFromContextBean(traceContext);
- traceDataSegment.addTraceTransferBean(traceTransferBean);
- }
- } catch (InterruptedException ignore) {
- log.debug("traceContextQueue#poll exception");
- }
- }
-
- // NOTE send the data in traceDataSegment which the first TraceTransferBean
- // is longer than waitTimeThreshold
- sendDataByTimeThreshold();
-
- if (AsyncTraceDispatcher.this.stopped) {
- this.stopped = true;
- }
+ try {
+ flushTraceContext(false);
+ } catch (Throwable e) {
+ log.error("flushTraceContext error", e);
}
}
-
+ if (AsyncTraceDispatcher.this.stopped) {
+ this.stopped = true;
+ }
}
+ }
- private void sendDataByTimeThreshold() {
- long now = System.currentTimeMillis();
- for (TraceDataSegment taskInfo : taskQueueByTopic.values()) {
- if (now - taskInfo.firstBeanAddTime >= waitTimeThresholdMil) {
- taskInfo.sendAllData();
+ private void flushTraceContext(boolean forceFlush) throws InterruptedException {
+ List contextList = new ArrayList<>(batchNum);
+ int size = traceContextQueue.size();
+ if (size != 0) {
+ if (forceFlush || size >= batchNum || System.currentTimeMillis() - lastFlushTime > flushTraceInterval) {
+ for (int i = 0; i < batchNum; i++) {
+ TraceContext context = traceContextQueue.poll();
+ if (context != null) {
+ contextList.add(context);
+ } else {
+ break;
+ }
}
+ asyncSendTraceMessage(contextList);
+ return;
}
}
+ // To prevent an infinite loop, add a wait time between each two task executions
+ Thread.sleep(5);
+ }
- private String getTraceTopicName(String regionId) {
- AccessChannel accessChannel = AsyncTraceDispatcher.this.getAccessChannel();
- if (AccessChannel.CLOUD == accessChannel) {
- return TraceConstants.TRACE_TOPIC_PREFIX + regionId;
- }
-
- return AsyncTraceDispatcher.this.getTraceTopicName();
- }
+ private void asyncSendTraceMessage(List contextList) {
+ AsyncDataSendTask request = new AsyncDataSendTask(contextList);
+ traceExecutor.submit(request);
+ lastFlushTime = System.currentTimeMillis();
}
- class TraceDataSegment {
- private long firstBeanAddTime;
- private int currentMsgSize;
- private int currentMsgKeySize;
- private final String traceTopicName;
- private final String regionId;
- private final List traceTransferBeanList = new ArrayList<>();
+ class AsyncDataSendTask implements Runnable {
+ private final List contextList;
- TraceDataSegment(String traceTopicName, String regionId) {
- this.traceTopicName = traceTopicName;
- this.regionId = regionId;
+ public AsyncDataSendTask(List contextList) {
+ this.contextList = contextList;
}
- public void addTraceTransferBean(TraceTransferBean traceTransferBean) {
- initFirstBeanAddTime();
- this.traceTransferBeanList.add(traceTransferBean);
- this.currentMsgSize += traceTransferBean.getTransData().length();
-
- this.currentMsgKeySize = traceTransferBean.getTransKey().stream()
- .reduce(currentMsgKeySize, (acc, x) -> acc + x.length(), Integer::sum);
- if (currentMsgSize >= traceProducer.getMaxMessageSize() - 10 * 1000 || currentMsgKeySize >= MAX_MSG_KEY_SIZE) {
- List dataToSend = new ArrayList<>(traceTransferBeanList);
- AsyncDataSendTask asyncDataSendTask = new AsyncDataSendTask(traceTopicName, regionId, dataToSend);
- traceExecutor.submit(asyncDataSendTask);
- this.clear();
- }
+ @Override
+ public void run() {
+ sendTraceData(contextList);
}
- public void sendAllData() {
- if (this.traceTransferBeanList.isEmpty()) {
- return;
- }
- List dataToSend = new ArrayList<>(traceTransferBeanList);
- AsyncDataSendTask asyncDataSendTask = new AsyncDataSendTask(traceTopicName, regionId, dataToSend);
- traceExecutor.submit(asyncDataSendTask);
-
- this.clear();
- }
+ public void sendTraceData(List contextList) {
+ Map> transBeanMap = new HashMap<>(16);
+ String traceTopic;
+ for (TraceContext context : contextList) {
+ AccessChannel accessChannel = context.getAccessChannel();
+ if (accessChannel == null) {
+ accessChannel = AsyncTraceDispatcher.this.accessChannel;
+ }
+ String currentRegionId = context.getRegionId();
+ if (currentRegionId == null || context.getTraceBeans().isEmpty()) {
+ continue;
+ }
+ if (AccessChannel.CLOUD == accessChannel) {
+ traceTopic = TraceConstants.TRACE_TOPIC_PREFIX + currentRegionId;
+ } else {
+ traceTopic = traceTopicName;
+ }
- private void initFirstBeanAddTime() {
- if (firstBeanAddTime == 0) {
- firstBeanAddTime = System.currentTimeMillis();
+ String topic = context.getTraceBeans().get(0).getTopic();
+ String key = topic + TraceConstants.CONTENT_SPLITOR + traceTopic;
+ List transBeanList = transBeanMap.computeIfAbsent(key, k -> new ArrayList<>());
+ TraceTransferBean traceData = TraceDataEncoder.encoderFromContextBean(context);
+ transBeanList.add(traceData);
+ }
+ for (Map.Entry> entry : transBeanMap.entrySet()) {
+ String[] key = entry.getKey().split(String.valueOf(TraceConstants.CONTENT_SPLITOR));
+ flushData(entry.getValue(), key[0], key[1]);
}
}
- private void clear() {
- this.firstBeanAddTime = 0;
- this.currentMsgSize = 0;
- this.currentMsgKeySize = 0;
- this.traceTransferBeanList.clear();
- }
- }
-
- class AsyncDataSendTask implements Runnable {
- private final String traceTopicName;
- private final String regionId;
- private final List traceTransferBeanList;
-
- public AsyncDataSendTask(String traceTopicName, String regionId, List traceTransferBeanList) {
- this.traceTopicName = traceTopicName;
- this.regionId = regionId;
- this.traceTransferBeanList = traceTransferBeanList;
- }
-
- @Override
- public void run() {
+ private void flushData(List transBeanList, String topic, String traceTopic) {
+ if (transBeanList.size() == 0) {
+ return;
+ }
StringBuilder buffer = new StringBuilder(1024);
- Set keySet = new HashSet<>();
- for (TraceTransferBean bean : traceTransferBeanList) {
+ int count = 0;
+ Set keySet = new HashSet();
+ for (TraceTransferBean bean : transBeanList) {
keySet.addAll(bean.getTransKey());
buffer.append(bean.getTransData());
+ count++;
+ if (buffer.length() >= traceProducer.getMaxMessageSize()) {
+ sendTraceDataByMQ(keySet, buffer.toString(), traceTopic);
+ buffer.delete(0, buffer.length());
+ keySet.clear();
+ count = 0;
+ }
+ }
+ if (count > 0) {
+ sendTraceDataByMQ(keySet, buffer.toString(), traceTopic);
}
- sendTraceDataByMQ(keySet, buffer.toString(), traceTopicName);
+ transBeanList.clear();
}
/**
* Send message trace data
*
- * @param keySet the keyset in this batch(including msgId in original message not offsetMsgId)
- * @param data the message trace data in this batch
+ * @param keySet the keyset in this batch(including msgId in original message not offsetMsgId)
+ * @param data the message trace data in this batch
* @param traceTopic the topic which message trace data will send to
*/
private void sendTraceDataByMQ(Set keySet, final String data, String traceTopic) {
@@ -467,4 +421,4 @@ private Set tryGetMessageQueueBrokerSet(DefaultMQProducerImpl producer,
}
}
-}
+}
\ No newline at end of file
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java
index 70c147e1eb3..17db1fbfa15 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceBean.java
@@ -21,7 +21,7 @@
import org.apache.rocketmq.common.message.MessageType;
public class TraceBean {
- private static final String LOCAL_ADDRESS = UtilAll.ipToIPv4Str(UtilAll.getIP());
+ private static final String LOCAL_ADDRESS;
private String topic = "";
private String msgId = "";
private String offsetMsgId = "";
@@ -37,6 +37,15 @@ public class TraceBean {
private String transactionId;
private boolean fromTransactionCheck;
+ static {
+ byte[] ip = UtilAll.getIP();
+ if (ip.length == 4) {
+ LOCAL_ADDRESS = UtilAll.ipToIPv4Str(ip);
+ } else {
+ LOCAL_ADDRESS = UtilAll.ipToIPv6Str(ip);
+ }
+ }
+
public MessageType getMsgType() {
return msgType;
}
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java
index 1ad4b610515..67f7ab3f8fb 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceConstants.java
@@ -32,7 +32,7 @@ public class TraceConstants {
public static final String ROCKETMQ_SUCCESS = "rocketmq.success";
public static final String ROCKETMQ_TAGS = "rocketmq.tags";
public static final String ROCKETMQ_KEYS = "rocketmq.keys";
- public static final String ROCKETMQ_SOTRE_HOST = "rocketmq.store_host";
+ public static final String ROCKETMQ_STORE_HOST = "rocketmq.store_host";
public static final String ROCKETMQ_BODY_LENGTH = "rocketmq.body_length";
public static final String ROCKETMQ_MSG_ID = "rocketmq.mgs_id";
public static final String ROCKETMQ_MSG_TYPE = "rocketmq.mgs_type";
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java
index 0fdd95243a5..57e9b6410db 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/TraceDataEncoder.java
@@ -193,9 +193,10 @@ public static TraceTransferBean encoderFromContextBean(TraceContext ctx) {
.append(bean.getKeys()).append(TraceConstants.CONTENT_SPLITOR)//
.append(ctx.getContextCode()).append(TraceConstants.CONTENT_SPLITOR);
if (!ctx.getAccessChannel().equals(AccessChannel.CLOUD)) {
- sb.append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR)
- .append(ctx.getGroupName()).append(TraceConstants.FIELD_SPLITOR);
+ sb.append(ctx.getTimeStamp()).append(TraceConstants.CONTENT_SPLITOR);
+ sb.append(ctx.getGroupName());
}
+ sb.append(TraceConstants.FIELD_SPLITOR);
}
}
break;
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionOpenTracingHookImpl.java b/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionOpenTracingHookImpl.java
index 62d310f1961..44e4b69776e 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionOpenTracingHookImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/hook/EndTransactionOpenTracingHookImpl.java
@@ -60,7 +60,7 @@ public void endTransaction(EndTransactionContext context) {
span.setTag(Tags.MESSAGE_BUS_DESTINATION, msg.getTopic());
span.setTag(TraceConstants.ROCKETMQ_TAGS, msg.getTags());
span.setTag(TraceConstants.ROCKETMQ_KEYS, msg.getKeys());
- span.setTag(TraceConstants.ROCKETMQ_SOTRE_HOST, context.getBrokerAddr());
+ span.setTag(TraceConstants.ROCKETMQ_STORE_HOST, context.getBrokerAddr());
span.setTag(TraceConstants.ROCKETMQ_MSG_ID, context.getMsgId());
span.setTag(TraceConstants.ROCKETMQ_MSG_TYPE, MessageType.Trans_msg_Commit.name());
span.setTag(TraceConstants.ROCKETMQ_TRANSACTION_ID, context.getTransactionId());
diff --git a/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageOpenTracingHookImpl.java b/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageOpenTracingHookImpl.java
index 60c18a22a77..3cb64933849 100644
--- a/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageOpenTracingHookImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/trace/hook/SendMessageOpenTracingHookImpl.java
@@ -60,7 +60,7 @@ public void sendMessageBefore(SendMessageContext context) {
span.setTag(Tags.MESSAGE_BUS_DESTINATION, msg.getTopic());
span.setTag(TraceConstants.ROCKETMQ_TAGS, msg.getTags());
span.setTag(TraceConstants.ROCKETMQ_KEYS, msg.getKeys());
- span.setTag(TraceConstants.ROCKETMQ_SOTRE_HOST, context.getBrokerAddr());
+ span.setTag(TraceConstants.ROCKETMQ_STORE_HOST, context.getBrokerAddr());
span.setTag(TraceConstants.ROCKETMQ_MSG_TYPE, context.getMsgType().name());
span.setTag(TraceConstants.ROCKETMQ_BODY_LENGTH, msg.getBody().length);
context.setMqTraceContext(span);
diff --git a/client/src/test/java/org/apache/rocketmq/client/ClientConfigTest.java b/client/src/test/java/org/apache/rocketmq/client/ClientConfigTest.java
new file mode 100644
index 00000000000..5afe9cc011d
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/ClientConfigTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.client;
+
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.protocol.LanguageCode;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ClientConfigTest {
+
+ private ClientConfig clientConfig;
+
+ private final String resource = "resource";
+
+ @Before
+ public void init() {
+ clientConfig = createClientConfig();
+ }
+
+ @Test
+ public void testWithNamespace() {
+ Set resources = clientConfig.withNamespace(Collections.singleton(resource));
+ assertTrue(resources.contains("lmq%resource"));
+ }
+
+ @Test
+ public void testWithoutNamespace() {
+ String actual = clientConfig.withoutNamespace(resource);
+ assertEquals(resource, actual);
+ Set resources = clientConfig.withoutNamespace(Collections.singleton(resource));
+ assertTrue(resources.contains(resource));
+ }
+
+ @Test
+ public void testQueuesWithNamespace() {
+ MessageQueue messageQueue = new MessageQueue();
+ messageQueue.setTopic("defaultTopic");
+ Collection messageQueues = clientConfig.queuesWithNamespace(Collections.singleton(messageQueue));
+ assertTrue(messageQueues.contains(messageQueue));
+ assertEquals("lmq%defaultTopic", messageQueues.iterator().next().getTopic());
+ }
+
+ private ClientConfig createClientConfig() {
+ ClientConfig result = new ClientConfig();
+ result.setUnitName("unitName");
+ result.setClientIP("127.0.0.1");
+ result.setClientCallbackExecutorThreads(1);
+ result.setPollNameServerInterval(1000 * 30);
+ result.setHeartbeatBrokerInterval(1000 * 30);
+ result.setPersistConsumerOffsetInterval(1000 * 5);
+ result.setPullTimeDelayMillsWhenException(1000);
+ result.setUnitMode(true);
+ result.setSocksProxyConfig("{}");
+ result.setLanguage(LanguageCode.JAVA);
+ result.setDecodeReadBody(true);
+ result.setDecodeDecompressBody(true);
+ result.setAccessChannel(AccessChannel.LOCAL);
+ result.setMqClientApiTimeout(1000 * 3);
+ result.setEnableStreamRequestType(true);
+ result.setSendLatencyEnable(true);
+ result.setEnableHeartbeatChannelEventListener(true);
+ result.setDetectTimeout(200);
+ result.setDetectInterval(1000 * 2);
+ result.setUseHeartbeatV2(false);
+ result.buildMQClientId();
+ result.setNamespace("lmq");
+ return result;
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
index 65237bc8f76..592c247057b 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultLitePullConsumerTest.java
@@ -63,8 +63,6 @@
import org.mockito.Spy;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
-import org.mockito.quality.Strictness;
-import org.mockito.junit.jupiter.MockitoSettings;
import org.mockito.stubbing.Answer;
import static org.assertj.core.api.Assertions.assertThat;
@@ -81,8 +79,7 @@
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
-@RunWith(MockitoJUnitRunner.class)
-@MockitoSettings(strictness = Strictness.LENIENT)
+@RunWith(MockitoJUnitRunner.Silent.class)
public class DefaultLitePullConsumerTest {
@Spy
private MQClientInstance mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(new ClientConfig());
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
index a10fd74b34f..834be5cf16f 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/DefaultMQPushConsumerTest.java
@@ -209,7 +209,9 @@ public PullResult answer(InvocationOnMock mock) throws Throwable {
@AfterClass
public static void terminate() {
- pushConsumer.shutdown();
+ if (pushConsumer != null) {
+ pushConsumer.shutdown();
+ }
}
@Test
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/store/ControllableOffsetTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/store/ControllableOffsetTest.java
new file mode 100644
index 00000000000..23a56ca51ca
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/store/ControllableOffsetTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.client.consumer.store;
+
+
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
+public class ControllableOffsetTest {
+
+ private ControllableOffset controllableOffset;
+
+ @Before
+ public void setUp() {
+ controllableOffset = new ControllableOffset(0);
+ }
+
+ @Test
+ public void testUpdateAndFreeze_ShouldFreezeOffsetAtTargetValue() {
+ controllableOffset.updateAndFreeze(100);
+ assertEquals(100, controllableOffset.getOffset());
+ controllableOffset.update(200);
+ assertEquals(100, controllableOffset.getOffset());
+ }
+
+ @Test
+ public void testUpdate_ShouldUpdateOffsetWhenNotFrozen() {
+ controllableOffset.update(200);
+ assertEquals(200, controllableOffset.getOffset());
+ }
+
+ @Test
+ public void testUpdate_ShouldNotUpdateOffsetWhenFrozen() {
+ controllableOffset.updateAndFreeze(100);
+ controllableOffset.update(200);
+ assertEquals(100, controllableOffset.getOffset());
+ }
+
+ @Test
+ public void testUpdate_ShouldNotDecreaseOffsetWhenIncreaseOnly() {
+ controllableOffset.update(200);
+ controllableOffset.update(100, true);
+ assertEquals(200, controllableOffset.getOffset());
+ }
+
+ @Test
+ public void testUpdate_ShouldUpdateOffsetToGreaterValueWhenIncreaseOnly() {
+ controllableOffset.update(100);
+ controllableOffset.update(200, true);
+ assertEquals(200, controllableOffset.getOffset());
+ }
+
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/ClientRemotingProcessorTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/ClientRemotingProcessorTest.java
new file mode 100644
index 00000000000..ed31aa10430
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/ClientRemotingProcessorTest.java
@@ -0,0 +1,213 @@
+/*
+ * 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.client.impl;
+
+import io.netty.channel.ChannelHandlerContext;
+import org.apache.rocketmq.client.ClientConfig;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.client.impl.producer.MQProducerInner;
+import org.apache.rocketmq.common.message.MessageConst;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
+import org.apache.rocketmq.remoting.protocol.header.CheckTransactionStateRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ConsumeMessageDirectlyResultRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerRunningInfoRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.GetConsumerStatusRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.NotifyConsumerIdsChangedRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ReplyMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.ResetOffsetRequestHeader;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+
+import static org.apache.rocketmq.common.message.MessageDecoder.NAME_VALUE_SEPARATOR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ClientRemotingProcessorTest {
+
+ @Mock
+ private MQClientInstance mQClientFactory;
+
+ private ClientRemotingProcessor processor;
+
+ private final String defaultTopic = "defaultTopic";
+
+ private final String defaultBroker = "defaultBroker";
+
+ private final String defaultGroup = "defaultGroup";
+
+ @Before
+ public void init() throws RemotingException, InterruptedException, MQClientException {
+ processor = new ClientRemotingProcessor(mQClientFactory);
+ ClientConfig clientConfig = mock(ClientConfig.class);
+ when(clientConfig.getNamespace()).thenReturn("namespace");
+ when(mQClientFactory.getClientConfig()).thenReturn(clientConfig);
+ MQProducerInner producerInner = mock(MQProducerInner.class);
+ when(mQClientFactory.selectProducer(defaultGroup)).thenReturn(producerInner);
+ }
+
+ @Test
+ public void testCheckTransactionState() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.CHECK_TRANSACTION_STATE);
+ when(request.getBody()).thenReturn(getMessageResult());
+ CheckTransactionStateRequestHeader requestHeader = new CheckTransactionStateRequestHeader();
+ when(request.decodeCommandCustomHeader(CheckTransactionStateRequestHeader.class)).thenReturn(requestHeader);
+ assertNull(processor.processRequest(ctx, request));
+ }
+
+ @Test
+ public void testNotifyConsumerIdsChanged() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.NOTIFY_CONSUMER_IDS_CHANGED);
+ NotifyConsumerIdsChangedRequestHeader requestHeader = new NotifyConsumerIdsChangedRequestHeader();
+ when(request.decodeCommandCustomHeader(NotifyConsumerIdsChangedRequestHeader.class)).thenReturn(requestHeader);
+ assertNull(processor.processRequest(ctx, request));
+ }
+
+ @Test
+ public void testResetOffset() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.RESET_CONSUMER_CLIENT_OFFSET);
+ ResetOffsetBody offsetBody = new ResetOffsetBody();
+ when(request.getBody()).thenReturn(RemotingSerializable.encode(offsetBody));
+ ResetOffsetRequestHeader requestHeader = new ResetOffsetRequestHeader();
+ when(request.decodeCommandCustomHeader(ResetOffsetRequestHeader.class)).thenReturn(requestHeader);
+ assertNull(processor.processRequest(ctx, request));
+ }
+
+ @Test
+ public void testGetConsumeStatus() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.GET_CONSUMER_STATUS_FROM_CLIENT);
+ GetConsumerStatusRequestHeader requestHeader = new GetConsumerStatusRequestHeader();
+ when(request.decodeCommandCustomHeader(GetConsumerStatusRequestHeader.class)).thenReturn(requestHeader);
+ assertNotNull(processor.processRequest(ctx, request));
+ }
+
+ @Test
+ public void testGetConsumerRunningInfo() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.GET_CONSUMER_RUNNING_INFO);
+ ConsumerRunningInfo consumerRunningInfo = new ConsumerRunningInfo();
+ consumerRunningInfo.setJstack("jstack");
+ when(mQClientFactory.consumerRunningInfo(anyString())).thenReturn(consumerRunningInfo);
+ GetConsumerRunningInfoRequestHeader requestHeader = new GetConsumerRunningInfoRequestHeader();
+ requestHeader.setJstackEnable(true);
+ requestHeader.setConsumerGroup(defaultGroup);
+ when(request.decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class)).thenReturn(requestHeader);
+ RemotingCommand command = processor.processRequest(ctx, request);
+ assertNotNull(command);
+ assertEquals(ResponseCode.SUCCESS, command.getCode());
+ }
+
+ @Test
+ public void testConsumeMessageDirectly() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.CONSUME_MESSAGE_DIRECTLY);
+ when(request.getBody()).thenReturn(getMessageResult());
+ ConsumeMessageDirectlyResult directlyResult = mock(ConsumeMessageDirectlyResult.class);
+ when(mQClientFactory.consumeMessageDirectly(any(MessageExt.class), anyString(), anyString())).thenReturn(directlyResult);
+ ConsumeMessageDirectlyResultRequestHeader requestHeader = new ConsumeMessageDirectlyResultRequestHeader();
+ requestHeader.setConsumerGroup(defaultGroup);
+ requestHeader.setBrokerName(defaultBroker);
+ when(request.decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class)).thenReturn(requestHeader);
+ RemotingCommand command = processor.processRequest(ctx, request);
+ assertNotNull(command);
+ assertEquals(ResponseCode.SUCCESS, command.getCode());
+ }
+
+ @Test
+ public void testReceiveReplyMessage() throws Exception {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ RemotingCommand request = mock(RemotingCommand.class);
+ when(request.getCode()).thenReturn(RequestCode.PUSH_REPLY_MESSAGE_TO_CLIENT);
+ when(request.getBody()).thenReturn(getMessageResult());
+ when(request.decodeCommandCustomHeader(ReplyMessageRequestHeader.class)).thenReturn(createReplyMessageRequestHeader());
+ when(request.getBody()).thenReturn(new byte[1]);
+ RemotingCommand command = processor.processRequest(ctx, request);
+ assertNotNull(command);
+ assertEquals(ResponseCode.SUCCESS, command.getCode());
+ }
+
+ private ReplyMessageRequestHeader createReplyMessageRequestHeader() {
+ ReplyMessageRequestHeader result = new ReplyMessageRequestHeader();
+ result.setTopic(defaultTopic);
+ result.setQueueId(0);
+ result.setStoreTimestamp(System.currentTimeMillis());
+ result.setBornTimestamp(System.currentTimeMillis());
+ result.setReconsumeTimes(1);
+ result.setBornHost("127.0.0.1:12911");
+ result.setStoreHost("127.0.0.1:10911");
+ result.setSysFlag(1);
+ result.setFlag(1);
+ result.setProperties("CORRELATION_ID" + NAME_VALUE_SEPARATOR + "1");
+ return result;
+ }
+
+ private byte[] getMessageResult() throws Exception {
+ byte[] bytes = MessageDecoder.encode(createMessageExt(), false);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
+ byteBuffer.put(bytes);
+ return byteBuffer.array();
+ }
+
+ private MessageExt createMessageExt() {
+ MessageExt result = new MessageExt();
+ result.setBody("body".getBytes(StandardCharsets.UTF_8));
+ result.setTopic(defaultTopic);
+ result.setBrokerName(defaultBroker);
+ result.putUserProperty("key", "value");
+ result.getProperties().put(MessageConst.PROPERTY_PRODUCER_GROUP, defaultGroup);
+ result.getProperties().put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, "TX1");
+ result.setKeys("keys");
+ SocketAddress bornHost = new InetSocketAddress("127.0.0.1", 12911);
+ SocketAddress storeHost = new InetSocketAddress("127.0.0.1", 10911);
+ result.setBornHost(bornHost);
+ result.setStoreHost(storeHost);
+ return result;
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/MQAdminImplTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/MQAdminImplTest.java
new file mode 100644
index 00000000000..f52aba2dc00
--- /dev/null
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/MQAdminImplTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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.client.impl;
+
+import org.apache.rocketmq.client.ClientConfig;
+import org.apache.rocketmq.client.QueryResult;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
+import org.apache.rocketmq.client.impl.factory.MQClientInstance;
+import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.message.MessageDecoder;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.common.sysflag.MessageSysFlag;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.header.QueryMessageResponseHeader;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.class)
+public class MQAdminImplTest {
+
+ @Mock
+ private MQClientInstance mQClientFactory;
+
+ @Mock
+ private MQClientAPIImpl mQClientAPIImpl;
+
+ private MQAdminImpl mqAdminImpl;
+
+ private final String defaultTopic = "defaultTopic";
+
+ private final String defaultBroker = "defaultBroker";
+
+ private final String defaultCluster = "defaultCluster";
+
+ private final String defaultBrokerAddr = "127.0.0.1:10911";
+
+ private final long defaultTimeout = 3000L;
+
+ @Before
+ public void init() throws RemotingException, InterruptedException, MQClientException {
+ when(mQClientFactory.getMQClientAPIImpl()).thenReturn(mQClientAPIImpl);
+ when(mQClientAPIImpl.getTopicRouteInfoFromNameServer(anyString(), anyLong())).thenReturn(createRouteData());
+ ClientConfig clientConfig = mock(ClientConfig.class);
+ when(clientConfig.getNamespace()).thenReturn("namespace");
+ when(mQClientFactory.getClientConfig()).thenReturn(clientConfig);
+ when(mQClientFactory.findBrokerAddressInPublish(any())).thenReturn(defaultBrokerAddr);
+ when(mQClientFactory.getAnExistTopicRouteData(any())).thenReturn(createRouteData());
+ mqAdminImpl = new MQAdminImpl(mQClientFactory);
+ }
+
+ @Test
+ public void assertTimeoutMillis() {
+ assertEquals(6000L, mqAdminImpl.getTimeoutMillis());
+ mqAdminImpl.setTimeoutMillis(defaultTimeout);
+ assertEquals(defaultTimeout, mqAdminImpl.getTimeoutMillis());
+ }
+
+ @Test
+ public void testCreateTopic() throws MQClientException {
+ mqAdminImpl.createTopic("", defaultTopic, 6);
+ }
+
+ @Test
+ public void assertFetchPublishMessageQueues() throws MQClientException {
+ List queueList = mqAdminImpl.fetchPublishMessageQueues(defaultTopic);
+ assertNotNull(queueList);
+ assertEquals(6, queueList.size());
+ for (MessageQueue each : queueList) {
+ assertEquals(defaultTopic, each.getTopic());
+ assertEquals(defaultBroker, each.getBrokerName());
+ }
+ }
+
+ @Test
+ public void assertFetchSubscribeMessageQueues() throws MQClientException {
+ Set queueList = mqAdminImpl.fetchSubscribeMessageQueues(defaultTopic);
+ assertNotNull(queueList);
+ assertEquals(6, queueList.size());
+ for (MessageQueue each : queueList) {
+ assertEquals(defaultTopic, each.getTopic());
+ assertEquals(defaultBroker, each.getBrokerName());
+ }
+ }
+
+ @Test
+ public void assertSearchOffset() throws MQClientException {
+ assertEquals(0, mqAdminImpl.searchOffset(new MessageQueue(), defaultTimeout));
+ }
+
+ @Test
+ public void assertMaxOffset() throws MQClientException {
+ assertEquals(0, mqAdminImpl.maxOffset(new MessageQueue()));
+ }
+
+ @Test
+ public void assertMinOffset() throws MQClientException {
+ assertEquals(0, mqAdminImpl.minOffset(new MessageQueue()));
+ }
+
+ @Test
+ public void assertEarliestMsgStoreTime() throws MQClientException {
+ assertEquals(0, mqAdminImpl.earliestMsgStoreTime(new MessageQueue()));
+ }
+
+ @Test(expected = MQClientException.class)
+ public void assertViewMessage() throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
+ MessageExt actual = mqAdminImpl.viewMessage(defaultTopic, "1");
+ assertNotNull(actual);
+ }
+
+ @Test
+ public void assertQueryMessage() throws InterruptedException, MQClientException, MQBrokerException, RemotingException {
+ doAnswer(invocation -> {
+ InvokeCallback callback = invocation.getArgument(3);
+ QueryMessageResponseHeader responseHeader = new QueryMessageResponseHeader();
+ responseHeader.setIndexLastUpdatePhyoffset(1L);
+ responseHeader.setIndexLastUpdateTimestamp(System.currentTimeMillis());
+ RemotingCommand response = mock(RemotingCommand.class);
+ when(response.decodeCommandCustomHeader(QueryMessageResponseHeader.class)).thenReturn(responseHeader);
+ when(response.getBody()).thenReturn(getMessageResult());
+ when(response.getCode()).thenReturn(ResponseCode.SUCCESS);
+ callback.operationSucceed(response);
+ return null;
+ }).when(mQClientAPIImpl).queryMessage(anyString(), any(), anyLong(), any(InvokeCallback.class), any());
+ QueryResult actual = mqAdminImpl.queryMessage(defaultTopic, "keys", 100, 1L, 50L);
+ assertNotNull(actual);
+ assertEquals(1, actual.getMessageList().size());
+ assertEquals(defaultTopic, actual.getMessageList().get(0).getTopic());
+ }
+
+ @Test
+ public void assertQueryMessageByUniqKey() throws InterruptedException, MQClientException, MQBrokerException, RemotingException {
+ doAnswer(invocation -> {
+ InvokeCallback callback = invocation.getArgument(3);
+ QueryMessageResponseHeader responseHeader = new QueryMessageResponseHeader();
+ responseHeader.setIndexLastUpdatePhyoffset(1L);
+ responseHeader.setIndexLastUpdateTimestamp(System.currentTimeMillis());
+ RemotingCommand response = mock(RemotingCommand.class);
+ when(response.decodeCommandCustomHeader(QueryMessageResponseHeader.class)).thenReturn(responseHeader);
+ when(response.getBody()).thenReturn(getMessageResult());
+ when(response.getCode()).thenReturn(ResponseCode.SUCCESS);
+ callback.operationSucceed(response);
+ return null;
+ }).when(mQClientAPIImpl).queryMessage(anyString(), any(), anyLong(), any(InvokeCallback.class), any());
+ String msgId = buildMsgId();
+ MessageExt actual = mqAdminImpl.queryMessageByUniqKey(defaultTopic, msgId);
+ assertNotNull(actual);
+ assertEquals(msgId, actual.getMsgId());
+ assertEquals(defaultTopic, actual.getTopic());
+ actual = mqAdminImpl.queryMessageByUniqKey(defaultCluster, defaultTopic, msgId);
+ assertNotNull(actual);
+ assertEquals(msgId, actual.getMsgId());
+ assertEquals(defaultTopic, actual.getTopic());
+ QueryResult queryResult = mqAdminImpl.queryMessageByUniqKey(defaultTopic, msgId, 1, 0L, 1L);
+ assertNotNull(queryResult);
+ assertEquals(1, queryResult.getMessageList().size());
+ assertEquals(defaultTopic, queryResult.getMessageList().get(0).getTopic());
+ }
+
+ private String buildMsgId() {
+ MessageExt msgExt = createMessageExt();
+ int storeHostIPLength = (msgExt.getFlag() & MessageSysFlag.STOREHOSTADDRESS_V6_FLAG) == 0 ? 4 : 16;
+ int msgIDLength = storeHostIPLength + 4 + 8;
+ ByteBuffer byteBufferMsgId = ByteBuffer.allocate(msgIDLength);
+ return MessageDecoder.createMessageId(byteBufferMsgId, msgExt.getStoreHostBytes(), msgExt.getCommitLogOffset());
+ }
+
+ private TopicRouteData createRouteData() {
+ TopicRouteData result = new TopicRouteData();
+ result.setBrokerDatas(createBrokerData());
+ result.setQueueDatas(createQueueData());
+ return result;
+ }
+
+ private List createBrokerData() {
+ HashMap brokerAddrs = new HashMap<>();
+ brokerAddrs.put(MixAll.MASTER_ID, defaultBrokerAddr);
+ return Collections.singletonList(new BrokerData(defaultCluster, defaultBroker, brokerAddrs));
+ }
+
+ private List createQueueData() {
+ QueueData queueData = new QueueData();
+ queueData.setPerm(6);
+ queueData.setBrokerName(defaultBroker);
+ queueData.setReadQueueNums(6);
+ queueData.setWriteQueueNums(6);
+ return Collections.singletonList(queueData);
+ }
+
+ private byte[] getMessageResult() throws Exception {
+ byte[] bytes = MessageDecoder.encode(createMessageExt(), false);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(bytes.length);
+ byteBuffer.put(bytes);
+ return byteBuffer.array();
+ }
+
+ private MessageExt createMessageExt() {
+ MessageExt result = new MessageExt();
+ result.setBody("body".getBytes(StandardCharsets.UTF_8));
+ result.setTopic(defaultTopic);
+ result.setBrokerName(defaultBroker);
+ result.putUserProperty("key", "value");
+ result.setKeys("keys");
+ SocketAddress bornHost = new InetSocketAddress("127.0.0.1", 12911);
+ SocketAddress storeHost = new InetSocketAddress("127.0.0.1", 10911);
+ result.setBornHost(bornHost);
+ result.setStoreHost(storeHost);
+ return result;
+ }
+}
diff --git a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
index b0876c7c0d9..e311e0c9b85 100644
--- a/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/impl/MQClientAPIImplTest.java
@@ -16,13 +16,6 @@
*/
package org.apache.rocketmq.client.impl;
-import java.lang.reflect.Field;
-import java.net.InetSocketAddress;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.CountDownLatch;
import org.apache.rocketmq.client.ClientConfig;
import org.apache.rocketmq.client.consumer.AckCallback;
import org.apache.rocketmq.client.consumer.AckResult;
@@ -30,6 +23,9 @@
import org.apache.rocketmq.client.consumer.PopCallback;
import org.apache.rocketmq.client.consumer.PopResult;
import org.apache.rocketmq.client.consumer.PopStatus;
+import org.apache.rocketmq.client.consumer.PullCallback;
+import org.apache.rocketmq.client.consumer.PullResult;
+import org.apache.rocketmq.client.consumer.PullStatus;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.hook.SendMessageContext;
@@ -39,8 +35,10 @@
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.MixAll;
+import org.apache.rocketmq.common.Pair;
import org.apache.rocketmq.common.PlainAccessConfig;
import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
@@ -49,20 +47,65 @@
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.message.MessageQueueAssignment;
import org.apache.rocketmq.common.message.MessageRequestMode;
+import org.apache.rocketmq.common.namesrv.TopAddressing;
+import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.InvokeCallback;
import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.common.HeartbeatV2Result;
+import org.apache.rocketmq.remoting.exception.RemotingCommandException;
+import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingException;
+import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.netty.NettyClientConfig;
import org.apache.rocketmq.remoting.netty.ResponseFuture;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RemotingSerializable;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
+import org.apache.rocketmq.remoting.protocol.admin.TopicOffset;
+import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
+import org.apache.rocketmq.remoting.protocol.body.AclInfo;
+import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
+import org.apache.rocketmq.remoting.protocol.body.BrokerReplicasInfo;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsData;
+import org.apache.rocketmq.remoting.protocol.body.BrokerStatsItem;
+import org.apache.rocketmq.remoting.protocol.body.ClusterAclVersionInfo;
+import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
+import org.apache.rocketmq.remoting.protocol.body.Connection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeMessageDirectlyResult;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeQueueData;
+import org.apache.rocketmq.remoting.protocol.body.ConsumeStatsList;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ConsumerRunningInfo;
+import org.apache.rocketmq.remoting.protocol.body.EpochEntryCache;
+import org.apache.rocketmq.remoting.protocol.body.GetConsumerStatusBody;
+import org.apache.rocketmq.remoting.protocol.body.GroupList;
+import org.apache.rocketmq.remoting.protocol.body.HARuntimeInfo;
+import org.apache.rocketmq.remoting.protocol.body.KVTable;
+import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.ProducerConnection;
+import org.apache.rocketmq.remoting.protocol.body.ProducerInfo;
+import org.apache.rocketmq.remoting.protocol.body.ProducerTableInfo;
import org.apache.rocketmq.remoting.protocol.body.QueryAssignmentResponseBody;
+import org.apache.rocketmq.remoting.protocol.body.QueryConsumeQueueResponseBody;
+import org.apache.rocketmq.remoting.protocol.body.QueryConsumeTimeSpanBody;
+import org.apache.rocketmq.remoting.protocol.body.QueryCorrectionOffsetBody;
+import org.apache.rocketmq.remoting.protocol.body.QuerySubscriptionResponseBody;
+import org.apache.rocketmq.remoting.protocol.body.QueueTimeSpan;
+import org.apache.rocketmq.remoting.protocol.body.ResetOffsetBody;
+import org.apache.rocketmq.remoting.protocol.body.SubscriptionGroupWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
+import org.apache.rocketmq.remoting.protocol.body.TopicList;
+import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.UserInfo;
import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ChangeInvisibleTimeResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.EndTransactionRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.ExtraInfoUtil;
+import org.apache.rocketmq.remoting.protocol.header.GetBrokerAclConfigResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseBody;
import org.apache.rocketmq.remoting.protocol.header.GetConsumerListByGroupResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.GetEarliestMsgStoretimeResponseHeader;
@@ -70,15 +113,33 @@
import org.apache.rocketmq.remoting.protocol.header.GetMinOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.PopMessageResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.PullMessageRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.PullMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.QueryConsumerOffsetResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.QueryMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SearchOffsetResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.UpdateConsumerOffsetResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.UpdateGroupForbiddenRequestHeader;
+import org.apache.rocketmq.remoting.protocol.header.controller.ElectMasterResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.controller.GetMetaDataResponseHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.AddWritePermOfBrokerResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.namesrv.GetKVConfigResponseHeader;
+import org.apache.rocketmq.remoting.protocol.header.namesrv.WipeWritePermOfBrokerResponseHeader;
+import org.apache.rocketmq.remoting.protocol.heartbeat.ConsumeType;
+import org.apache.rocketmq.remoting.protocol.heartbeat.HeartbeatData;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
+import org.apache.rocketmq.remoting.protocol.heartbeat.SubscriptionData;
+import org.apache.rocketmq.remoting.protocol.route.BrokerData;
+import org.apache.rocketmq.remoting.protocol.route.QueueData;
+import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicConfigAndQueueMapping;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingDetail;
+import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;
+import org.apache.rocketmq.remoting.protocol.subscription.GroupForbidden;
import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
import org.assertj.core.api.Assertions;
import org.junit.Before;
@@ -86,34 +147,74 @@
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
-import org.mockito.invocation.InvocationOnMock;
import org.mockito.junit.MockitoJUnitRunner;
import org.mockito.stubbing.Answer;
+import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Fail.failBecauseExceptionWasNotThrown;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class MQClientAPIImplTest {
+
private MQClientAPIImpl mqClientAPI = new MQClientAPIImpl(new NettyClientConfig(), null, null, new ClientConfig());
+
@Mock
private RemotingClient remotingClient;
+
@Mock
private DefaultMQProducerImpl defaultMQProducerImpl;
- private String brokerAddr = "127.0.0.1";
- private String brokerName = "DefaultBroker";
- private String clusterName = "DefaultCluster";
- private static String group = "FooBarGroup";
- private static String topic = "FooBar";
- private Message msg = new Message("FooBar", new byte[] {});
- private static String clientId = "127.0.0.2@UnitTest";
+ @Mock
+ private RemotingCommand response;
+
+ private final String brokerAddr = "127.0.0.1";
+
+ private final String brokerName = "DefaultBroker";
+
+ private final String clusterName = "DefaultCluster";
+
+ private final String group = "FooBarGroup";
+
+ private final String topic = "FooBar";
+
+ private final Message msg = new Message("FooBar", new byte[]{});
+
+ private final String clientId = "127.0.0.2@UnitTest";
+
+ private final String defaultTopic = "defaultTopic";
+
+ private final String defaultBrokerAddr = "127.0.0.1:10911";
+
+ private final String defaultNsAddr = "127.0.0.1:9876";
+
+ private final long defaultTimeout = 3000L;
@Before
public void init() throws Exception {
@@ -153,12 +254,9 @@ public void testSendMessageOneWay_WithException() throws RemotingException, Inte
@Test
public void testSendMessageSync_Success() throws InterruptedException, RemotingException, MQBrokerException {
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createSendMessageSuccessResponse(request);
- }
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createSendMessageSuccessResponse(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
SendMessageRequestHeader requestHeader = createSendMessageRequestHeader();
@@ -173,17 +271,14 @@ public Object answer(InvocationOnMock mock) throws Throwable {
}
@Test
- public void testSendMessageSync_WithException() throws InterruptedException, RemotingException, MQBrokerException {
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setOpaque(request.getOpaque());
- response.setRemark("Broker is broken.");
- return response;
- }
+ public void testSendMessageSync_WithException() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setOpaque(request.getOpaque());
+ response.setRemark("Broker is broken.");
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
SendMessageRequestHeader requestHeader = createSendMessageRequestHeader();
@@ -204,16 +299,13 @@ public void testSendMessageAsync_Success() throws RemotingException, Interrupted
3 * 1000, CommunicationMode.ASYNC, new SendMessageContext(), defaultMQProducerImpl);
assertThat(sendResult).isNull();
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- responseFuture.setResponseCommand(createSendMessageSuccessResponse(request));
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer(mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ responseFuture.setResponseCommand(createSendMessageSuccessResponse(request));
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
SendMessageContext sendMessageContext = new SendMessageContext();
sendMessageContext.setProducer(new DefaultMQProducerImpl(new DefaultMQProducer()));
@@ -266,34 +358,26 @@ public void onException(Throwable e) {
}
@Test
- public void testCreatePlainAccessConfig_Success() throws InterruptedException, RemotingException, MQBrokerException {
-
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createSuccessResponse4UpdateAclConfig(request);
- }
+ public void testCreatePlainAccessConfig_Success() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createSuccessResponse4UpdateAclConfig(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
PlainAccessConfig config = createUpdateAclConfig();
try {
mqClientAPI.createPlainAccessConfig(brokerAddr, config, 3 * 1000);
- } catch (MQClientException ex) {
+ } catch (MQClientException ignored) {
}
}
@Test
- public void testCreatePlainAccessConfig_Exception() throws InterruptedException, RemotingException, MQBrokerException {
-
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createErrorResponse4UpdateAclConfig(request);
- }
+ public void testCreatePlainAccessConfig_Exception() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createErrorResponse4UpdateAclConfig(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
PlainAccessConfig config = createUpdateAclConfig();
@@ -306,33 +390,25 @@ public Object answer(InvocationOnMock mock) throws Throwable {
}
@Test
- public void testDeleteAccessConfig_Success() throws InterruptedException, RemotingException, MQBrokerException {
-
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createSuccessResponse4DeleteAclConfig(request);
- }
+ public void testDeleteAccessConfig_Success() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createSuccessResponse4DeleteAclConfig(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
String accessKey = "1234567";
try {
mqClientAPI.deleteAccessConfig(brokerAddr, accessKey, 3 * 1000);
- } catch (MQClientException ex) {
+ } catch (MQClientException ignored) {
}
}
@Test
- public void testDeleteAccessConfig_Exception() throws InterruptedException, RemotingException, MQBrokerException {
-
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createErrorResponse4DeleteAclConfig(request);
- }
+ public void testDeleteAccessConfig_Exception() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createErrorResponse4DeleteAclConfig(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
try {
@@ -344,17 +420,14 @@ public Object answer(InvocationOnMock mock) throws Throwable {
}
@Test
- public void testResumeCheckHalfMessage_WithException() throws RemotingException, InterruptedException, MQBrokerException, MQClientException {
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
- response.setCode(ResponseCode.SYSTEM_ERROR);
- response.setOpaque(request.getOpaque());
- response.setRemark("Put message back to RMQ_SYS_TRANS_HALF_TOPIC failed.");
- return response;
- }
+ public void testResumeCheckHalfMessage_WithException() throws RemotingException, InterruptedException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
+ response.setCode(ResponseCode.SYSTEM_ERROR);
+ response.setOpaque(request.getOpaque());
+ response.setRemark("Put message back to RMQ_SYS_TRANS_HALF_TOPIC failed.");
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
boolean result = mqClientAPI.resumeCheckHalfMessage(brokerAddr, "topic,", "test", 3000);
@@ -362,13 +435,10 @@ public Object answer(InvocationOnMock mock) throws Throwable {
}
@Test
- public void testResumeCheckHalfMessage_Success() throws InterruptedException, RemotingException, MQBrokerException, MQClientException {
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- RemotingCommand request = mock.getArgument(1);
- return createResumeSuccessResponse(request);
- }
+ public void testResumeCheckHalfMessage_Success() throws InterruptedException, RemotingException {
+ doAnswer(mock -> {
+ RemotingCommand request = mock.getArgument(1);
+ return createResumeSuccessResponse(request);
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
boolean result = mqClientAPI.resumeCheckHalfMessage(brokerAddr, "topic", "test", 3000);
@@ -378,16 +448,13 @@ public Object answer(InvocationOnMock mock) throws Throwable {
@Test
public void testSendMessageTypeofReply() throws Exception {
- doAnswer(new Answer() {
- @Override
- public Object answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- responseFuture.setResponseCommand(createSendMessageSuccessResponse(request));
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer(mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ responseFuture.setResponseCommand(createSendMessageSuccessResponse(request));
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(ArgumentMatchers.anyString(), ArgumentMatchers.any(RemotingCommand.class), ArgumentMatchers.anyLong(), ArgumentMatchers.any(InvokeCallback.class));
SendMessageContext sendMessageContext = new SendMessageContext();
sendMessageContext.setProducer(new DefaultMQProducerImpl(new DefaultMQProducer()));
@@ -410,19 +477,16 @@ public void onException(Throwable e) {
@Test
public void testQueryAssignment_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
-
- RemotingCommand response = RemotingCommand.createResponseCommand(null);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- QueryAssignmentResponseBody b = new QueryAssignmentResponseBody();
- b.setMessageQueueAssignments(Collections.singleton(new MessageQueueAssignment()));
- response.setBody(b.encode());
- return response;
- }
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
+
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ QueryAssignmentResponseBody b = new QueryAssignmentResponseBody();
+ b.setMessageQueueAssignments(Collections.singleton(new MessageQueueAssignment()));
+ response.setBody(b.encode());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
Set assignments = mqClientAPI.queryAssignment(brokerAddr, topic, group, clientId, null, MessageModel.CLUSTERING, 10 * 1000);
assertThat(assignments).size().isEqualTo(1);
@@ -432,48 +496,45 @@ public RemotingCommand answer(InvocationOnMock mock) {
public void testPopMessageAsync_Success() throws Exception {
final long popTime = System.currentTimeMillis();
final int invisibleTime = 10 * 1000;
- doAnswer(new Answer() {
- @Override
- public Void answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
-
- PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
- responseHeader.setInvisibleTime(invisibleTime);
- responseHeader.setPopTime(popTime);
- responseHeader.setReviveQid(0);
- responseHeader.setRestNum(1);
- StringBuilder startOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
- responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
- StringBuilder msgOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
- responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
- response.setRemark("FOUND");
- response.makeCustomHeaderToNet();
-
- MessageExt message = new MessageExt();
- message.setQueueId(0);
- message.setFlag(12);
- message.setQueueOffset(0L);
- message.setCommitLogOffset(100L);
- message.setSysFlag(0);
- message.setBornTimestamp(System.currentTimeMillis());
- message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
- message.setStoreTimestamp(System.currentTimeMillis());
- message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
- message.setBody("body".getBytes());
- message.setTopic(topic);
- message.putUserProperty("key", "value");
- response.setBody(MessageDecoder.encode(message, false));
- responseFuture.setResponseCommand(response);
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer((Answer) mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+
+ PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
+ responseHeader.setInvisibleTime(invisibleTime);
+ responseHeader.setPopTime(popTime);
+ responseHeader.setReviveQid(0);
+ responseHeader.setRestNum(1);
+ StringBuilder startOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
+ responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
+ StringBuilder msgOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
+ responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
+ response.setRemark("FOUND");
+ response.makeCustomHeaderToNet();
+
+ MessageExt message = new MessageExt();
+ message.setQueueId(0);
+ message.setFlag(12);
+ message.setQueueOffset(0L);
+ message.setCommitLogOffset(100L);
+ message.setSysFlag(0);
+ message.setBornTimestamp(System.currentTimeMillis());
+ message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
+ message.setStoreTimestamp(System.currentTimeMillis());
+ message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
+ message.setBody("body".getBytes());
+ message.setTopic(topic);
+ message.putUserProperty("key", "value");
+ response.setBody(MessageDecoder.encode(message, false));
+ responseFuture.setResponseCommand(response);
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
final CountDownLatch done = new CountDownLatch(1);
mqClientAPI.popMessageAsync(brokerName, brokerAddr, new PopMessageRequestHeader(), 10 * 1000, new PopCallback() {
@@ -501,50 +562,47 @@ public void testPopLmqMessage_async() throws Exception {
final long popTime = System.currentTimeMillis();
final int invisibleTime = 10 * 1000;
final String lmqTopic = MixAll.LMQ_PREFIX + "lmq1";
- doAnswer(new Answer() {
- @Override
- public Void answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
-
- PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
- responseHeader.setInvisibleTime(invisibleTime);
- responseHeader.setPopTime(popTime);
- responseHeader.setReviveQid(0);
- responseHeader.setRestNum(1);
- StringBuilder startOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
- responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
- StringBuilder msgOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
- responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
- response.setRemark("FOUND");
- response.makeCustomHeaderToNet();
-
- MessageExt message = new MessageExt();
- message.setQueueId(3);
- message.setFlag(0);
- message.setQueueOffset(5L);
- message.setCommitLogOffset(11111L);
- message.setSysFlag(0);
- message.setBornTimestamp(System.currentTimeMillis());
- message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
- message.setStoreTimestamp(System.currentTimeMillis());
- message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
- message.setBody("body".getBytes());
- message.setTopic(topic);
- message.putUserProperty("key", "value");
- message.putUserProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH, lmqTopic);
- message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, String.valueOf(0));
- response.setBody(MessageDecoder.encode(message, false));
- responseFuture.setResponseCommand(response);
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer((Answer) mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+
+ PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
+ responseHeader.setInvisibleTime(invisibleTime);
+ responseHeader.setPopTime(popTime);
+ responseHeader.setReviveQid(0);
+ responseHeader.setRestNum(1);
+ StringBuilder startOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
+ responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
+ StringBuilder msgOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
+ responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
+ response.setRemark("FOUND");
+ response.makeCustomHeaderToNet();
+
+ MessageExt message = new MessageExt();
+ message.setQueueId(3);
+ message.setFlag(0);
+ message.setQueueOffset(5L);
+ message.setCommitLogOffset(11111L);
+ message.setSysFlag(0);
+ message.setBornTimestamp(System.currentTimeMillis());
+ message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
+ message.setStoreTimestamp(System.currentTimeMillis());
+ message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
+ message.setBody("body".getBytes());
+ message.setTopic(topic);
+ message.putUserProperty("key", "value");
+ message.putUserProperty(MessageConst.PROPERTY_INNER_MULTI_DISPATCH, lmqTopic);
+ message.getProperties().put(MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, String.valueOf(0));
+ response.setBody(MessageDecoder.encode(message, false));
+ responseFuture.setResponseCommand(response);
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
final CountDownLatch done = new CountDownLatch(1);
final PopMessageRequestHeader requestHeader = new PopMessageRequestHeader();
@@ -580,49 +638,46 @@ public void testPopMultiLmqMessage_async() throws Exception {
final String lmqTopic2 = MixAll.LMQ_PREFIX + "lmq2";
final String multiDispatch = String.join(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER, lmqTopic, lmqTopic2);
final String multiOffset = String.join(MixAll.MULTI_DISPATCH_QUEUE_SPLITTER, "0", "0");
- doAnswer(new Answer() {
- @Override
- public Void answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
-
- PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
- responseHeader.setInvisibleTime(invisibleTime);
- responseHeader.setPopTime(popTime);
- responseHeader.setReviveQid(0);
- responseHeader.setRestNum(1);
- StringBuilder startOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
- responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
- StringBuilder msgOffsetInfo = new StringBuilder(64);
- ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
- responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
- response.setRemark("FOUND");
- response.makeCustomHeaderToNet();
-
- MessageExt message = new MessageExt();
- message.setQueueId(0);
- message.setFlag(0);
- message.setQueueOffset(10L);
- message.setCommitLogOffset(10000L);
- message.setSysFlag(0);
- message.setBornTimestamp(System.currentTimeMillis());
- message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
- message.setStoreTimestamp(System.currentTimeMillis());
- message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
- message.setBody("body".getBytes());
- message.setTopic(topic);
- MessageAccessor.putProperty(message, MessageConst.PROPERTY_INNER_MULTI_DISPATCH, multiDispatch);
- MessageAccessor.putProperty(message, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, multiOffset);
- response.setBody(MessageDecoder.encode(message, false));
- responseFuture.setResponseCommand(response);
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer((Answer) mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ RemotingCommand response = RemotingCommand.createResponseCommand(PopMessageResponseHeader.class);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+
+ PopMessageResponseHeader responseHeader = (PopMessageResponseHeader) response.readCustomHeader();
+ responseHeader.setInvisibleTime(invisibleTime);
+ responseHeader.setPopTime(popTime);
+ responseHeader.setReviveQid(0);
+ responseHeader.setRestNum(1);
+ StringBuilder startOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildStartOffsetInfo(startOffsetInfo, topic, 0, 0L);
+ responseHeader.setStartOffsetInfo(startOffsetInfo.toString());
+ StringBuilder msgOffsetInfo = new StringBuilder(64);
+ ExtraInfoUtil.buildMsgOffsetInfo(msgOffsetInfo, topic, 0, Collections.singletonList(0L));
+ responseHeader.setMsgOffsetInfo(msgOffsetInfo.toString());
+ response.setRemark("FOUND");
+ response.makeCustomHeaderToNet();
+
+ MessageExt message = new MessageExt();
+ message.setQueueId(0);
+ message.setFlag(0);
+ message.setQueueOffset(10L);
+ message.setCommitLogOffset(10000L);
+ message.setSysFlag(0);
+ message.setBornTimestamp(System.currentTimeMillis());
+ message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
+ message.setStoreTimestamp(System.currentTimeMillis());
+ message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
+ message.setBody("body".getBytes());
+ message.setTopic(topic);
+ MessageAccessor.putProperty(message, MessageConst.PROPERTY_INNER_MULTI_DISPATCH, multiDispatch);
+ MessageAccessor.putProperty(message, MessageConst.PROPERTY_INNER_MULTI_QUEUE_OFFSET, multiOffset);
+ response.setBody(MessageDecoder.encode(message, false));
+ responseFuture.setResponseCommand(response);
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
final CountDownLatch done = new CountDownLatch(1);
final PopMessageRequestHeader requestHeader = new PopMessageRequestHeader();
@@ -654,19 +709,16 @@ public void onException(Throwable e) {
@Test
public void testAckMessageAsync_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public Void answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
- response.setOpaque(request.getOpaque());
- response.setCode(ResponseCode.SUCCESS);
- responseFuture.setResponseCommand(response);
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer((Answer) mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ RemotingCommand response = RemotingCommand.createResponseCommand(ResponseCode.SUCCESS, null);
+ response.setOpaque(request.getOpaque());
+ response.setCode(ResponseCode.SUCCESS);
+ responseFuture.setResponseCommand(response);
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
final CountDownLatch done = new CountDownLatch(1);
@@ -688,22 +740,19 @@ public void onException(Throwable e) {
@Test
public void testChangeInvisibleTimeAsync_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public Void answer(InvocationOnMock mock) throws Throwable {
- InvokeCallback callback = mock.getArgument(3);
- RemotingCommand request = mock.getArgument(1);
- ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
- RemotingCommand response = RemotingCommand.createResponseCommand(ChangeInvisibleTimeResponseHeader.class);
- response.setOpaque(request.getOpaque());
- response.setCode(ResponseCode.SUCCESS);
- ChangeInvisibleTimeResponseHeader responseHeader = (ChangeInvisibleTimeResponseHeader) response.readCustomHeader();
- responseHeader.setPopTime(System.currentTimeMillis());
- responseHeader.setInvisibleTime(10 * 1000L);
- responseFuture.setResponseCommand(response);
- callback.operationSucceed(responseFuture.getResponseCommand());
- return null;
- }
+ doAnswer((Answer) mock -> {
+ InvokeCallback callback = mock.getArgument(3);
+ RemotingCommand request = mock.getArgument(1);
+ ResponseFuture responseFuture = new ResponseFuture(null, request.getOpaque(), 3 * 1000, null, null);
+ RemotingCommand response = RemotingCommand.createResponseCommand(ChangeInvisibleTimeResponseHeader.class);
+ response.setOpaque(request.getOpaque());
+ response.setCode(ResponseCode.SUCCESS);
+ ChangeInvisibleTimeResponseHeader responseHeader = (ChangeInvisibleTimeResponseHeader) response.readCustomHeader();
+ responseHeader.setPopTime(System.currentTimeMillis());
+ responseHeader.setInvisibleTime(10 * 1000L);
+ responseFuture.setResponseCommand(response);
+ callback.operationSucceed(responseFuture.getResponseCommand());
+ return null;
}).when(remotingClient).invokeAsync(anyString(), any(RemotingCommand.class), anyLong(), any(InvokeCallback.class));
final CountDownLatch done = new CountDownLatch(1);
@@ -730,16 +779,13 @@ public void onException(Throwable e) {
@Test
public void testSetMessageRequestMode_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
- RemotingCommand response = RemotingCommand.createResponseCommand(null);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
mqClientAPI.setMessageRequestMode(brokerAddr, topic, group, MessageRequestMode.POP, 8, 10 * 1000L);
@@ -747,16 +793,13 @@ public RemotingCommand answer(InvocationOnMock mock) {
@Test
public void testCreateSubscriptionGroup_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
- RemotingCommand response = RemotingCommand.createResponseCommand(null);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
mqClientAPI.createSubscriptionGroup(brokerAddr, new SubscriptionGroupConfig(), 10000);
@@ -764,16 +807,13 @@ public RemotingCommand answer(InvocationOnMock mock) {
@Test
public void testCreateTopic_Success() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
- RemotingCommand response = RemotingCommand.createResponseCommand(null);
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
mqClientAPI.createTopic(brokerAddr, topic, new TopicConfig(), 10000);
@@ -781,31 +821,28 @@ public RemotingCommand answer(InvocationOnMock mock) {
@Test
public void testViewMessage() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) throws Exception {
- RemotingCommand request = mock.getArgument(1);
-
- RemotingCommand response = RemotingCommand.createResponseCommand(null);
- MessageExt message = new MessageExt();
- message.setQueueId(0);
- message.setFlag(12);
- message.setQueueOffset(0L);
- message.setCommitLogOffset(100L);
- message.setSysFlag(0);
- message.setBornTimestamp(System.currentTimeMillis());
- message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
- message.setStoreTimestamp(System.currentTimeMillis());
- message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
- message.setBody("body".getBytes());
- message.setTopic(topic);
- message.putUserProperty("key", "value");
- response.setBody(MessageDecoder.encode(message, false));
- response.makeCustomHeaderToNet();
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
+
+ RemotingCommand response = RemotingCommand.createResponseCommand(null);
+ MessageExt message = new MessageExt();
+ message.setQueueId(0);
+ message.setFlag(12);
+ message.setQueueOffset(0L);
+ message.setCommitLogOffset(100L);
+ message.setSysFlag(0);
+ message.setBornTimestamp(System.currentTimeMillis());
+ message.setBornHost(new InetSocketAddress("127.0.0.1", 10));
+ message.setStoreTimestamp(System.currentTimeMillis());
+ message.setStoreHost(new InetSocketAddress("127.0.0.1", 11));
+ message.setBody("body".getBytes());
+ message.setTopic(topic);
+ message.putUserProperty("key", "value");
+ response.setBody(MessageDecoder.encode(message, false));
+ response.makeCustomHeaderToNet();
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
MessageExt messageExt = mqClientAPI.viewMessage(brokerAddr, "topic", 100L, 10000);
@@ -814,19 +851,16 @@ public RemotingCommand answer(InvocationOnMock mock) throws Exception {
@Test
public void testSearchOffset() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
-
- final RemotingCommand response = RemotingCommand.createResponseCommand(SearchOffsetResponseHeader.class);
- final SearchOffsetResponseHeader responseHeader = (SearchOffsetResponseHeader) response.readCustomHeader();
- responseHeader.setOffset(100L);
- response.makeCustomHeaderToNet();
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
+
+ final RemotingCommand response = RemotingCommand.createResponseCommand(SearchOffsetResponseHeader.class);
+ final SearchOffsetResponseHeader responseHeader = (SearchOffsetResponseHeader) response.readCustomHeader();
+ responseHeader.setOffset(100L);
+ response.makeCustomHeaderToNet();
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
long offset = mqClientAPI.searchOffset(brokerAddr, topic, 0, System.currentTimeMillis() - 1000, 10000);
@@ -835,19 +869,16 @@ public RemotingCommand answer(InvocationOnMock mock) {
@Test
public void testGetMaxOffset() throws Exception {
- doAnswer(new Answer() {
- @Override
- public RemotingCommand answer(InvocationOnMock mock) {
- RemotingCommand request = mock.getArgument(1);
-
- final RemotingCommand response = RemotingCommand.createResponseCommand(GetMaxOffsetResponseHeader.class);
- final GetMaxOffsetResponseHeader responseHeader = (GetMaxOffsetResponseHeader) response.readCustomHeader();
- responseHeader.setOffset(100L);
- response.makeCustomHeaderToNet();
- response.setCode(ResponseCode.SUCCESS);
- response.setOpaque(request.getOpaque());
- return response;
- }
+ doAnswer((Answer) mock -> {
+ RemotingCommand request = mock.getArgument(1);
+
+ final RemotingCommand response = RemotingCommand.createResponseCommand(GetMaxOffsetResponseHeader.class);
+ final GetMaxOffsetResponseHeader responseHeader = (GetMaxOffsetResponseHeader) response.readCustomHeader();
+ responseHeader.setOffset(100L);
+ response.makeCustomHeaderToNet();
+ response.setCode(ResponseCode.SUCCESS);
+ response.setOpaque(request.getOpaque());
+ return response;
}).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong());
long offset = mqClientAPI.getMaxOffset(brokerAddr, new MessageQueue(topic, brokerName, 0), 10000);
@@ -856,19 +887,16 @@ public RemotingCommand answer(InvocationOnMock mock) {
@Test
public void testGetMinOffset() throws Exception {
- doAnswer(new Answer