Client based 4.x remoting consumer retry message failure

[wangxye]

Background: Client subscription consumption through push mode, for the message delivered over, the first two times do not consume, so that the broker to go retry logic, in the third time only normal consumption.

situation：Producer sends 5 messages, but for the retry message, the client receives only one and the others are not received.

2023-11-15 10:47:54 INFO  [o.a.r.f.BaseOperate           #getTopic            :  72] create topic: topic_testNormalMessageReconsumeTime_xRXlIv , topicId:1536
2023-11-15 10:47:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x3db2b183, L:/127.0.0.1:54012 - R:/127.0.0.1:8081] OUTBOUND HEADERS: streamId=5 headers=GrpcHttp2OutboundHeaders[:authority: 127.0.0.1:8081, :path: /apache.rocketmq.controller.v1.ControllerService/createGroup, :method: POST, :scheme: http, content-type: application/grpc, te: trailers, user-agent: grpc-java-netty/1.58.0, grpc-accept-encoding: gzip, grpc-timeout: 2999758u] streamDependency=0 weight=16 exclusive=false padding=0 endStream=false
2023-11-15 10:47:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x3db2b183, L:/127.0.0.1:54012 - R:/127.0.0.1:8081] OUTBOUND DATA: streamId=5 padding=0 endStream=true length=54 bytes=000000003112294749445f746573744e6f726d616c4d6573736167655265636f6e73756d6554696d655f59564a6a6e47181020013801
2023-11-15 10:47:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x3db2b183, L:/127.0.0.1:54012 - R:/127.0.0.1:8081] INBOUND HEADERS: streamId=5 headers=GrpcHttp2ResponseHeaders[:status: 200, content-type: application/grpc, grpc-encoding: identity, grpc-accept-encoding: gzip] padding=0 endStream=false
2023-11-15 10:47:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x3db2b183, L:/127.0.0.1:54012 - R:/127.0.0.1:8081] INBOUND DATA: streamId=5 padding=0 endStream=false length=10 bytes=00000000050a0010f206
2023-11-15 10:47:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x3db2b183, L:/127.0.0.1:54012 - R:/127.0.0.1:8081] INBOUND HEADERS: streamId=5 headers=GrpcHttp2ResponseHeaders[grpc-status: 0] padding=0 endStream=true
2023-11-15 14:08:54 INFO  [o.a.r.f.BaseOperate           #getGroupId          :  98] [ConsumerGroupId] groupId:GID_testNormalMessageReconsumeTime_DUURJv , methodName:testNormalMessageReconsumeTime , mode: SUB_MODE_PULL , reply:status {
}
group_id: 898

