[Issue 183] [Feat] Support cumulative acknowledge (#903)

Master Issue: #183 

### Motivation

Cumulative acknowledgement is a useful feature. Users can use this feature to ack messages in the stream up to (and including) provided message. 

Issue #183 shows more details.

### Modifications

- Add two api `AckCumulative` and `AckIDCumulative` for `Consumer`.

``` golang

// AckCumulative the reception of all the messages in the stream up to (and including)
// the provided message.
AckCumulative(msg Message) error

// AckIDCumulative the reception of all the messages in the stream up to (and including)
// the provided message, identified by its MessageID
AckIDCumulative(msgID MessageID) error

```

- Add the `AckCumulative` and `AckIDCumulative` implementation for `consumer`, `multiTopicConsumer`, `regexConsumer` and `mockConsumer`.

- Add the unit test `TestConsumerNoBatchCumulativeAck` `TestConsumerBatchCumulativeAck` `TestCumulativeAckWithResponse` for cumulative ack in `consumer_test.go`.

pulsar/consumer.go

@@ -234,6 +234,14 @@ type Consumer interface {
 	// AckID the consumption of a single message, identified by its MessageID
 	AckID(MessageID) error
 
+	// AckCumulative the reception of all the messages in the stream up to (and including)
+	// the provided message.
+	AckCumulative(msg Message) error
+
+	// AckIDCumulative the reception of all the messages in the stream up to (and including)
+	// the provided message, identified by its MessageID
+	AckIDCumulative(msgID MessageID) error
+
 	// ReconsumeLater mark a message for redelivery after custom delay
 	ReconsumeLater(msg Message, delay time.Duration)
 

pulsar/consumer_impl.go

@@ -38,6 +38,8 @@ type acker interface {
 	// AckID does not handle errors returned by the Broker side, so no need to wait for doneCh to finish.
 	AckID(id MessageID) error
 	AckIDWithResponse(id MessageID) error
+	AckIDCumulative(msgID MessageID) error
+	AckIDWithResponseCumulative(msgID MessageID) error
 	NackID(id MessageID)
 	NackMsg(msg Message)
 }
@@ -496,6 +498,26 @@ func (c *consumer) AckID(msgID MessageID) error {
 	return c.consumers[msgID.PartitionIdx()].AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *consumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *consumer) AckIDCumulative(msgID MessageID) error {
+	if err := c.checkMsgIDPartition(msgID); err != nil {
+		return err
+	}
+
+	if c.options.AckWithResponse {
+		return c.consumers[msgID.PartitionIdx()].AckIDWithResponseCumulative(msgID)
+	}
+
+	return c.consumers[msgID.PartitionIdx()].AckIDCumulative(msgID)
+}
+
 // ReconsumeLater mark a message for redelivery after custom delay
 func (c *consumer) ReconsumeLater(msg Message, delay time.Duration) {
 	if delay < 0 {

pulsar/consumer_multitopic.go

@@ -143,6 +143,33 @@ func (c *multiTopicConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message
+func (c *multiTopicConsumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *multiTopicConsumer) AckIDCumulative(msgID MessageID) error {
+	mid, ok := toTrackingMessageID(msgID)
+	if !ok {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return errors.New("invalid message id type in multi_consumer")
+	}
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("unable to ack message because consumer is nil")
+	}
+
+	if c.options.AckWithResponse {
+		return mid.consumer.AckIDWithResponseCumulative(msgID)
+	}
+
+	return mid.consumer.AckIDCumulative(msgID)
+}
+
 func (c *multiTopicConsumer) ReconsumeLater(msg Message, delay time.Duration) {
 	names, err := validateTopicNames(msg.Topic())
 	if err != nil {

pulsar/consumer_partition.go

@@ -451,6 +451,7 @@ func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
 
 	ackReq := new(ackRequest)
 	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
@@ -483,6 +484,7 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 
 	ackReq := new(ackRequest)
 	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = individualAck
 	if !trackingID.Undefined() && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
@@ -497,6 +499,62 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 	return nil
 }
 
+func (pc *partitionConsumer) AckIDCumulative(msgID MessageID) error {
+	return pc.internalAckIDCumulative(msgID, false)
+}
+
+func (pc *partitionConsumer) AckIDWithResponseCumulative(msgID MessageID) error {
+	return pc.internalAckIDCumulative(msgID, true)
+}
+
+func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withResponse bool) 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")
+	}
+
+	// chunk message id will be converted to tracking message id
+	trackingID, ok := toTrackingMessageID(msgID)
+	if !ok {
+		return errors.New("failed to convert trackingMessageID")
+	}
+	if trackingID.Undefined() {
+		return nil
+	}
+
+	ackReq := new(ackRequest)
+	ackReq.doneCh = make(chan struct{})
+	ackReq.ackType = cumulativeAck
+	if trackingID.ackCumulative() {
+		ackReq.msgID = trackingID
+	} else if !trackingID.tracker.hasPrevBatchAcked() {
+		// get previous batch message id
+		ackReq.msgID = trackingID.prev()
+		trackingID.tracker.setPrevBatchAcked()
+	} else {
+		// waiting for all the msgs are acked in this batch
+		return nil
+	}
+
+	pc.metrics.AcksCounter.Inc()
+	pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
+	// send ack request to eventsCh
+	pc.eventsCh <- ackReq
+
+	if withResponse {
+		// wait for the request to complete if withResponse set true
+		<-ackReq.doneCh
+	}
+
+	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
+
+	if cmid, ok := toChunkedMessageID(msgID); ok {
+		pc.unAckChunksTracker.remove(cmid)
+	}
+
+	return nil
+}
+
 func (pc *partitionConsumer) NackID(msgID MessageID) {
 	if cmid, ok := toChunkedMessageID(msgID); ok {
 		pc.unAckChunksTracker.nack(cmid)
@@ -711,7 +769,13 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 	cmdAck := &pb.CommandAck{
 		ConsumerId: proto.Uint64(pc.consumerID),
 		MessageId:  messageIDs,
-		AckType:    pb.CommandAck_Individual.Enum(),
+	}
+
+	switch req.ackType {
+	case individualAck:
+		cmdAck.AckType = pb.CommandAck_Individual.Enum()
+	case cumulativeAck:
+		cmdAck.AckType = pb.CommandAck_Cumulative.Enum()
 	}
 
 	if pc.options.ackWithResponse {
@@ -1189,10 +1253,16 @@ func (pc *partitionConsumer) dispatcher() {
 	}
 }
 
+const (
+	individualAck = iota
+	cumulativeAck
+)
+
 type ackRequest struct {
-	doneCh chan struct{}
-	msgID  trackingMessageID
-	err    error
+	doneCh  chan struct{}
+	msgID   trackingMessageID
+	ackType int
+	err     error
 }
 
 type unsubscribeRequest struct {

pulsar/consumer_regex.go

@@ -187,6 +187,33 @@ func (c *regexConsumer) AckID(msgID MessageID) error {
 	return mid.consumer.AckID(msgID)
 }
 
+// AckCumulative the reception of all the messages in the stream up to (and including)
+// the provided message.
+func (c *regexConsumer) AckCumulative(msg Message) error {
+	return c.AckIDCumulative(msg.ID())
+}
+
+// AckIDCumulative the reception of all the messages in the stream up to (and including)
+// the provided message, identified by its MessageID
+func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
+	mid, ok := toTrackingMessageID(msgID)
+	if !ok {
+		c.log.Warnf("invalid message id type %T", msgID)
+		return errors.New("invalid message id type")
+	}
+
+	if mid.consumer == nil {
+		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
+		return errors.New("unable to ack message because consumer is nil")
+	}
+
+	if c.options.AckWithResponse {
+		return mid.consumer.AckIDWithResponseCumulative(msgID)
+	}
+
+	return mid.consumer.AckIDCumulative(msgID)
+}
+
 func (c *regexConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()