You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/12/07 06:39:32 UTC
[GitHub] [pulsar-client-go] Gleiphir2769 opened a new pull request, #903: [Issue 183] [Feat] Support cumulative acknowledge
Gleiphir2769 opened a new pull request, #903:
URL: https://github.com/apache/pulsar-client-go/pull/903
Master Issue: #183
### Motivation
Cumulative confirmation 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 interface `CumulativeAck` and `CumulativeAckID` for `Consumer`.
``` golang
// CumulativeAck the reception of all the messages in the stream up to (and including)
// the provided message.
CumulativeAck(msg Message) error
// CumulativeAckID the reception of all the messages in the stream up to (and including)
// the provided message, identified by its MessageID
CumulativeAckID(msgID MessageID) error
```
- Add the `CumulativeAck` and `CumulativeAckID` implementation for `consumer`, `multiTopicConsumer`, `regexConsumer` and `mockConsumer`.
- Add the unit test for cumulative ack in `consumer_test.go`. (not finished)
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Does this pull request potentially affect one of the following parts:
*If `yes` was chosen, please highlight the changes*
- The public API: yes
--
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
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
Gleiphir2769 commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1060259306
##########
pulsar/impl_message_test.go:
##########
@@ -80,6 +80,12 @@ func TestAckTracker(t *testing.T) {
}
assert.Equal(t, true, tracker.completed())
+
+ // test large number 1000 cumulative
+ tracker = newAckTracker(1000)
+
+ assert.Equal(t, true, tracker.ackCumulative(999))
Review Comment:
Good suggestion! I will add test code to cover the cumulative tracker.
--
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
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1058310298
##########
pulsar/impl_message_test.go:
##########
@@ -80,6 +80,12 @@ func TestAckTracker(t *testing.T) {
}
assert.Equal(t, true, tracker.completed())
+
+ // test large number 1000 cumulative
+ tracker = newAckTracker(1000)
+
+ assert.Equal(t, true, tracker.ackCumulative(999))
Review Comment:
It's better to add a test to verify `ackCumulative` for 0 to 998 returns false.
```golang
tracker = newAckTracker(1000)
for i := 0; i < 999; i++ {
assert.False(t, tracker.ackCumulative(i))
}
assert.False(t, tracker.completed())
```
--
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
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1057098391
##########
pulsar/consumer.go:
##########
@@ -231,6 +231,14 @@ type Consumer interface {
// AckID the consumption of a single message, identified by its MessageID
AckID(MessageID) error
+ // CumulativeAck the reception of all the messages in the stream up to (and including)
+ // the provided message.
+ CumulativeAck(msg Message) error
+
+ // CumulativeAckID the reception of all the messages in the stream up to (and including)
+ // the provided message, identified by its MessageID
+ CumulativeAckID(msgID MessageID) error
Review Comment:
Maybe it's better to use `AckIDCumulative` as the method name? Like the `acknowledge` and `acknowledgeCumulative` methods in other clients.
--
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
[GitHub] [pulsar-client-go] Gleiphir2769 commented on pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
Gleiphir2769 commented on PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#issuecomment-1342441329
Hi, @RobertIndie @nodece, could you give a review when you are free? Thanks.
--
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
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
Gleiphir2769 commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1057104466
##########
pulsar/consumer.go:
##########
@@ -231,6 +231,14 @@ type Consumer interface {
// AckID the consumption of a single message, identified by its MessageID
AckID(MessageID) error
+ // CumulativeAck the reception of all the messages in the stream up to (and including)
+ // the provided message.
+ CumulativeAck(msg Message) error
+
+ // CumulativeAckID the reception of all the messages in the stream up to (and including)
+ // the provided message, identified by its MessageID
+ CumulativeAckID(msgID MessageID) error
Review Comment:
Good suggestion. I will rename the two methods.
--
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
[GitHub] [pulsar-client-go] Gleiphir2769 commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
Gleiphir2769 commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1060264770
##########
pulsar/consumer_test.go:
##########
@@ -731,6 +732,161 @@ func TestConsumerAck(t *testing.T) {
}
}
+func TestConsumerNoBatchCumulativeAck(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ // disable batching
+ DisableBatching: true,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+
+ const N = 100
+
+ for i := 0; i < N; i++ {
+ if _, err := producer.Send(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i)),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ for i := 0; i < N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ if i == N/2-1 {
+ // cumulative acks the first half of messages
+ consumer.AckCumulative(msg)
+ }
+ }
+
+ consumer.Close()
+
+ // Subscribe again
+ consumer, err = client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ // We should only receive the 2nd half of messages
+ for i := N / 2; i < N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ consumer.Ack(msg)
+ }
+}
+
+func TestConsumerBatchCumulativeAck(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+
+ const N = 100
+
+ // send a batch
+ wg := sync.WaitGroup{}
+ for i := 0; i < N; i++ {
+ wg.Add(1)
+ producer.SendAsync(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+ func(id MessageID, producerMessage *ProducerMessage, e error) {
+ assert.NoError(t, e)
+ wg.Done()
+ })
+ }
+ wg.Wait()
+
+ err = producer.Flush()
+ assert.NoError(t, err)
+
+ // send another batch
+ wg = sync.WaitGroup{}
+ for i := N; i < 2*N; i++ {
+ wg.Add(1)
+ producer.SendAsync(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+ func(id MessageID, producerMessage *ProducerMessage, e error) {
+ assert.NoError(t, e)
+ wg.Done()
+ })
+ }
+ wg.Wait()
+
+ for i := 0; i < 2*N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ if i == N-1 {
+ // cumulative acks the first half of messages
+ consumer.AckCumulative(msg)
+ }
Review Comment:
> I think you can test the case i == N to cover the following case
You are right. I will modify `TestConsumerBatchCumulativeAck` to cover this case.
--
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
[GitHub] [pulsar-client-go] BewareMyPower commented on a diff in pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903#discussion_r1058327748
##########
pulsar/consumer_test.go:
##########
@@ -731,6 +732,161 @@ func TestConsumerAck(t *testing.T) {
}
}
+func TestConsumerNoBatchCumulativeAck(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ // disable batching
+ DisableBatching: true,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+
+ const N = 100
+
+ for i := 0; i < N; i++ {
+ if _, err := producer.Send(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i)),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ for i := 0; i < N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ if i == N/2-1 {
+ // cumulative acks the first half of messages
+ consumer.AckCumulative(msg)
+ }
+ }
+
+ consumer.Close()
+
+ // Subscribe again
+ consumer, err = client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+ defer consumer.Close()
+
+ // We should only receive the 2nd half of messages
+ for i := N / 2; i < N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ consumer.Ack(msg)
+ }
+}
+
+func TestConsumerBatchCumulativeAck(t *testing.T) {
+ client, err := NewClient(ClientOptions{
+ URL: lookupURL,
+ })
+
+ assert.Nil(t, err)
+ defer client.Close()
+
+ topicName := newTopicName()
+ ctx := context.Background()
+
+ producer, err := client.CreateProducer(ProducerOptions{
+ Topic: topicName,
+ })
+ assert.Nil(t, err)
+ defer producer.Close()
+
+ consumer, err := client.Subscribe(ConsumerOptions{
+ Topic: topicName,
+ SubscriptionName: "sub-1",
+ Type: Exclusive,
+ })
+ assert.Nil(t, err)
+
+ const N = 100
+
+ // send a batch
+ wg := sync.WaitGroup{}
+ for i := 0; i < N; i++ {
+ wg.Add(1)
+ producer.SendAsync(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+ func(id MessageID, producerMessage *ProducerMessage, e error) {
+ assert.NoError(t, e)
+ wg.Done()
+ })
+ }
+ wg.Wait()
+
+ err = producer.Flush()
+ assert.NoError(t, err)
+
+ // send another batch
+ wg = sync.WaitGroup{}
+ for i := N; i < 2*N; i++ {
+ wg.Add(1)
+ producer.SendAsync(ctx, &ProducerMessage{
+ Payload: []byte(fmt.Sprintf("msg-content-%d", i))},
+ func(id MessageID, producerMessage *ProducerMessage, e error) {
+ assert.NoError(t, e)
+ wg.Done()
+ })
+ }
+ wg.Wait()
+
+ for i := 0; i < 2*N; i++ {
+ msg, err := consumer.Receive(ctx)
+ assert.Nil(t, err)
+ assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
+
+ if i == N-1 {
+ // cumulative acks the first half of messages
+ consumer.AckCumulative(msg)
+ }
Review Comment:
I think you can test the case `i == N` to cover the following case
```
} else if !trackingID.tracker.hasPrevBatchAcked() {
// get previous batch message id
ackReq.msgID = trackingID.prev()
trackingID.tracker.setPrevBatchAcked()
```
Not sure if there is a way to reduce duplicated code. Maybe you can create another consumer to call `AckCumulative` when `i == N`?
```golang
if i == N - 1 {
consumer.AckCumulative(msg)
} else if i == N {
consumer2.AckCumulative(msg)
}
```
--
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
[GitHub] [pulsar-client-go] BewareMyPower merged pull request #903: [Issue 183] [Feat] Support cumulative acknowledge
Posted by GitBox <gi...@apache.org>.
BewareMyPower merged PR #903:
URL: https://github.com/apache/pulsar-client-go/pull/903
--
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