提交 01a0eb00 编写于 作者: J Jaskey 提交者: Zhanhui Li

Fix possible NullPointerException when retry in send Async way

上级 15af63e2
...@@ -438,34 +438,40 @@ public class MQClientAPIImpl { ...@@ -438,34 +438,40 @@ public class MQClientAPIImpl {
) { ) {
int tmp = curTimes.incrementAndGet(); int tmp = curTimes.incrementAndGet();
if (needRetry && tmp <= timesTotal) { if (needRetry && tmp <= timesTotal) {
MessageQueue tmpmq = producer.selectOneMessageQueue(topicPublishInfo, brokerName); String retryBrokerName = brokerName;//by default, it will send to the same broker
String addr = instance.findBrokerAddressInPublish(tmpmq.getBrokerName()); 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();
}
String addr = instance.findBrokerAddressInPublish(retryBrokerName);
log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr, log.info("async send msg by retry {} times. topic={}, brokerAddr={}, brokerName={}", tmp, msg.getTopic(), addr,
tmpmq.getBrokerName()); retryBrokerName);
try { try {
request.setOpaque(RemotingCommand.createNewRequestId()); request.setOpaque(RemotingCommand.createNewRequestId());
sendMessageAsync(addr, tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, sendMessageAsync(addr, retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance,
timesTotal, curTimes, context, producer); timesTotal, curTimes, context, producer);
} catch (InterruptedException e1) { } catch (InterruptedException e1) {
onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, false, producer); context, false, producer);
} catch (RemotingConnectException e1) { } catch (RemotingConnectException e1) {
producer.updateFaultItem(brokerName, 3000, true); producer.updateFaultItem(brokerName, 3000, true);
onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, true, producer); context, true, producer);
} catch (RemotingTooMuchRequestException e1) { } catch (RemotingTooMuchRequestException e1) {
onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, false, producer); context, false, producer);
} catch (RemotingException e1) { } catch (RemotingException e1) {
producer.updateFaultItem(brokerName, 3000, true); producer.updateFaultItem(brokerName, 3000, true);
onExceptionImpl(tmpmq.getBrokerName(), msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1, onExceptionImpl(retryBrokerName, msg, timeoutMillis, request, sendCallback, topicPublishInfo, instance, timesTotal, curTimes, e1,
context, true, producer); context, true, producer);
} }
} else { } else {
if (context != null) { if (context != null) {
context.setException(e); context.setException(e);
context.getProducer().executeSendMessageHookAfter(context); context.getProducer().executeSendMessageHookAfter(context);
} }
try { try {
sendCallback.onException(e); sendCallback.onException(e);
} catch (Exception ignored) { } catch (Exception ignored) {
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册