[GitHub] [pulsar-client-go] gunli opened a new issue, #1042: [Bug][Producer]Forget to consume the sendRequests before closing the producer

["gunli (via GitHub)" <gi...@apache.org>]

gunli opened a new issue, #1042:
URL: https://github.com/apache/pulsar-client-go/issues/1042

   #### Expected behavior
   
   Consume the sendRequests before closing the producer, send and flush them, invoke the callbacks of the input messages, so that the application can know the producing is succeed or failed.
   
   #### Actual behavior
   
   Currently, when we close the producer, we forget to consume the sendRequests in partitionProducer.dataChan before closing.
   In the case that producing is faster than consuming in partitionProducer, we have send a lot of sendRequests  into partitionProducer.dataChan, when closing, many of them are not consumed, these sendRequests will looks like get lost, their callback won't get invoked, the application won't know the producing result of these messages.
   
   #### Steps to reproduce
   
   Review the code of 
   ``` go
   func (p *partitionProducer) internalSendAsync(ctx context.Context, msg *ProducerMessage,
   	callback func(MessageID, *ProducerMessage, error), flushImmediately bool) {
   	//Register transaction operation to transaction and the transaction coordinator.
   	var newCallback func(MessageID, *ProducerMessage, error)
   	if msg.Transaction != nil {
   		transactionImpl := (msg.Transaction).(*transaction)
   		if transactionImpl.state != TxnOpen {
   			p.log.WithField("state", transactionImpl.state).Error("Failed to send message" +
   				" by a non-open transaction.")
   			callback(nil, msg, newError(InvalidStatus, "Failed to send message by a non-open transaction."))
   			return
   		}
   
   		if err := transactionImpl.registerProducerTopic(p.topic); err != nil {
   			callback(nil, msg, err)
   			return
   		}
   		if err := transactionImpl.registerSendOrAckOp(); err != nil {
   			callback(nil, msg, err)
   		}
   		newCallback = func(id MessageID, producerMessage *ProducerMessage, err error) {
   			callback(id, producerMessage, err)
   			transactionImpl.endSendOrAckOp(err)
   		}
   	} else {
   		newCallback = callback
   	}
   	if p.getProducerState() != producerReady {
   		// Producer is closing
   		newCallback(nil, msg, errProducerClosed)
   		return
   	}
   
   	// bc only works when DisableBlockIfQueueFull is false
   	bc := make(chan struct{})
   
   	// callbackOnce make sure the callback is only invoked once in chunking
   	callbackOnce := &sync.Once{}
   	var txn *transaction
   	if msg.Transaction != nil {
   		txn = (msg.Transaction).(*transaction)
   	}
   	sr := &sendRequest{
   		ctx:              ctx,
   		msg:              msg,
   		callback:         newCallback,
   		callbackOnce:     callbackOnce,
   		flushImmediately: flushImmediately,
   		publishTime:      time.Now(),
   		blockCh:          bc,
   		closeBlockChOnce: &sync.Once{},
   		transaction:      txn,
   	}
   	p.options.Interceptors.BeforeSend(p, msg)
   
   	p.dataChan <- sr
   
   	if !p.options.DisableBlockIfQueueFull {
   		// block if queue full
   		<-bc
   	}
   }
   
   func (p *partitionProducer) internalClose(req *closeProducer) {
   	defer close(req.doneCh)
   	if !p.casProducerState(producerReady, producerClosing) {
   		return
   	}
   
   	p.log.Info("Closing producer")
   
   	id := p.client.rpcClient.NewRequestID()
   	_, err := p.client.rpcClient.RequestOnCnx(p._getConn(), id, pb.BaseCommand_CLOSE_PRODUCER, &pb.CommandCloseProducer{
   		ProducerId: &p.producerID,
   		RequestId:  &id,
   	})
   
   	if err != nil {
   		p.log.WithError(err).Warn("Failed to close producer")
   	} else {
   		p.log.Info("Closed producer")
   	}
   
   	if p.batchBuilder != nil {
   		if err = p.batchBuilder.Close(); err != nil {
   			p.log.WithError(err).Warn("Failed to close batch builder")
   		}
   	}
   
   	p.setProducerState(producerClosed)
   	p._getConn().UnregisterListener(p.producerID)
   	p.batchFlushTicker.Stop()
   }
   ```
   
   #### System configuration
   **Pulsar version**: x.y
   
   @zengguan @merlimat @wolfstudy 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [pulsar-client-go] tisonkun closed issue #1042: [Bug][Producer]Forget to consume the sendRequests before closing the producer

["tisonkun (via GitHub)" <gi...@apache.org>]

tisonkun closed issue #1042: [Bug][Producer]Forget to consume the sendRequests before closing the producer
URL: https://github.com/apache/pulsar-client-go/issues/1042


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org