2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  68] Producer start to send messages
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  75] SendResult [sendStatus=SEND_OK, msgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A4F0000, offsetMsgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A4F0000, messageQueue=MessageQueue [topic=topic_testNormalMessageReconsumeTime_bGVeiG, brokerName=1568_5, queueId=0], queueOffset=0], index: 0, tag: tag-server-jVlHhwhvmlpkrLIqle6R
2023-11-15 14:08:54 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 114] recv msg(first) MessageExt [brokerName=1568_5, queueId=0, storeSize=272, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534361, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='topic_testNormalMessageReconsumeTime_bGVeiG', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028534391, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223A4F0000, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[54, 48, 50, 57, 99, 99, 48, 51, 45, 101, 53, 54, 98, 45, 52, 55, 52, 100, 45, 57, 55, 52, 102, 45, 98, 49, 97, 50, 101, 101, 50, 53, 100, 56, 48, 50], transactionId='null'}] 
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  75] SendResult [sendStatus=SEND_OK, msgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A720001, offsetMsgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A720001, messageQueue=MessageQueue [topic=topic_testNormalMessageReconsumeTime_bGVeiG, brokerName=1568_6, queueId=0], queueOffset=0], index: 1, tag: tag-server-jVlHhwhvmlpkrLIqle6R
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  75] SendResult [sendStatus=SEND_OK, msgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A930002, offsetMsgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223A930002, messageQueue=MessageQueue [topic=topic_testNormalMessageReconsumeTime_bGVeiG, brokerName=1568_7, queueId=0], queueOffset=0], index: 2, tag: tag-server-jVlHhwhvmlpkrLIqle6R
2023-11-15 14:08:54 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 114] recv msg(first) MessageExt [brokerName=1568_7, queueId=0, storeSize=272, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534420, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='topic_testNormalMessageReconsumeTime_bGVeiG', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028534455, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223A930002, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[99, 98, 54, 54, 55, 52, 48, 55, 45, 48, 55, 48, 52, 45, 52, 98, 53, 100, 45, 98, 53, 52, 102, 45, 48, 48, 97, 102, 49, 48, 56, 51, 102, 53, 54, 98], transactionId='null'}] 
2023-11-15 14:08:54 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 114] recv msg(first) MessageExt [brokerName=1568_6, queueId=0, storeSize=272, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534386, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='topic_testNormalMessageReconsumeTime_bGVeiG', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028534457, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223A720001, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[100, 48, 100, 49, 57, 101, 97, 51, 45, 51, 52, 50, 100, 45, 52, 54, 48, 56, 45, 57, 55, 53, 53, 45, 53, 101, 54, 99, 53, 99, 100, 51, 49, 48, 102, 49], transactionId='null'}] 
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  75] SendResult [sendStatus=SEND_OK, msgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003, offsetMsgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003, messageQueue=MessageQueue [topic=topic_testNormalMessageReconsumeTime_bGVeiG, brokerName=1568_0, queueId=0], queueOffset=0], index: 3, tag: tag-server-jVlHhwhvmlpkrLIqle6R
2023-11-15 14:08:54 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 114] recv msg(first) MessageExt [brokerName=1568_0, queueId=0, storeSize=272, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534454, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='topic_testNormalMessageReconsumeTime_bGVeiG', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028534482, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[55, 102, 54, 57, 100, 53, 54, 48, 45, 56, 54, 50, 55, 45, 52, 51, 50, 100, 45, 98, 48, 100, 55, 45, 54, 57, 102, 100, 53, 52, 50, 102, 101, 52, 51, 50], transactionId='null'}] 
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  75] SendResult [sendStatus=SEND_OK, msgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223AD00004, offsetMsgId=240E03900E6E3C60000000000000068A0B1B251A69D74B223AD00004, messageQueue=MessageQueue [topic=topic_testNormalMessageReconsumeTime_bGVeiG, brokerName=1568_1, queueId=0], queueOffset=0], index: 4, tag: tag-server-jVlHhwhvmlpkrLIqle6R
2023-11-15 14:08:54 INFO  [o.a.r.c.r.RMQNormalProducer   #send                :  81] Producer send messages finished
2023-11-15 14:08:54 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 114] recv msg(first) MessageExt [brokerName=1568_1, queueId=0, storeSize=272, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534481, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message{topic='topic_testNormalMessageReconsumeTime_bGVeiG', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028534499, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223AD00004, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[50, 100, 51, 99, 97, 54, 101, 49, 45, 55, 98, 102, 100, 45, 52, 51, 99, 52, 45, 97, 102, 98, 50, 45, 49, 54, 102, 51, 52, 55, 99, 54, 97, 52, 56, 53], transactionId='null'}] 
2023-11-15 14:09:04 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 123] first retry time is: 10
2023-11-15 14:09:04 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 137] recv msg(reconsume later) MessageExt [brokerName=1569_0, queueId=0, storeSize=308, queueOffset=0, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534454, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000000, commitLogOffset=0, bodyCRC=0, reconsumeTimes=1, preparedTransactionOffset=0, toString()=Message{topic='%RETRY%GID_testNormalMessageReconsumeTime_DUURJv', flag=0, properties={MIN_OFFSET=0, MAX_OFFSET=0, CONSUME_START_TIME=1700028544592, TIMER_DELIVER_MS=1700028544483, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[55, 102, 54, 57, 100, 53, 54, 48, 45, 56, 54, 50, 55, 45, 52, 51, 50, 100, 45, 98, 48, 100, 55, 45, 54, 57, 102, 100, 53, 52, 50, 102, 101, 52, 51, 50], transactionId='null'}] 
2023-11-15 14:09:39 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 126] second retry time is: 30
2023-11-15 14:09:39 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 137] recv msg(reconsume later) MessageExt [brokerName=1569_0, queueId=0, storeSize=308, queueOffset=1, sysFlag=0, bornTimestamp=0, bornHost=/192.168.123.212:0, storeTimestamp=1700028534454, storeHost=/192.168.123.212:8081, msgId=C0A87BD400001F910000000000000001, commitLogOffset=1, bodyCRC=0, reconsumeTimes=2, preparedTransactionOffset=0, toString()=Message{topic='%RETRY%GID_testNormalMessageReconsumeTime_DUURJv', flag=0, properties={MIN_OFFSET=1, MAX_OFFSET=1, CONSUME_START_TIME=1700028574636, TIMER_DELIVER_MS=1700028574593, UNIQ_KEY=240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003, TAGS=tag-server-jVlHhwhvmlpkrLIqle6R}, body=[55, 102, 54, 57, 100, 53, 54, 48, 45, 56, 54, 50, 55, 45, 52, 51, 50, 100, 45, 98, 48, 100, 55, 45, 54, 57, 102, 100, 53, 52, 50, 102, 101, 52, 51, 50], transactionId='null'}] 
2023-11-15 14:09:39 INFO  [.a.r.s.a.PushConsumerRetryTest#consumeMessage      : 147] recv msgid(success) 240E03900E6E3C60000000000000068A0B1B251A69D74B223AB60003 
2023-11-15 14:10:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x18df981e, L:/127.0.0.1:51072 - R:/127.0.0.1:8081] INBOUND GO_AWAY: lastStreamId=2147483647 errorCode=0 length=8 bytes=6d61785f69646c65
2023-11-15 14:10:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x18df981e, L:/127.0.0.1:51072 - R:/127.0.0.1:8081] INBOUND PING: ack=false bytes=40715087873
2023-11-15 14:10:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x18df981e, L:/127.0.0.1:51072 - R:/127.0.0.1:8081] OUTBOUND PING: ack=true bytes=40715087873
2023-11-15 14:10:54 DEBUG [g.n.s.i.g.n.NettyClientHandler#log                 : 214] [id: 0x18df981e, L:/127.0.0.1:51072 - R:/127.0.0.1:8081] INBOUND GO_AWAY: lastStreamId=5 errorCode=0 length=8 bytes=6d61785f69646c65


java.util.concurrent.TimeoutException: testNormalMessageReconsumeTime() timed out after 180 seconds