feat(proxy): optimize forward message to another node

[ShadowySpirits]

Test by reassigning queue 0 from node 2 to node 1 while sending messages at a rate of 500 tps with two producers.

Timeline:

• 15:37:24.500: start to reassign queue 0 from node 2 to node 1
• 15:37:27.488: close queue in node 2
• 15:37:27.901: open queue in node 1
• 15:37:34: all messages are sent successfully after retrying

Server log:

docker-broker-2-1  | 2023-12-11 15:37:24.500 INFO  Node[node-id=1] starts to yield topic-id=1 queue-id=0 to node[node-id=2]
docker-broker-2-1  | 2023-12-11 15:37:24.504 INFO  Invoke DataStore to close queue[topic-id=1, queue-id=0]
docker-broker-2-1  | 2023-12-11 15:37:24.505 INFO  [DefaultLogicQueueManager]: Close logic queue: 1 queue: 0
docker-broker-2-1  | 2023-12-11 15:37:27.404 INFO  commitStreamSetObject with StreamSetObject=[broker_id: 1 object_id: -1 sequence_id: -1 sub_streams { }], streamObjects=[stream_id: 1 start_offset: 110674 end_offset: 148797 object_id: 4 object_size: 50667535], compactedObjects=[]
docker-broker-2-1  | 2023-12-11 15:37:27.423 INFO  Extend stream range[stream-id=1, range-id=2] with segment [110674, 148797)
docker-broker-2-1  | 2023-12-11 15:37:27.459 INFO  broker[broke-id=1] commit StreamSet object[object-id=-1] success, compacted objects[[]], stream objects[[stream_id: 1
docker-broker-2-1  | start_offset: 110674
docker-broker-2-1  | end_offset: 148797
docker-broker-2-1  | object_id: 4
docker-broker-2-1  | object_size: 50667535
docker-broker-2-1  | ]]
docker-broker-2-1  | 2023-12-11 15:37:27.460 INFO  Upload delta WAL CommitStreamSetObjectRequest{objectId=-1, orderId=-1, objectSize=0, streamRanges=null, streamObjects=[StreamObject{objectId=4, objectSize=50667535, streamId=1, startOffset=110674, endOffset=148797, sourceObjectIds=null}], compactedObjectIds=null}, cost 2944ms, rate limiter 2.09716128E7bytes/s
docker-broker-2-1  | 2023-12-11 15:37:27.462 INFO  try trim WAL to 362373120
docker-broker-2-1  | 2023-12-11 15:37:27.476 INFO  [Stream id=1 epoch=2] closed
docker-broker-2-1  | 2023-12-11 15:37:27.482 INFO  [Stream id=3 epoch=2] closed
docker-broker-2-1  | 2023-12-11 15:37:27.488 INFO  Notify controller leader that DataStore has already closed queue[topic-id=1, queue-id=0]
docker-broker-2-1  | 2023-12-11 15:37:27.493 INFO  [Stream id=2 epoch=1] closed
docker-broker-2-1  | 2023-12-11 15:37:27.500 INFO  Update status of queue assignment and stream since all its belonging streams are closed, having topic-id=1, queue-id=0
docker-broker-1-1  | 2023-12-11 15:37:27.772 INFO  [DefaultLogicQueueManager]: Create and open logic queue success: topic: 1 queue: 0
docker-broker-2-1  | 2023-12-11 15:37:27.836 INFO  Node[node-id=1] opens stream [stream-id=3] with epoch=3
docker-broker-2-1  | 2023-12-11 15:37:27.841 INFO  Node[node-id=1] opens stream [stream-id=2] with epoch=2
docker-broker-2-1  | 2023-12-11 15:37:27.850 INFO  Node[node-id=1] opens stream [stream-id=1] with epoch=3
docker-broker-1-1  | 2023-12-11 15:37:27.875 INFO  Open Stream[stream-id=3, epoch=2] returns metadata: stream_id: 3
docker-broker-1-1  | epoch: 3
docker-broker-1-1  | range_id: 3
docker-broker-1-1  | state: OPEN
docker-broker-1-1  | 
docker-broker-1-1  | 2023-12-11 15:37:27.878 INFO  Stream 3 opened
docker-broker-1-1  | 2023-12-11 15:37:27.881 INFO  Open Stream[stream-id=2, epoch=1] returns metadata: stream_id: 2
docker-broker-1-1  | epoch: 2
docker-broker-1-1  | range_id: 2
docker-broker-1-1  | state: OPEN
docker-broker-1-1  | 
docker-broker-1-1  | 2023-12-11 15:37:27.883 INFO  Stream 2 opened
docker-broker-1-1  | 2023-12-11 15:37:27.900 INFO  Open Stream[stream-id=1, epoch=2] returns metadata: stream_id: 1
docker-broker-1-1  | epoch: 3
docker-broker-1-1  | range_id: 3
docker-broker-1-1  | start_offset: 110674
docker-broker-1-1  | end_offset: 148797
docker-broker-1-1  | state: OPEN
docker-broker-1-1  | 
docker-broker-1-1  | 2023-12-11 15:37:27.901 INFO  Stream 1 opened
docker-broker-2-1  | 2023-12-11 15:37:30.536 INFO  Controller leader has completed assignment/stream status update for topic-id=1, queue-id=0

