[Fix] Fix the key-based batch can't guarantee the ordering when flushing

[RobertIndie]

Motivation

When using key-based producing, the key-based batcher cannot guarantee the sequence ID order when flushing messages. This leads to frequent disconnections between the client and the broker, resulting in poor performance.

Here is the related log:

time="2024-07-19T03:38:07Z" level=warning msg="Received send error from server: [PersistenceError] : [Cannot determine whether the message is a duplicate at this time]"
time="2024-07-19T03:38:07Z" level=warning msg="Connection was closed" cnx="xxx" producerID=1 producer_name=xxx topic="xxx"
time="2024-07-19T03:38:07Z" level=warning msg="Failed to write on connection" error="use of closed network connection" local_addr="xxx" remote_addr="xxx"

The broker receives unordered messages and returns a SendError to the client. This causes the client to reconnect to the broker.

Modifications

• Refactor the BatchBuilder interface.
• Sort batches before flushing in the key-based batcher builder.

Verifying this change

This change added tests

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

• Dependencies (does it add or upgrade a dependency): (yes / no)
• The public API: (yes / no)
• The schema: (yes / no / don't know)
• The default values of configurations: (yes / no)
• The wire protocol: (yes / no)

Documentation

• Does this pull request introduce a new feature? (yes / no)
• If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation