You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by do...@apache.org on 2021/11/09 08:16:10 UTC
[rocketmq] branch 5.0.0-alpha-static-topic updated: Finish the
producer to get the real broker addr
This is an automated email from the ASF dual-hosted git repository.
dongeforever pushed a commit to branch 5.0.0-alpha-static-topic
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/5.0.0-alpha-static-topic by this push:
new c0ffc5b Finish the producer to get the real broker addr
c0ffc5b is described below
commit c0ffc5bde5f0ab7c8014c7ef0ba25d4c500d33c5
Author: dongeforever <do...@apache.org>
AuthorDate: Tue Nov 9 16:15:36 2021 +0800
Finish the producer to get the real broker addr
---
.../apache/rocketmq/client/impl/MQAdminImpl.java | 24 ++-
.../rocketmq/client/impl/MQClientAPIImpl.java | 36 +---
.../ConsumeMessageConcurrentlyService.java | 2 +-
.../impl/consumer/DefaultMQPullConsumerImpl.java | 1 +
.../impl/consumer/DefaultMQPushConsumerImpl.java | 21 +-
.../client/impl/consumer/PullAPIWrapper.java | 6 +-
.../client/impl/factory/MQClientInstance.java | 38 +++-
.../impl/producer/DefaultMQProducerImpl.java | 225 +--------------------
.../rocketmq/common/sysflag/MessageSysFlag.java | 1 +
9 files changed, 90 insertions(+), 264 deletions(-)
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 dce830c..04492e3 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
@@ -191,10 +191,10 @@ public class MQAdminImpl {
if (logicalQueueRouteData != null) {
mq = logicalQueueRouteData.getMessageQueue();
}
- String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
if (null == brokerAddr) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
- brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
}
if (brokerAddr != null) {
@@ -227,10 +227,10 @@ public class MQAdminImpl {
previousQueueRouteData = maxQueueRouteData;
mq = maxQueueRouteData.getMessageQueue();
}
- String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
if (null == brokerAddr) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic);
- brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
}
if (brokerAddr != null) {
@@ -238,7 +238,8 @@ public class MQAdminImpl {
long offset = this.mQClientFactory.getMQClientAPIImpl().getMaxOffset(brokerAddr, topic, mq.getQueueId(), committed, maxQueueRouteData != null, timeoutMillis);
return correctLogicalQueueOffset(offset, maxQueueRouteData);
} catch (MQRedirectException e) {
- this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, false, null, Collections.singleton(mq.getQueueId()));
+ //TODO
+ //this.mQClientFactory.updateTopicRouteInfoFromNameServer(topic, false, null, Collections.singleton(mq.getQueueId()));
continue;
} catch (Exception e) {
throw new MQClientException("Invoke Broker[" + brokerAddr + "] exception", e);
@@ -255,10 +256,10 @@ public class MQAdminImpl {
mq = minQueueRouteData.getMessageQueue();
}
- String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
if (null == brokerAddr) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
- brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
}
if (brokerAddr != null) {
@@ -281,10 +282,11 @@ public class MQAdminImpl {
topicRouteData = this.mQClientFactory.queryTopicRouteData(mq.getTopic());
}
if (topicRouteData != null) {
- LogicalQueuesInfo logicalQueuesInfo = topicRouteData.getLogicalQueuesInfo();
+ //TODO
+ /*LogicalQueuesInfo logicalQueuesInfo = topicRouteData.getLogicalQueuesInfo();
if (logicalQueuesInfo != null) {
return logicalQueuesInfo.get(mq.getQueueId());
- }
+ }*/
}
}
return null;
@@ -296,10 +298,10 @@ public class MQAdminImpl {
mq = minQueueRouteData.getMessageQueue();
}
- String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
if (null == brokerAddr) {
this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic());
- brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
}
if (brokerAddr != null) {
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 114fd0c..c07f9fa 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
@@ -624,13 +624,9 @@ public class MQClientAPIImpl {
producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), false);
} catch (Exception e) {
- if (e instanceof MQRedirectException) {
- sendCallback.onException(e);
- } else {
- producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
- onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
- retryTimesWhenSendFailed, times, e, context, false, producer);
- }
+ producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
+ onExceptionImpl(brokerName, msg, timeoutMillis - cost, request, sendCallback, topicPublishInfo, instance,
+ retryTimesWhenSendFailed, times, e, context, false, producer);
}
} else {
producer.updateFaultItem(brokerName, System.currentTimeMillis() - responseFuture.getBeginTimestamp(), true);
@@ -672,7 +668,7 @@ public class MQClientAPIImpl {
String retryBrokerName = brokerName;//by default, it will send to the same broker
if (topicPublishInfo != null) { //select one message queue accordingly, in order to determine which broker to send
MessageQueue mqChosen = producer.selectOneMessageQueue(topicPublishInfo, brokerName);
- retryBrokerName = mqChosen.getBrokerName();
+ retryBrokerName = instance.getBrokerNameFromMessageQueue(mqChosen);
}
String addr = instance.findBrokerAddressInPublish(retryBrokerName);
log.warn(String.format("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
@@ -709,11 +705,6 @@ public class MQClientAPIImpl {
final RemotingCommand response,
final String addr
) throws MQBrokerException, RemotingCommandException {
- HashMap<String, String> extFields = response.getExtFields();
- if (extFields != null && extFields.containsKey(MessageConst.PROPERTY_REDIRECT)) {
- throw new MQRedirectException(response.getBody());
- }
-
SendStatus sendStatus;
switch (response.getCode()) {
case ResponseCode.FLUSH_DISK_TIMEOUT: {
@@ -962,11 +953,6 @@ public class MQClientAPIImpl {
private PullResult processPullResponse(
final RemotingCommand response,
final String addr) throws MQBrokerException, RemotingCommandException {
- HashMap<String, String> extFields = response.getExtFields();
- if (extFields != null && extFields.containsKey(MessageConst.PROPERTY_REDIRECT)) {
- throw new MQRedirectException(response.getBody());
- }
-
PullStatus pullStatus = PullStatus.NO_NEW_MSG;
switch (response.getCode()) {
case ResponseCode.SUCCESS:
@@ -981,6 +967,7 @@ public class MQClientAPIImpl {
case ResponseCode.PULL_OFFSET_MOVED:
pullStatus = PullStatus.OFFSET_ILLEGAL;
break;
+
default:
throw new MQBrokerException(response.getCode(), response.getRemark(), addr);
}
@@ -1649,15 +1636,8 @@ public class MQClientAPIImpl {
public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
boolean allowTopicNotExist) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
- return getTopicRouteInfoFromNameServer(topic, timeoutMillis, allowTopicNotExist, null);
- }
-
- public TopicRouteData getTopicRouteInfoFromNameServer(final String topic, final long timeoutMillis,
- boolean allowTopicNotExist, Set<Integer> logicalQueueIdsFilter) throws MQClientException, InterruptedException, RemotingTimeoutException, RemotingSendRequestException, RemotingConnectException {
GetRouteInfoRequestHeader requestHeader = new GetRouteInfoRequestHeader();
requestHeader.setTopic(topic);
- requestHeader.setSysFlag(MessageSysFlag.LOGICAL_QUEUE_FLAG);
- requestHeader.setLogicalQueueIdsFilter(logicalQueueIdsFilter);
RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_ROUTEINFO_BY_TOPIC, requestHeader);
@@ -1674,11 +1654,7 @@ public class MQClientAPIImpl {
case ResponseCode.SUCCESS: {
byte[] body = response.getBody();
if (body != null) {
- return fromNullable(RemotingSerializable.decode(body, TopicRouteDataNameSrv.class)).transform(new Function<TopicRouteDataNameSrv, TopicRouteData>() {
- @Override public TopicRouteData apply(TopicRouteDataNameSrv srv) {
- return srv.toTopicRouteData();
- }
- }).orNull();
+ return TopicRouteData.decode(body, TopicRouteData.class);
}
}
default:
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
index 702aef6..ce9355f 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java
@@ -306,7 +306,7 @@ public class ConsumeMessageConcurrentlyService implements ConsumeMessageService
// Wrap topic with namespace before sending back message.
msg.setTopic(this.defaultMQPushConsumer.withNamespace(msg.getTopic()));
try {
- this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, context.getMessageQueue().getBrokerName());
+ this.defaultMQPushConsumerImpl.sendMessageBack(msg, delayLevel, this.defaultMQPushConsumer.queueWithNamespace(context.getMessageQueue()));
return true;
} catch (Exception e) {
log.error("sendMessageBack exception, group: " + this.consumerGroup + " msg: " + msg.toString(), e);
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
index e54d9b6..d04b040 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/DefaultMQPullConsumerImpl.java
@@ -574,6 +574,7 @@ public class DefaultMQPullConsumerImpl implements MQConsumerInner {
this.offsetStore.updateConsumeOffsetToBroker(mq, offset, isOneway);
}
+ @Deprecated
public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName, String consumerGroup)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
try {
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 92ee926..d236646 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
@@ -704,11 +704,28 @@ public class DefaultMQPushConsumerImpl implements MQConsumerInner {
log.info("resume this consumer, {}", this.defaultMQPushConsumer.getConsumerGroup());
}
+ @Deprecated
public void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName)
+ throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+ sendMessageBack(msg, delayLevel, brokerName, null);
+ }
+
+ public void sendMessageBack(MessageExt msg, int delayLevel, final MessageQueue mq)
+ throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
+ sendMessageBack(msg, delayLevel, null, mq);
+ }
+
+ private void sendMessageBack(MessageExt msg, int delayLevel, final String brokerName, final MessageQueue mq)
throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
try {
- String brokerAddr = (null != brokerName) ? this.mQClientFactory.findBrokerAddressInPublish(brokerName)
- : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
+ String brokerAddr = null;
+ if (null != mq) {
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
+ } else if (null != brokerName) {
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(brokerName);
+ } else {
+ RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
+ }
this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg,
this.defaultMQPushConsumer.getConsumerGroup(), delayLevel, 5000, getMaxReconsumeTimes());
} catch (Exception e) {
diff --git a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
index 5868e30..a379b1c 100644
--- a/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
+++ b/client/src/main/java/org/apache/rocketmq/client/impl/consumer/PullAPIWrapper.java
@@ -452,7 +452,8 @@ public class PullAPIWrapper {
private void buildLogicalQueuesInfo() {
TopicRouteData topicRouteData = PullAPIWrapper.this.mQClientFactory.queryTopicRouteData(mq.getTopic());
if (topicRouteData != null) {
- this.logicalQueuesInfo = topicRouteData.getLogicalQueuesInfo();
+ //TODO
+ //this.logicalQueuesInfo = topicRouteData.getLogicalQueuesInfo();
}
}
@@ -623,7 +624,8 @@ public class PullAPIWrapper {
log.warn("LogicalQueueContext.processResponseBody {} update exception, fallback to updateTopicRouteInfoFromNameServer", this.logicalQueueRouteData, e);
}
}
- PullAPIWrapper.this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic(), false, null, Collections.singleton(this.mq.getQueueId()));
+ //TODO
+ //PullAPIWrapper.this.mQClientFactory.updateTopicRouteInfoFromNameServer(mq.getTopic(), false, null, Collections.singleton(this.mq.getQueueId()));
this.buildLogicalQueuesInfo();
}
}
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 b31757e..a7d521f 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
@@ -1046,6 +1046,24 @@ public class MQClientInstance {
return this.consumerTable.get(group);
}
+
+ public String getBrokerNameFromMessageQueue(final MessageQueue mq) {
+ if (topicEndPointsTable.get(mq.getTopic()) != null
+ && !topicEndPointsTable.get(mq.getTopic()).isEmpty()) {
+ return topicEndPointsTable.get(mq.getTopic()).get(mq);
+ }
+ return mq.getBrokerName();
+ }
+
+ public FindBrokerResult findBrokerAddressInAdmin(final MessageQueue mq) {
+ String brokerName = getBrokerNameFromMessageQueue(mq);
+ if (brokerName == null) {
+ return null;
+ } else {
+ return findBrokerAddressInAdmin(brokerName);
+ }
+ }
+
public FindBrokerResult findBrokerAddressInAdmin(final String brokerName) {
String brokerAddr = null;
boolean slave = false;
@@ -1076,6 +1094,15 @@ public class MQClientInstance {
return null;
}
+ public String findBrokerAddressInPublish(final MessageQueue mq) {
+ String brokerName = getBrokerNameFromMessageQueue(mq);
+ if (brokerName == null) {
+ return null;
+ } else {
+ return findBrokerAddressInPublish(brokerName);
+ }
+ }
+ //This is used for retry only
public String findBrokerAddressInPublish(final String brokerName) {
HashMap<Long/* brokerId */, String/* address */> map = this.brokerAddrTable.get(brokerName);
if (map != null && !map.isEmpty()) {
@@ -1085,6 +1112,15 @@ public class MQClientInstance {
return null;
}
+ public FindBrokerResult findBrokerAddressInSubscribe(final MessageQueue mq, final long brokerId, final boolean onlyThisBroker) {
+ String brokerName = getBrokerNameFromMessageQueue(mq);
+ if (brokerName == null) {
+ return null;
+ } else {
+ return findBrokerAddressInSubscribe(brokerName, brokerId, onlyThisBroker);
+ }
+ }
+
public FindBrokerResult findBrokerAddressInSubscribe(
final String brokerName,
final long brokerId,
@@ -1120,7 +1156,7 @@ public class MQClientInstance {
return null;
}
- public int findBrokerVersion(String brokerName, String brokerAddr) {
+ private int findBrokerVersion(String brokerName, String brokerAddr) {
if (this.brokerVersionTable.containsKey(brokerName)) {
if (this.brokerVersionTable.get(brokerName).containsKey(brokerAddr)) {
return this.brokerVersionTable.get(brokerName).get(brokerAddr);
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 00df89c..578c843 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
@@ -730,45 +730,15 @@ public class DefaultMQProducerImpl implements MQProducerInner {
final SendCallback sendCallback,
final TopicPublishInfo topicPublishInfo,
final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
- if (MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME.equals(mq.getBrokerName())) {
- LogicalQueueSendContext logicalQueueContext = new LogicalQueueSendContext(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
- while (true) {
- try {
- SendResult sendResult = this.sendKernelImplWithoutRetry(msg,
- logicalQueueContext.getModifiedMessageQueue(),
- communicationMode,
- logicalQueueContext.wrapSendCallback(),
- topicPublishInfo,
- timeout);
- return logicalQueueContext.wrapSendResult(sendResult);
- } catch (MQRedirectException e) {
- if (!logicalQueueContext.shouldRetry(e)) {
- throw new MQBrokerException(ResponseCode.SYSTEM_ERROR, "redirect");
- }
- } catch (RemotingException e) {
- if (!logicalQueueContext.shouldRetry(e)) {
- throw e;
- }
- }
- }
- } else {
- return sendKernelImplWithoutRetry(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
- }
- }
-
- private SendResult sendKernelImplWithoutRetry(final Message msg,
- final MessageQueue mq,
- final CommunicationMode communicationMode,
- SendCallback sendCallback,
- final TopicPublishInfo topicPublishInfo,
- final long timeout) throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
long beginStartTime = System.currentTimeMillis();
- String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
if (null == brokerAddr) {
tryToFindTopicPublishInfo(mq.getTopic());
- brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq.getBrokerName());
+ brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(mq);
}
+ String brokerName = this.mQClientFactory.getBrokerNameFromMessageQueue(mq);
+
SendMessageContext context = null;
if (brokerAddr != null) {
brokerAddr = MixAll.brokerVIPChannel(this.defaultMQProducer.isSendMessageWithVIPChannel(), brokerAddr);
@@ -798,10 +768,6 @@ public class DefaultMQProducerImpl implements MQProducerInner {
sysFlag |= MessageSysFlag.TRANSACTION_PREPARED_TYPE;
}
- if (!CommunicationMode.ONEWAY.equals(communicationMode)) {
- sysFlag |= MessageSysFlag.LOGICAL_QUEUE_FLAG;
- }
-
if (hasCheckForbiddenHook()) {
CheckForbiddenContext checkForbiddenContext = new CheckForbiddenContext();
checkForbiddenContext.setNameSrvAddr(this.defaultMQProducer.getNamesrvAddr());
@@ -890,7 +856,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
- mq.getBrokerName(),
+ brokerName,
tmpMessage,
requestHeader,
timeout - costTimeAsync,
@@ -910,7 +876,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage(
brokerAddr,
- mq.getBrokerName(),
+ brokerName,
msg,
requestHeader,
timeout - costTimeSync,
@@ -941,7 +907,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
}
}
- throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
+ throw new MQClientException("The broker[" + brokerName + "] not exist", null);
}
public MQClientInstance getmQClientFactory() {
@@ -1042,7 +1008,6 @@ public class DefaultMQProducerImpl implements MQProducerInner {
executeEndTransactionHook(context);
}
}
-
/**
* DEFAULT ONEWAY -------------------------------------------------------
*/
@@ -1377,7 +1342,7 @@ public class DefaultMQProducerImpl implements MQProducerInner {
id = MessageDecoder.decodeMessageId(sendResult.getMsgId());
}
String transactionId = sendResult.getTransactionId();
- final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue().getBrokerName());
+ final String brokerAddr = this.mQClientFactory.findBrokerAddressInPublish(sendResult.getMessageQueue());
EndTransactionRequestHeader requestHeader = new EndTransactionRequestHeader();
requestHeader.setTransactionId(transactionId);
requestHeader.setCommitLogOffset(id.getOffset());
@@ -1682,178 +1647,4 @@ public class DefaultMQProducerImpl implements MQProducerInner {
public DefaultMQProducer getDefaultMQProducer() {
return defaultMQProducer;
}
-
- private class LogicalQueueSendContext implements SendCallback {
- private final Message msg;
- private final MessageQueue mq;
- private final CommunicationMode communicationMode;
- private final SendCallback sendCallback;
- private final TopicPublishInfo topicPublishInfo;
- private final long timeout;
-
- private volatile LogicalQueuesInfo logicalQueuesInfo;
- private volatile LogicalQueueRouteData writableQueueRouteData;
-
- private final AtomicInteger retry = new AtomicInteger();
-
- public LogicalQueueSendContext(Message msg, MessageQueue mq,
- CommunicationMode communicationMode, SendCallback sendCallback,
- TopicPublishInfo topicPublishInfo, long timeout) {
- this.msg = msg;
- this.mq = mq;
- this.communicationMode = communicationMode;
- this.sendCallback = sendCallback;
- this.topicPublishInfo = topicPublishInfo;
- this.timeout = timeout;
-
- if (topicPublishInfo == null) {
- topicPublishInfo = DefaultMQProducerImpl.this.tryToFindTopicPublishInfo(mq.getTopic());
- }
- if (topicPublishInfo != null) {
- this.logicalQueuesInfo = topicPublishInfo.getTopicRouteData().getLogicalQueuesInfo();
- } else {
- this.logicalQueuesInfo = null;
- }
- }
-
- private boolean notUsingLogicalQueue() {
- return !Objects.equal(mq.getBrokerName(), MixAll.LOGICAL_QUEUE_MOCK_BROKER_NAME) || this.logicalQueuesInfo == null;
- }
-
- public MessageQueue getModifiedMessageQueue() throws MQClientException {
- if (this.notUsingLogicalQueue()) {
- return this.mq;
- }
- this.writableQueueRouteData = getWritableQueueRouteData();
- MessageQueue mq = new MessageQueue(this.mq);
- mq.setBrokerName(writableQueueRouteData.getBrokerName());
- mq.setQueueId(writableQueueRouteData.getQueueId());
- return mq;
- }
-
- private LogicalQueueRouteData getWritableQueueRouteData() throws MQClientException {
- this.logicalQueuesInfo.readLock().lock();
- try {
- List<LogicalQueueRouteData> queueRouteDataList = logicalQueuesInfo.get(mq.getQueueId());
- if (queueRouteDataList == null || queueRouteDataList.size() == 0) {
- throw new MQClientException(String.format(Locale.ENGLISH, "send to a logical queue %d but no queue route data found", mq.getQueueId()), null);
- }
- // usually writable queue is placed in the last position, or second last when queue migrating
- for (int i = queueRouteDataList.size() - 1; i >= 0; i--) {
- LogicalQueueRouteData queueRouteData = queueRouteDataList.get(i);
- if (queueRouteData.isWritable()) {
- return queueRouteData;
- }
- }
- throw new MQClientException(String.format(Locale.ENGLISH, "send to a logical queue %d but no writable queue route data found", mq.getQueueId()), null);
- } finally {
- this.logicalQueuesInfo.readLock().unlock();
- }
- }
-
- @Override public void onSuccess(SendResult sendResult) {
- this.sendCallback.onSuccess(this.wrapSendResult(sendResult));
- }
-
- @Override public void onException(Throwable t) {
- if (this.shouldRetry(t)) {
- try {
- DefaultMQProducerImpl.this.sendKernelImplWithoutRetry(msg, this.getModifiedMessageQueue(), communicationMode, this, topicPublishInfo, timeout);
- return;
- } catch (Exception e) {
- t = e;
- }
- }
- if (t instanceof MQRedirectException) {
- t = new MQBrokerException(ResponseCode.SYSTEM_ERROR, "redirect");
- }
- this.sendCallback.onException(t);
- }
-
- private void handleRedirectException(MQRedirectException re) {
- byte[] responseBody = re.getBody();
- log.info("LogicalQueueContext.processResponseBody got redirect {}: {}", this.writableQueueRouteData, responseBody != null ? new String(responseBody, MessageDecoder.CHARSET_UTF8) : null);
-
- try {
- List<LogicalQueueRouteData> newQueueRouteDataList = JSON.parseObject(responseBody, MixAll.TYPE_LIST_LOGICAL_QUEUE_ROUTE_DATA);
- this.logicalQueuesInfo.updateLogicalQueueRouteDataList(this.mq.getQueueId(), newQueueRouteDataList);
- } catch (Exception e) {
- log.warn("LogicalQueueContext.processResponseBody {} update exception, fallback to updateTopicRouteInfoFromNameServer", this.writableQueueRouteData, e);
- DefaultMQProducerImpl.this.mQClientFactory.updateTopicRouteInfoFromNameServer(this.mq.getTopic(), false, null, Collections.singleton(mq.getQueueId()));
- TopicRouteData topicRouteData = DefaultMQProducerImpl.this.mQClientFactory.getAnExistTopicRouteData(mq.getTopic());
- if (topicRouteData != null) {
- this.logicalQueuesInfo = topicRouteData.getLogicalQueuesInfo();
- } else {
- this.logicalQueuesInfo = null;
- }
- }
- }
-
- public SendCallback wrapSendCallback() {
- if (this.notUsingLogicalQueue()) {
- return this.sendCallback;
- }
- if (!CommunicationMode.ASYNC.equals(this.communicationMode)) {
- return this.sendCallback;
- }
- return this;
- }
-
- public boolean shouldRetry(Throwable t) {
- this.incrRetry();
- if (this.exceedMaxRetry()) {
- log.warn("retry {} too many times: {}", this.retry.get(), this.writableQueueRouteData);
- return false;
- }
- if (!this.writableQueueRouteData.isWritable()) {
- log.warn("no writable queue: {}", this.writableQueueRouteData);
- return false;
- }
- if (t instanceof MQRedirectException) {
- this.handleRedirectException((MQRedirectException) t);
- return true;
- }
- return !(t instanceof RemotingException) || this.handleRemotingException((RemotingException) t);
- }
-
- public boolean exceedMaxRetry() {
- return this.retry.get() >= 3;
- }
-
- public void incrRetry() {
- this.retry.incrementAndGet();
- }
-
- public SendResult wrapSendResult(SendResult sendResult) {
- if (sendResult == null) {
- return null;
- }
- SendResultForLogicalQueue newSendResult = new SendResultForLogicalQueue(sendResult, this.writableQueueRouteData.getLogicalQueueIndex());
- long queueOffset = newSendResult.getQueueOffset();
- if (queueOffset >= 0) {
- newSendResult.setQueueOffset(LogicalQueueSendContext.this.writableQueueRouteData.toLogicalQueueOffset(queueOffset));
- }
- return newSendResult;
- }
-
- public boolean handleRemotingException(RemotingException e) {
- if (e instanceof RemotingTooMuchRequestException) {
- return false;
- }
- DefaultMQProducerImpl.this.mQClientFactory.updateTopicRouteInfoFromNameServer(this.mq.getTopic(), false, null, Collections.singleton(mq.getQueueId()));
- this.logicalQueuesInfo = DefaultMQProducerImpl.this.getTopicPublishInfoTable().get(mq.getTopic()).getTopicRouteData().getLogicalQueuesInfo();
- LogicalQueueRouteData writableQueueRouteData;
- try {
- writableQueueRouteData = this.getWritableQueueRouteData();
- } catch (MQClientException ce) {
- log.warn("getWritableQueueRouteData exception: {}", this.logicalQueuesInfo.get(mq.getQueueId()), ce);
- return false;
- }
- if (Objects.equal(this.writableQueueRouteData.getMessageQueue(), writableQueueRouteData.getMessageQueue()) && writableQueueRouteData.isWritable()) {
- // still same MessageQueue and still writable, no need to retry
- return false;
- }
- return true;
- }
- }
}
diff --git a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java b/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
index 9f39f48..ef0cd62 100644
--- a/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
+++ b/common/src/main/java/org/apache/rocketmq/common/sysflag/MessageSysFlag.java
@@ -25,6 +25,7 @@ public class MessageSysFlag {
public final static int TRANSACTION_ROLLBACK_TYPE = 0x3 << 2;
public final static int BORNHOST_V6_FLAG = 0x1 << 4;
public final static int STOREHOSTADDRESS_V6_FLAG = 0x1 << 5;
+ //TODO remove the LOGICAL_QUEUE_FLAG
public final static int LOGICAL_QUEUE_FLAG = 0x1 << 6;
public static int getTransactionValue(final int flag) {