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

Author: nodece

pulsar/producer_partition.go

@@ -110,6 +110,7 @@ type partitionProducer struct {
 	compressionProvider      compression.Provider
 
 	// Channel where app is posting messages to be published
+	writeChan            chan *pendingItem
 	dataChan             chan *sendRequest
 	cmdChan              chan interface{}
 	connectClosedCh      chan *connectionClosed
@@ -184,6 +185,7 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 		options:          options,
 		producerID:       client.rpcClient.NewProducerID(),
 		dataChan:         make(chan *sendRequest, maxPendingMessages),
+		writeChan:        make(chan *pendingItem),
 		cmdChan:          make(chan interface{}, 10),
 		connectClosedCh:  make(chan *connectionClosed, 1),
 		batchFlushTicker: time.NewTicker(batchingMaxPublishDelay),
@@ -559,31 +561,42 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 }
 
 func (p *partitionProducer) runEventsLoop() {
+	go func() {
+		for {
+			select {
+			case data, ok := <-p.dataChan:
+				if !ok {
+					return
+				}
+				p.internalSend(data)
+			case <-p.batchFlushTicker.C:
+				p.internalFlushCurrentBatch()
+			case cmd, ok := <-p.cmdChan:
+				// when doClose() is call, p.dataChan will be closed, cmd will be nil
+				if !ok {
+					return
+				}
+				switch v := cmd.(type) {
+				case *flushRequest:
+					p.internalFlush(v)
+				case *closeProducer:
+					p.internalClose(v)
+					return
+				}
+			}
+		}
+	}()
+
 	for {
 		select {
-		case data, ok := <-p.dataChan:
-			// when doClose() is call, p.dataChan will be closed, data will be nil
+		case pi, ok := <-p.writeChan:
 			if !ok {
 				return
 			}
-			p.internalSend(data)
-		case cmd, ok := <-p.cmdChan:
-			// when doClose() is call, p.dataChan will be closed, cmd will be nil
-			if !ok {
-				return
-			}
-			switch v := cmd.(type) {
-			case *flushRequest:
-				p.internalFlush(v)
-			case *closeProducer:
-				p.internalClose(v)
-				return
-			}
+			p._getConn().WriteData(pi.ctx, pi.buffer)
 		case connectionClosed := <-p.connectClosedCh:
 			p.log.Info("runEventsLoop will reconnect in producer")
 			p.reconnectToBroker(connectionClosed)
-		case <-p.batchFlushTicker.C:
-			p.internalFlushCurrentBatch()
 		}
 	}
 }
@@ -898,16 +911,17 @@ func (p *partitionProducer) writeData(buffer internal.Buffer, sequenceID uint64,
 	default:
 		now := time.Now()
 		ctx, cancel := context.WithCancel(context.Background())
-		p.pendingQueue.Put(&pendingItem{
+		item := pendingItem{
 			ctx:          ctx,
 			cancel:       cancel,
 			createdAt:    now,
 			sentAt:       now,
 			buffer:       buffer,
 			sequenceID:   sequenceID,
 			sendRequests: callbacks,
-		})
-		p._getConn().WriteData(ctx, buffer)
+		}
+		p.pendingQueue.Put(&item)
+		p.writeChan <- &item
 	}
 }
 
@@ -1443,6 +1457,7 @@ func (p *partitionProducer) doClose(reason error) {
 
 	p.log.Info("Closing producer")
 	defer close(p.dataChan)
+	defer close(p.writeChan)
 	defer close(p.cmdChan)
 
 	id := p.client.rpcClient.NewRequestID()

pulsar/producer_test.go

@@ -2345,7 +2345,7 @@ func TestProducerSendWithContext(t *testing.T) {
 	// Make ctx be canceled to invalidate the context immediately
 	cancel()
 	_, err = producer.Send(ctx, &ProducerMessage{
-		Payload: make([]byte, 1024*1024),
+		Payload: make([]byte, 1024),
 	})
 	//  producer.Send should fail and return err context.Canceled
 	assert.True(t, errors.Is(err, context.Canceled))
@@ -2605,3 +2605,82 @@ func TestSelectConnectionForSameProducer(t *testing.T) {
 
 	client.Close()
 }
+
+func TestProducerKeepReconnectingAndThenCallSendAsync(t *testing.T) {
+	testProducerKeepReconnectingAndThenCallSendAsync(t, false)
+	testProducerKeepReconnectingAndThenCallSendAsync(t, true)
+}
+
+func testProducerKeepReconnectingAndThenCallSendAsync(t *testing.T, isEnabledBatching bool) {
+	t.Helper()
+
+	req := testcontainers.ContainerRequest{
+		Image:        getPulsarTestImage(),
+		ExposedPorts: []string{"6650/tcp", "8080/tcp"},
+		WaitingFor:   wait.ForExposedPort(),
+		Cmd:          []string{"bin/pulsar", "standalone", "-nfw"},
+	}
+	c, err := testcontainers.GenericContainer(context.Background(), testcontainers.GenericContainerRequest{
+		ContainerRequest: req,
+		Started:          true,
+	})
+	require.NoError(t, err, "Failed to start the pulsar container")
+	defer c.Terminate(context.Background())
+
+	endpoint, err := c.PortEndpoint(context.Background(), "6650", "pulsar")
+	require.NoError(t, err, "Failed to get the pulsar endpoint")
+
+	client, err := NewClient(ClientOptions{
+		URL:               endpoint,
+		ConnectionTimeout: 5 * time.Second,
+		OperationTimeout:  5 * time.Second,
+	})
+	require.NoError(t, err)
+	defer client.Close()
+
+	var testProducer Producer
+	require.Eventually(t, func() bool {
+		testProducer, err = client.CreateProducer(ProducerOptions{
+			Topic:           newTopicName(),
+			Schema:          NewBytesSchema(nil),
+			SendTimeout:     3 * time.Second,
+			DisableBatching: isEnabledBatching,
+		})
+		return err == nil
+	}, 30*time.Second, 1*time.Second)
+
+	// send a message
+	errChan := make(chan error)
+	defer close(errChan)
+
+	testProducer.SendAsync(context.Background(), &ProducerMessage{
+		Payload: []byte("test"),
+	}, func(_ MessageID, _ *ProducerMessage, err error) {
+		errChan <- err
+	})
+	select {
+	case <-time.After(10 * time.Second):
+		t.Fatal("test timeout")
+	case err := <-errChan:
+		require.NoError(t, err)
+	}
+
+	// stop pulsar server
+	timeout := 10 * time.Second
+	err = c.Stop(context.Background(), &timeout)
+	require.NoError(t, err)
+
+	// send again
+	testProducer.SendAsync(context.Background(), &ProducerMessage{
+		Payload: []byte("test"),
+	}, func(_ MessageID, _ *ProducerMessage, err error) {
+		errChan <- err
+	})
+	select {
+	case <-time.After(10 * time.Second):
+		t.Fatal("test timeout")
+	case err = <-errChan:
+		// should get a timeout error
+		require.ErrorIs(t, err, ErrSendTimeout)
+	}
+}