Introduce doneCh for ack error

pulsar/consumer_impl.go

@@ -36,7 +36,9 @@ import (
 const defaultNackRedeliveryDelay = 1 * time.Minute
 
 type acker interface {
+	// AckID does not handle errors returned by the Broker side, so no need to wait for doneCh to finish.
 	AckID(id trackingMessageID) error
+	AckIDWithResponse(id trackingMessageID) error
 	NackID(id trackingMessageID)
 	NackMsg(msg Message)
 }
@@ -462,6 +464,10 @@ func (c *consumer) AckID(msgID MessageID) error {
 		return mid.Ack()
 	}
 
+	if c.options.AckWithResponse {
+		return c.consumers[mid.partitionIdx].AckIDWithResponse(mid)
+	}
+
 	return c.consumers[mid.partitionIdx].AckID(mid)
 }
 

pulsar/consumer_multitopic.go

@@ -136,6 +136,10 @@ func (c *multiTopicConsumer) AckID(msgID MessageID) error {
 		return errors.New("unable to ack message because consumer is nil")
 	}
 
+	if c.options.AckWithResponse {
+		return mid.AckWithResponse()
+	}
+
 	return mid.Ack()
 }
 

pulsar/consumer_partition.go

@@ -366,19 +366,44 @@ func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error
 	return convertToMessageID(id), nil
 }
 
+func (pc *partitionConsumer) AckIDWithResponse(msgID trackingMessageID) error {
+	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
+		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
+		return errors.New("consumer state is closed")
+	}
+
+	ackReq := new(ackRequest)
+	ackReq.doneCh = make(chan struct{})
+	if !msgID.Undefined() && msgID.ack() {
+		pc.metrics.AcksCounter.Inc()
+		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-msgID.receivedTime.UnixNano()) / 1.0e9)
+		ackReq.msgID = msgID
+		// send ack request to eventsCh
+		pc.eventsCh <- ackReq
+		// wait for the request to complete
+		<-ackReq.doneCh
+
+		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
+	}
+
+	return ackReq.err
+}
+
 func (pc *partitionConsumer) AckID(msgID trackingMessageID) error {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
 		return errors.New("consumer state is closed")
 	}
 
 	ackReq := new(ackRequest)
+	ackReq.doneCh = make(chan struct{})
 	if !msgID.Undefined() && msgID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-msgID.receivedTime.UnixNano()) / 1.0e9)
 		ackReq.msgID = msgID
 		// send ack request to eventsCh
 		pc.eventsCh <- ackReq
+		// No need to wait for ackReq.doneCh to finish
 
 		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
 	}
@@ -562,6 +587,7 @@ func (pc *partitionConsumer) clearMessageChannels() {
 }
 
 func (pc *partitionConsumer) internalAck(req *ackRequest) {
+	defer close(req.doneCh)
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to ack by closing or closed consumer")
 		return
@@ -986,8 +1012,9 @@ func (pc *partitionConsumer) dispatcher() {
 }
 
 type ackRequest struct {
-	msgID trackingMessageID
-	err   error
+	doneCh chan struct{}
+	msgID  trackingMessageID
+	err    error
 }
 
 type unsubscribeRequest struct {

pulsar/consumer_regex.go

@@ -180,6 +180,10 @@ func (c *regexConsumer) AckID(msgID MessageID) error {
 		return errors.New("consumer is nil in consumer_regex")
 	}
 
+	if c.options.AckWithResponse {
+		return mid.AckWithResponse()
+	}
+
 	return mid.Ack()
 }
 

pulsar/impl_message.go

@@ -75,6 +75,17 @@ func (id trackingMessageID) Ack() error {
 	return nil
 }
 
+func (id trackingMessageID) AckWithResponse() error {
+	if id.consumer == nil {
+		return errors.New("consumer is nil in trackingMessageID")
+	}
+	if id.ack() {
+		return id.consumer.AckIDWithResponse(id)
+	}
+
+	return nil
+}
+
 func (id trackingMessageID) Nack() {
 	if id.consumer == nil {
 		return