Client log:

$ bin/mqadmin -e localhost:8181 produceMessage -i 1 -p 2 -r 500 -d 120
[RocketmqClientCallbackWorker-0-40] INFO org.apache.rocketmq.client.java.impl.producer.ProducerImpl - Resend message successfully, topic=Benchmark_Topic_0, messageId(s)=[01E2EBD4D46CE29CDC0588563700002296], maxAttempts=3, attempt=2, endpoints=ipv4:192.168.123.175:8182, clientId=SSpiritss-MacBook-Pro.local@40156@0@5gw9xvhhr9
[RocketmqClientCallbackWorker-0-51] INFO org.apache.rocketmq.client.java.impl.producer.ProducerImpl - Resend message successfully, topic=Benchmark_Topic_0, messageId(s)=[01E2EBD4D46CE29CDC0588563700002298], maxAttempts=3, attempt=2, endpoints=ipv4:192.168.123.175:8182, clientId=SSpiritss-MacBook-Pro.local@40156@0@5gw9xvhhr9
[RocketmqClientCallbackWorker-0-52] INFO org.apache.rocketmq.client.java.impl.producer.ProducerImpl - Resend message successfully, topic=Benchmark_Topic_0, messageId(s)=[01E2EBD4D46CE29CDC058856370000228E], maxAttempts=3, attempt=2, endpoints=ipv4:192.168.123.175:8182, clientId=SSpiritss-MacBook-Pro.local@40156@0@5gw9xvhhr9
11/12/2023, 3:37:34 pm =========================================================

-- Counters --------------------------------------------------------------------
Counter for sending messages failed
             count = 0

-- Timers ----------------------------------------------------------------------
Timer for sending messages
             count = 12007
         mean rate = 488.69 calls/second

[codecov]

Codecov Report

Merging #821 (340e844) into main (1065f55) will decrease coverage by 0.09%.
Report is 1 commits behind head on main.
The diff coverage is 10.34%.

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #821      +/-   ##
============================================
- Coverage     56.25%   56.17%   -0.09%     
+ Complexity     1261     1256       -5     
============================================
  Files           132      132              
  Lines          8516     8523       +7     
  Branches        784      785       +1     
============================================
- Hits           4791     4788       -3     
- Misses         3294     3301       +7     
- Partials        431      434       +3     

Components	Coverage Δ
RocketMQ Broker	∅ <ø> (∅)
RocketMQ Common	5.80% <ø> (ø)
RocketMQ Controller	57.39% <0.00%> (-0.19%)	⬇️
RocketMQ Proxy	37.55% <15.00%> (-0.12%)	⬇️
RocketMQ Store	74.33% <ø> (ø)
RocketMQ Stream	∅ <ø> (∅)
RocketMQ Metadata	35.44% <ø> (ø)

Files	Coverage Δ
.../rocketmq/proxy/service/TopicRouteServiceImpl.java	78.16% <0.00%> (-3.45%)	⬇️
.../controller/server/store/DefaultMetadataStore.java	63.13% <0.00%> (-1.40%)	⬇️
...omq/rocketmq/proxy/service/MessageServiceImpl.java	51.52% <15.78%> (+0.20%)	⬆️