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

• Separated internal data writing and network requests from the event loop.
  • Data flow: SendAsync -> dataChan -> writeChan -> broker.
• 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"
  

• 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?

[RobertIndie]

This PR introduces two separate message writing paths:

1. Non-batch message: SendAsync -> dataChan
2. Batch messages: SendAsync -> batchChan -> dataChan

Both paths are asynchronous, which breaks the message order guaranteed by SendAsync.

For example, if we send two messages using SendAsync like this:

producer.SendAsync(context.Background(), &ProducerMessage{
    Payload: []byte("A"),
}, func(_ MessageID, _ *ProducerMessage, err error) {
    errChan <- err
})

producer.SendAsync(context.Background(), &ProducerMessage{
    Payload: []byte("B"),
}, func(_ MessageID, _ *ProducerMessage, err error) {
    errChan <- err
})

We need to ensure the message order is A, B.

[nodece]

ping @RobertIndie

[RobertIndie]

Suggested change

	// send again
	testProducer.SendAsync(context.Background(), &ProducerMessage{
	Payload: []byte("test"),
	}, func(_ MessageID, _ *ProducerMessage, err error) {
	errChan <- err
	})
	for i := 0; i < 10; i++ {
	testProducer.SendAsync(context.Background(), &ProducerMessage{
	Payload: []byte("test"),
	}, func(id MessageID, producerMessage *ProducerMessage, err error) {
	fmt.Println("send async callback", id, producerMessage, err)
	})
	}
	// send again
	testProducer.SendAsync(context.Background(), &ProducerMessage{
	Payload: []byte("test"),
	}, func(_ MessageID, _ *ProducerMessage, err error) {
	errChan <- err
	})

You can eaisly reproduce the issue by sending some messages before this step. And also set the BatchingMaxMessages of the producer to a smaller value like 5.

[nodece]

Done.

[RobertIndie]

This doesn't address the root cause. If we send too many messages, the writeChan can still get stuck, and the issue persists. I can reproduce the problem by producing at least maxPendingMessages after the cluster is shut down.

[nodece]

@RobertIndie Once the internal queue reaches its configured capacity (maxPendingMessages), the producer should either block or reject incoming messages based on the DisableBlockIfQueueFull setting:

• If DisableBlockIfQueueFull is false, the producer should block until space becomes available.
• If DisableBlockIfQueueFull is true, the producer should fail fast by rejecting new messages.

Proper backpressure handling is essential to avoid unbounded memory usage and ensure the system remains responsive under load or during broker unavailability.

By default, DisableBlockIfQueueFull is set to false, which means the producer will block when the number of pending messages exceeds maxPendingMessages. If it is explicitly set to true, the producer will fail fast with a producer send queue is full error.

I think this topic is different from the current PR and might be better addressed separately. Let’s keep this PR focused on the specific issue it aims to resolve.

[nodece]

Closed by #1422