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

Author: nodece

pulsar/producer_partition.go

@@ -110,7 +110,7 @@ type partitionProducer struct {
 	compressionProvider      compression.Provider
 
 	// Channel where app is posting messages to be published
-	dataChan             chan *sendRequest
+	dataChan             chan *pendingItem
 	cmdChan              chan interface{}
 	connectClosedCh      chan *connectionClosed
 	publishSemaphore     internal.Semaphore
@@ -177,16 +177,15 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	p := &partitionProducer{
-		client:           client,
-		topic:            topic,
-		log:              logger,
-		cnxKeySuffix:     client.cnxPool.GenerateRoundRobinIndex(),
-		options:          options,
-		producerID:       client.rpcClient.NewProducerID(),
-		dataChan:         make(chan *sendRequest, maxPendingMessages),
-		cmdChan:          make(chan interface{}, 10),
-		connectClosedCh:  make(chan *connectionClosed, 1),
-		batchFlushTicker: time.NewTicker(batchingMaxPublishDelay),
+		client:          client,
+		topic:           topic,
+		log:             logger,
+		cnxKeySuffix:    client.cnxPool.GenerateRoundRobinIndex(),
+		options:         options,
+		producerID:      client.rpcClient.NewProducerID(),
+		dataChan:        make(chan *pendingItem, maxPendingMessages),
+		cmdChan:         make(chan interface{}, 10),
+		connectClosedCh: make(chan *connectionClosed, 1),
 		compressionProvider: internal.GetCompressionProvider(pb.CompressionType(options.CompressionType),
 			compression.Level(options.CompressionLevel)),
 		publishSemaphore:  internal.NewSemaphore(int32(maxPendingMessages)),
@@ -200,9 +199,6 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 		cancelFunc:        cancelFunc,
 		backOffPolicyFunc: boFunc,
 	}
-	if p.options.DisableBatching {
-		p.batchFlushTicker.Stop()
-	}
 	p.setProducerState(producerInit)
 
 	if options.Schema != nil && options.Schema.GetSchemaInfo() != nil {
@@ -219,7 +215,6 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 	}
 	err := p.grabCnx("")
 	if err != nil {
-		p.batchFlushTicker.Stop()
 		logger.WithError(err).Error("Failed to create producer at newPartitionProducer")
 		return nil, err
 	}
@@ -232,6 +227,11 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 	p.log.WithField("cnx", p._getConn().ID()).Info("Created producer")
 	p.setProducerState(producerReady)
 
+	if !p.options.DisableBatching {
+		p.batchFlushTicker = time.NewTicker(batchingMaxPublishDelay)
+		go p.listenBatch()
+	}
+
 	if p.options.SendTimeout > 0 {
 		go p.failTimeoutMessages()
 	}
@@ -558,6 +558,15 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 	})
 }
 
+func (p *partitionProducer) listenBatch() {
+	for {
+		select {
+		case <-p.batchFlushTicker.C:
+			p.internalFlushCurrentBatch()
+		}
+	}
+}
+
 func (p *partitionProducer) runEventsLoop() {
 	for {
 		select {
@@ -566,7 +575,7 @@ func (p *partitionProducer) runEventsLoop() {
 			if !ok {
 				return
 			}
-			p.internalSend(data)
+			p.internalWriteData(data)
 		case cmd, ok := <-p.cmdChan:
 			// when doClose() is call, p.dataChan will be closed, cmd will be nil
 			if !ok {
@@ -582,8 +591,6 @@ func (p *partitionProducer) runEventsLoop() {
 		case connectionClosed := <-p.connectClosedCh:
 			p.log.Info("runEventsLoop will reconnect in producer")
 			p.reconnectToBroker(connectionClosed)
-		case <-p.batchFlushTicker.C:
-			p.internalFlushCurrentBatch()
 		}
 	}
 }
@@ -688,6 +695,10 @@ func (p *partitionProducer) internalSend(sr *sendRequest) {
 	}
 }
 
+func (p *partitionProducer) internalWriteData(item *pendingItem) {
+	p._getConn().WriteData(item.ctx, item.buffer)
+}
+
 func addRequestToBatch(smm *pb.SingleMessageMetadata, p *partitionProducer,
 	uncompressedPayload []byte,
 	request *sendRequest, msg *ProducerMessage, deliverAt time.Time,
@@ -898,16 +909,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.dataChan <- &item
 	}
 }
 
@@ -1077,7 +1089,7 @@ func (p *partitionProducer) clearPendingSendRequests() {
 	for i := 0; i < sizeBeforeFlushing; i++ {
 		select {
 		case pendingData := <-p.dataChan:
-			p.internalSend(pendingData)
+			p.internalWriteData(pendingData)
 
 		default:
 			return
@@ -1352,7 +1364,7 @@ func (p *partitionProducer) internalSendAsync(
 		return
 	}
 
-	p.dataChan <- sr
+	p.internalSend(sr)
 }
 
 func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt) {
@@ -1466,7 +1478,10 @@ func (p *partitionProducer) doClose(reason error) {
 
 	p.setProducerState(producerClosed)
 	p._getConn().UnregisterListener(p.producerID)
-	p.batchFlushTicker.Stop()
+
+	if p.batchFlushTicker != nil {
+		p.batchFlushTicker.Stop()
+	}
 }
 
 func (p *partitionProducer) failPendingMessages(err error) {

pulsar/producer_test.go

@@ -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 <-errChan:
+		// fine
+	}
+
+	// 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.True(t, errors.Is(err, ErrSendTimeout))
+	}
+}