possbile LitePullConusmer rebalance bug

[Zanglei06]

1. Please describe the issue you observed:

• What did you do (The steps to reproduce)?

• What did you expect to see?

• What did you see instead?

In our production environment, I find some msgs lost when a new consumer started(causing rebalance), the RMQ version we use is 4.7.1 and we use the new LitePullConsumer API.

From the rocketmq-client log, something unexpected happened:

1. the same messageQueue is detected should cancel its pullTask in two different threads in almost the same time.

2021-03-09 20:16:19.911 WARN [PullMsgThread-c_g1] (Slf4jLoggerFactory.java:115) - The Pull Task is cancelled after doPullTask, MessageQueue [topic=t, brokerName=rmq-b5, queueId=3]
2021-03-09 20:16:19.911 WARN [PullMsgThread-c_g2] (Slf4jLoggerFactory.java:115) - The Pull Task is cancelled after doPullTask, MessageQueue [topic=t, brokerName=rmq-b5, queueId=3]

2. before rebalance, there are two consumer(two consumer cid), when a new consumer is started, it becomes three(3 consumer cid), but the rebalance triggered several times with wrong cid count, which means findConsumerIdList API returns wrong value. I think the wrong rebalance should not cause any msg lost since rebalance is done in a single thread and finally should be correct. But why wrong cid returned is interesting( calling a wrong broker? slave? but why ?).

below is the logs for rebalance result.(I changed some inner ip and brokerName info for security reasons)

2021-03-09 20:16:19.777 INFO [RebalanceService] (Slf4jLoggerFactory.java:100) - rebalanced result changed. allocateMessageQueueStrategyName=AVG, group=c_g, topic=t, clientId=XXX_C1, mqAllSize=24, cidAllSize=3, rebalanceResultSize=8, rebalanceResultSet=XXX (3 cid, corrent)
2021-03-09 20:16:19.779 INFO [RebalanceService] (Slf4jLoggerFactory.java:100) - rebalanced result changed. allocateMessageQueueStrategyName=AVG, group=c_g, topic=t, clientId=XXX_C1, mqAllSize=24, cidAllSize=2, rebalanceResultSize=12, rebalanceResultSet=XXX (2 cid, wrong)
2021-03-09 20:16:19.781 INFO [RebalanceService] (Slf4jLoggerFactory.java:100) - rebalanced result changed. allocateMessageQueueStrategyName=AVG, group=c_g, topic=t, clientId=XXX_C1, mqAllSize=24, cidAllSize=3, rebalanceResultSize=8, rebalanceResultSet=XXX (3 cid, correct)
2021-03-09 20:16:19.784 INFO [RebalanceService] (Slf4jLoggerFactory.java:100) - rebalanced result changed. allocateMessageQueueStrategyName=AVG, group=c_g, topic=t, clientId=XXX_C1, mqAllSize=24, cidAllSize=2, rebalanceResultSize=12, rebalanceResultSet=XXX (2 cid , wrong)
2021-03-09 20:16:19.785 INFO [RebalanceService] (Slf4jLoggerFactory.java:100) - rebalanced result changed. allocateMessageQueueStrategyName=AVG, group=c_g, topic=t, clientId=XXX_C1, mqAllSize=24, cidAllSize=3, rebalanceResultSize=8, rebalanceResultSet=XXX (3 cid, correct)

3. from rocketmq-client log the rebalance notification from broker found slave broker ips

additional info:
in one java process we have one consumer and one producer with different clientId; the consumer is polling messages for one group and one topic(only one subscription); the producer is sending messages to many topics( different from consumer topic);

2. Please tell us about your environment:

RMQ 4.7.1

LitePullConsumer

[ifplusor]

When a new consumer sends a heartbeat to a broker, the broker will notify all consumers of the related group to do rebalance. If there are multiple brokers, a consumer will receive multiple notifications.

In the rebalance process, the consumer selects a broker randomly to look up the consumer list. If the consumer did not send any heartbeat to the broker at the time, the returned list is incorrect.

[Zanglei06]

When a new consumer sends a heartbeat to a broker, the broker will notify all consumers of the related group to do rebalance. If there are multiple brokers, a consumer will receive multiple notifications.

In the rebalance process, the consumer selects a broker randomly to look up the consumer list. If the consumer did not send any heartbeat to the broker at the time, the returned list is incorrect.

yes, and incorrect consumer list should not result in lost msgs since it will be corrected anyway. My major confusion is the first question, why one messageQueue is processed(Pull Task) by two pull threads(suggested by rocketmq-client log). Do you have any ideas?

[ifplusor]

@Zanglei06 Multiple rebalance processes with different consumer lists will cause this issue.

[lizhimins]

Heartbeat failure may cause load balancing problems

[Zanglei06]

Heartbeat failure may cause load balancing problems

the LB problems should not include msgs lost. only some short time msg duplicates occurs. can you explain why the 1st log appears?

[areyouok]

I had reproduce the bug in this commit:
areyouok@a54599e

At first, create a topic with only one queue:
./mqadmin updateTopic -n 127.0.0.1:9876 -t TopicTest -r 1 -w 1 -c DefaultCluster

then run LitePullConsumerBug, it send two messages, receive one, and the offset is update to consumerOffset.json

The output:

RocketMQLog:WARN No appenders could be found for logger (io.netty.util.internal.PlatformDependent0).
RocketMQLog:WARN Please initialize the logger system properly.
SendResult [sendStatus=SEND_OK, msgId=AC1128381E4718B4AAC2691C5DF80000, offsetMsgId=AC11283800002A9F0000000000000C60, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=18]
09:51:07,801 1749887142 get queue
09:51:07,803 1749887142 pull step 1
09:51:08,776 start consumer 2
09:51:09,787 203029914 get queue
09:51:09,787 1749887142 lost queue
09:51:09,788 203029914 pull step 1
09:51:10,790 1749887142 get queue
09:51:10,791 1749887142 pull step 1
09:51:10,808 1749887142 pull step 2
09:51:10,814 submit 1 msgs
09:51:10,815 drop 1 msgs
09:51:12,792 203029914 pull step 2
09:51:13,793 1749887142 pull step 2
SendResult [sendStatus=SEND_OK, msgId=AC1128381E4718B4AAC2691C815A0001, offsetMsgId=AC11283800002A9F0000000000000D10, messageQueue=MessageQueue [topic=TopicTest, brokerName=broker-a, queueId=0], queueOffset=19]
09:51:15,806 submit 1 msgs
receive msg: AC1128381E4718B4AAC2691C815A0001
09:51:15,807 1749887142 pull step 1
09:51:18,807 1749887142 pull step 2

The root cause in DefaultLitePullConsumerImpl.PullTaskImpl.run
...
long offset = nextPullOffset(messageQueue);
...
updatePullOffset(messageQueue, pullResult.getNextBeginOffset());

After a ProcessQueue is dropped, but the PullTask is still run. The invalid task will call updatePullOffset.

[RongtongJin]

Merged