Fix: SendAsync callback was not invoked when producer is in reconnecting

[nodece]

Fixes #1332

Motivation

When the producer reconnects, the batching/non-batching messages can not be added to the pending queue, so the timeout error can not be triggered. This leads to situations where SendAsync callbacks are never invoked when the producer is in a reconnecting state, potentially causing resource leaks and unresponsive applications.

Modifications

1. Refactored batch processing:
   • Moved batch processing logic to a dedicated goroutine to eliminate race conditions.
   • Proper synchronization was added for batch operations when calling the flush method.
2. Separated data writing from message processing, means the message should be added to the queue, and then write the message to the broker by the event loop.
3. Fixed payload size handling in TestProducerSendWithContext test:

   time="2025-03-16T09:05:13Z" level=error msg="Single message serialize failed %!s(<nil>)" error="encryptedPayload 
   exceeds MaxMessageSize, size: 1048607, MaxMessageSize: 1048576" producerID=1 producer_name=standalone-0- 
   358 topic="persistent://public/default/my-topic-857833034"
   

4. Added a comprehensive test case TestProducerKeepReconnectingAndThenCallSendAsync that verifies:
   • SendAsync callbacks are properly invoked with appropriate errors during reconnection.
   • The fix works correctly with both batching enabled and disabled configurations.

[nodece]

@gunli Could you have a chance to review?