“This is the 17th day of my participation in the First Challenge 2022. For details: First Challenge 2022.”

RocketMQ consumption failure retry mechanism analysis

Today we will examine the RocketMQ consumption retry mechanism if the following code for the RocketMQ consumption service is executed

try {
            try {
                if (messageExtWrappers.size() > 0) {
                    try {
                        var22 = messageExtWrappers.iterator();
                        while(var22.hasNext()) { messageExt = (MessageExt)var22.next(); }}catch (Throwable var16) {
                        ;
                    }
                    this.consume(messageExtWrappers, context);
                }
                LOGGER.info("MQ_CON_SUCCESS {} BROKER {} QUEUE {}".new Object[]{topic, broker, queueId});
                ConsumeConcurrentlyStatus var23 = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                return var23;
            } catch (MessageListenerConcurrentlyException var17) {
                LOGGER.error("MQ_CON_EX {} BROKER {} QUEUE {}".new Object[]{topic, broker, queueId, var17});
                throw var17;
            } catch (Throwable var18) {
                LOGGER.error("MQ_CON_EX {} BROKER {} QUEUE {}".new Object[]{topic, broker, queueId, var18});
                LOGGER.info("MQ_CON_RECONSUME {} BROKER {} QUEUE {}".new Object[]{topic, broker, queueId});
                if (CollectionUtils.isNotEmpty(msgs) && ((MessageExt)msgs.get(0)).getDelayTimeLevel() >= 2 + this.retryTimes) {
                    context.setDelayLevelWhenNextConsume(-1); }}Copy the code

When consuming data, the resulting log is as follows:

First consumption

MQ_CON_MSG topic MSG MessageExt [queueId=1, storeSize=453, queueOffset=25, sysFlag=0, bornTimestamp=1566785215908, BornHost = / 10.42.0.77:54608, storeTimestamp = 1566785215908, storeHost = / 10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B77CE84, commitLogOffset=192401028, bodyCRC=53737244, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic=’gmcf-lsc-zhongbang-repu-calc-from-topic’, flag=0, properties={MIN_OFFSET=0, _catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15071, HASH_CODE=690132963, MAX_OFFSET=26, _catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785215911, SERIALIZE_CLASS=java.lang.String, _catParentMessageId=lts-0a2a006b-435218-524, _catChildMessageId=gmcf-lsc-job-0a2a004d-435218-15072, UNIQ_KEY=0A2A004D000938AF386882EAA5A40112, WAIT=true}, body=[71, 77, 66, 69, 67, 49, 57, 48, 56, 50, 52, 49, 48, 52, 48, 52, 54, 56, 57, 52, 48, 52, 52, 48, 56, 48], transactionId=’null ‘}]

First retry (reconsumeTimes=1,DELAY=3)

MQ_CON_MSG %RETRY%consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1187, sysFlag=0, BornTimestamp = 1566785215923, bornHost = / 10.42.0.77:54608, storeTimestamp = 1566785226241, storeHost = / 10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B785900, commitLogOffset=192436480, bodyCRC=893293938, reconsumeTimes=1, preparedTransactionOffset=0, toString()=Message{topic=’gmcf-lsc-zhongbang-repu-calc-from-topic’, flag=0, properties={_catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15075, _catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785226242, SERIALIZE_CLASS=java.lang.String, _catParentMessageId=lts-0a2a006b-435218-524, _catChildMessageId=gmcf-lsc-job-0a2a004d-435218-15076, MIN_OFFSET=0, REAL_TOPIC=%RETRY%gmcf-lsc-consumerGroup, ORIGIN_MESSAGE_ID=0A2A00F400002A9F000000000B77D049, RETRY_TOPIC=gmcf-lsc-zhongbang-repu-calc-from-topic, HASH_CODE=1506102461, MAX_OFFSET=1188, UNIQ_KEY=0A2A004D000938AF386882EAA5B30113, WAIT=false, DELAY=3, REAL_QID=0}, body=[71, 77, 66, 69, 67, 49, 57, 48, 56, 50, 52, 49, 48, 52, 49, 49, 49, 50, 48, 55, 48, 52, 48, 56, 49], transactionId=’null’}]

Second retry (reconsumeTimes=2, DELAY=4)

MQ_CON_MSG %RETRY%consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1209, sysFlag=0, BornTimestamp = 1566785215923, bornHost = / 10.42.0.77:54608, storeTimestamp = 1566785256680, storeHost = / 10.42.0.244:10911, msgId=0A2A00F400002A9F000000000B791399, commitLogOffset=192484249, bodyCRC=893293938, reconsumeTimes=2, preparedTransactionOffset=0, toString()=Message{topic=’gmcf-lsc-zhongbang-repu-calc-from-topic’, flag=0, properties={_catChildMessageId1=gmcf-lsc-job-0a2a004d-435218-15075, _catRootMessageId=lts-0a2a006b-434483-0, CONSUME_START_TIME=1566785256728, SERIALIZE_CLASS=java.lang.String, _catParentMessageId=lts-0a2a006b-435218-524, _catChildMessageId=gmcf-lsc-job-0a2a004d-435218-15076, MIN_OFFSET=0, REAL_TOPIC=%RETRY%gmcf-lsc-consumerGroup, ORIGIN_MESSAGE_ID=0A2A00F400002A9F000000000B77D049, RETRY_TOPIC=gmcf-lsc-zhongbang-repu-calc-from-topic, HASH_CODE=1506102461, MAX_OFFSET=1210, UNIQ_KEY=0A2A004D000938AF386882EAA5B30113, WAIT=false, DELAY=4, REAL_QID=0}, body=[71, 77, 66, 69, 67, 49, 57, 48, 56, 50, 52, 49, 48, 52, 49, 49, 49, 50, 48, 55, 48, 52, 48, 56, 49], transactionId=’null’}]

Third retry (reconsumeTimes=3, DELAY=5)

MQ_CON_MSG %RETRY%consumerGroup MSG MessageExt [queueId=0, storeSize=608, queueOffset=1228, sysFlag=0, bornTimestamp=1566785215923, b

The default configuration of RocketMQ is messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h, respectively representing latency level1-Level18, Why don’t you start at one?

if (CollectionUtils.isNotEmpty(msgs) && ((MessageExt)msgs.get(0)).getDelayTimeLevel() >= 2 + this.retryTimes) {
        context.setDelayLevelWhenNextConsume(-1);
}
Copy the code

When the retry condition is met, the DLQ queue is directly added to the DLQ queue. If not, the system will retry until the maximum DelayLevel is 18.

Find a piece of code from DefaultMQPullConsumerImpl class

public void sendMessageBack(MessageExt msg, int delayLevel, String brokerName, String consumerGroup) throws RemotingException, MQBrokerException, InterruptedException, MQClientException {
        try {
            String brokerAddr = null! = brokerName ?this.mQClientFactory.findBrokerAddressInPublish(brokerName) : RemotingHelper.parseSocketAddressAddr(msg.getStoreHost());
            if (UtilAll.isBlank(consumerGroup)) {
                consumerGroup = this.defaultMQPullConsumer.getConsumerGroup();
            }

            this.mQClientFactory.getMQClientAPIImpl().consumerSendMessageBack(brokerAddr, msg, consumerGroup, delayLevel, 3000L.this.defaultMQPullConsumer.getMaxReconsumeTimes());
        } catch (Exception var8) {
            this.log.error("sendMessageBack Exception, " + this.defaultMQPullConsumer.getConsumerGroup(), var8);
            Message newMsg = new Message(MixAll.getRetryTopic(this.defaultMQPullConsumer.getConsumerGroup()), msg.getBody());
            String originMsgId = MessageAccessor.getOriginMessageId(msg);
            MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
            newMsg.setFlag(msg.getFlag());
            MessageAccessor.setProperties(newMsg, msg.getProperties());
            MessageAccessor.putProperty(newMsg, "RETRY_TOPIC", msg.getTopic());
            MessageAccessor.setReconsumeTime(newMsg, String.valueOf(msg.getReconsumeTimes() + 1));
            MessageAccessor.setMaxReconsumeTimes(newMsg, String.valueOf(this.defaultMQPullConsumer.getMaxReconsumeTimes()));
            newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());
            this.mQClientFactory.getDefaultMQProducer().send(newMsg); }}Copy the code

See from the code DelayTimeLevel =3+reconsumeTime

newMsg.setDelayTimeLevel(3 + msg.getReconsumeTimes());

Therefore, the default retry starts from 3. From the perspective of time, it is also verified that four retries are performed, and each interval is 10s, 30s, and 1m.

  • In addition, the latency level is set to 9 when sending, so it will take 5m for the consumer to fetch the message from the broker. Failure or abnormal consumption, consumers returned to the broker, ConsumeConcurrentlyStatus. RECONSUME_LATER, so the broker will resend the message, as consumers set the consumption level of delay for 5 again, so after the 1 m, The consumer can then retrieve the resend message from the broker. Since the consumer does not set the retry count, the broker defaults to resending the message 16 times and then putting it on the delete queue.

  • DefaultRocketMQListenerContainer MessageListenerConcurrently method is achieved, it will cycle call rocketMQListener. The onMessage, Abnormal sets delayLevelWhenNextConsume, then immediately return ConsumeConcurrentlyStatus. RECONSUME_LATER