feat(cluster): support ack stream between multiple connection

[zhang2014]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

feat(cluster): support ack stream between multiple connection

Tests

• [x] Unit Test
• [x] Logic Test
• [ ] Benchmark Test
• [ ] No Test - Explain why

Type of change

• [x] Bug Fix (non-breaking change which fixes an issue)
• [x] New Feature (non-breaking change which adds functionality)
• [ ] Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• [ ] Documentation Update
• [ ] Refactoring
• [ ] Performance Improvement
• [ ] Other (please describe):

---

This change is