Repository: apache/pulsar-client-go (go)

Issue:
Support consumer event listener
**Describe the solution you'd like**
It would be awesome if the go client supported the consumer event listener, similar to the [java client](https://pulsar.apache.org/api/client/2.10.x/org/apache/pulsar/client/api/ConsumerEventListener.html).

**Is your feature request related to a problem? Please describe.**
We have an application that uses a partitioned topic. On application scale out, we need to know which consumers have been unassigned from a topic/partition in order to clear the internal memory of our application for this partition.

**Describe alternatives you've considered**
We have explored a `Key_Shared` subscription type with a non-partitioned topic. When a scale out of the application occurs with this type of subscription unacked messages are not automatically re-delivered and message consumption is paused until the pending messages are acked/nacked. See apache/pulsar#15705. This behaviour prevents our application from using a `Key_Shared` subscription type.

**Additional context**
- [slack thread](https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1669690833804909)
- [related stackoverflow question](https://stackoverflow.com/questions/74620723/group-messages-by-key-for-pulsar-key-shared-subscription-type)


Pull request description:
[feat] support for consumer event listener




Fixes #901


### Motivation

Support consumer active events.

### Modifications

Add an option to consumers, add callback on `BaseCommand_ACTIVE_CONSUMER_CHANGE`.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation


Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index ffef786..0a515e0 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -120,6 +120,9 @@ type ConsumerOptions struct {
 	// Default is `Latest`
 	SubscriptionInitialPosition
 
+	// EventListener will be called when active consumer changed (in failover subscription type)
+	EventListener ConsumerEventListener
+
 	// DLQ represents the configuration for Dead Letter Queue consumer policy.
 	// eg. route the message to topic X after N failed attempts at processing it
 	// By default is nil and there's no DLQ
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index a402a4d..707c13c 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -376,6 +376,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				maxPendingChunkedMessage:    c.options.MaxPendingChunkedMessage,
 				expireTimeOfIncompleteChunk: c.options.ExpireTimeOfIncompleteChunk,
 				autoAckIncompleteChunk:      c.options.AutoAckIncompleteChunk,
+				consumerEventListener:       c.options.EventListener,
 			}
 			cons, err := newPartitionConsumer(c, c.client, opts, c.messageCh, c.dlq, c.metrics)
 			ch <- ConsumerError{
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 4149a91..46f13d8 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -112,6 +112,13 @@ type partitionConsumerOpts struct {
 	maxPendingChunkedMessage    int
 	expireTimeOfIncompleteChunk time.Duration
 	autoAckIncompleteChunk      bool
+	// in failover mode, this callback will be called when consumer change
+	consumerEventListener ConsumerEventListener
+}
+
+type ConsumerEventListener interface {
+	BecameActive(consumer Consumer, topicName string, partition int32)
+	BecameInactive(consumer Consumer, topicName string, partition int32)
 }
 
 type partitionConsumer struct {
@@ -161,6 +168,19 @@ type partitionConsumer struct {
 	unAckChunksTracker *unAckChunksTracker
 }
 
+func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
+	listener := pc.options.consumerEventListener
+	if listener == nil {
+		// didn't set a listener
+		return
+	}
+	if isActive {
+		listener.BecameActive(pc.parentConsumer, pc.topic, pc.partitionIdx)
+	} else {
+		listener.BecameInactive(pc.parentConsumer, pc.topic, pc.partitionIdx)
+	}
+}
+
 type availablePermits struct {
 	permits int32
 	pc      *partitionConsumer
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index fc1b349..fc97455 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -85,6 +85,8 @@ type Connection interface {
 type ConsumerHandler interface {
 	MessageReceived(response *pb.CommandMessage, headersAndPayload Buffer) error
 
+	ActiveConsumerChanged(isActive bool)
+
 	// ConnectionClosed close the TCP connection.
 	ConnectionClosed()
 }
@@ -576,6 +578,7 @@ func (c *connection) internalReceivedCommand(cmd *pb.BaseCommand, headersAndPayl
 		c.handlePong()
 
 	case pb.BaseCommand_ACTIVE_CONSUMER_CHANGE:
+		c.handleActiveConsumerChange(cmd.GetActiveConsumerChange())
 
 	default:
 		c.log.Errorf("Received invalid command type: %s", cmd.Type)
@@ -857,6 +860,16 @@ func (c *connection) handleCloseConsumer(closeConsumer *pb.CommandCloseConsumer)
 	}
 }
 
+func (c *connection) handleActiveConsumerChange(consumerChange *pb.CommandActiveConsumerChange) {
+	consumerID := consumerChange.GetConsumerId()
+	isActive := consumerChange.GetIsActive()
+	if consumer, ok := c.consumerHandler(consumerID); ok {
+		consumer.ActiveConsumerChanged(isActive)
+	} else {
+		c.log.WithField("consumerID", consumerID).Warnf("Consumer not found while active consumer change")
+	}
+}
+
 func (c *connection) handleCloseProducer(closeProducer *pb.CommandCloseProducer) {
 	c.log.Infof("Broker notification of Closed producer: %d", closeProducer.GetProducerId())
 	producerID := closeProducer.GetProducerId()


Test patch:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 95462ba..58a524a 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -25,6 +25,7 @@ import (
 	"log"
 	"net/http"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -321,7 +322,7 @@ func TestConsumerKeyShared(t *testing.T) {
 	assert.NotEqual(t, 0, receivedConsumer1)
 	assert.NotEqual(t, 0, receivedConsumer2)
 
-	fmt.Printf("TestConsumerKeyShared received messages consumer1: %d consumser2: %d\n",
+	t.Logf("TestConsumerKeyShared received messages consumer1: %d consumser2: %d\n",
 		receivedConsumer1, receivedConsumer2)
 	assert.Equal(t, 100, receivedConsumer1+receivedConsumer2)
 }
@@ -375,7 +376,7 @@ func TestPartitionTopicsConsumerPubSub(t *testing.T) {
 		assert.Nil(t, err)
 		msgs = append(msgs, string(msg.Payload()))
 
-		fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+		t.Logf("Received message msgId: %#v -- content: '%s'\n",
 			msg.ID(), string(msg.Payload()))
 
 		consumer.Ack(msg)
@@ -384,6 +385,141 @@ func TestPartitionTopicsConsumerPubSub(t *testing.T) {
 	assert.Equal(t, len(msgs), 10)
 }
 
+type TestActiveConsumerListener struct {
+	t                *testing.T
+	lock             sync.RWMutex
+	nameToPartitions map[string]map[int32]struct{}
+}
+
+func (l *TestActiveConsumerListener) getConsumerCount() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return len(l.nameToPartitions)
+}
+
+func (l *TestActiveConsumerListener) getPartitionCount(consumerName string) int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return len(l.nameToPartitions[consumerName])
+}
+
+func (l *TestActiveConsumerListener) BecameActive(consumer Consumer, topicName string, partition int32) {
+	l.t.Logf("%s become active on %s - %d\n", consumer.Name(), topicName, partition)
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	partitionSet := l.nameToPartitions[consumer.Name()]
+	if partitionSet == nil {
+		partitionSet = map[int32]struct{}{}
+	}
+	partitionSet[partition] = struct{}{}
+	l.nameToPartitions[consumer.Name()] = partitionSet
+}
+
+func (l *TestActiveConsumerListener) BecameInactive(consumer Consumer, topicName string, partition int32) {
+	l.t.Logf("%s become inactive on %s - %d\n", consumer.Name(), topicName, partition)
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	partitionSet := l.nameToPartitions[consumer.Name()]
+	if _, ok := partitionSet[partition]; ok {
+		delete(partitionSet, partition)
+		if len(partitionSet) == 0 {
+			delete(l.nameToPartitions, consumer.Name())
+		}
+	}
+}
+
+func allConsume(consumers []Consumer) {
+	for i := 0; i < len(consumers); i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		consumers[i].Receive(ctx)
+	}
+}
+
+func TestPartitionTopic_ActiveConsumerChanged(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	randomName := newTopicName()
+	topic := "persistent://public/default/" + randomName
+	testURL := adminURL + "/" + "admin/v2/persistent/public/default/" + randomName + "/partitions"
+
+	makeHTTPCall(t, http.MethodPut, testURL, "3")
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: topic,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx := context.Background()
+	for i := 0; i < 10; i++ {
+		_, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.Nil(t, err)
+	}
+
+	var consumers []Consumer
+	listener := &TestActiveConsumerListener{
+		t:                t,
+		nameToPartitions: map[string]map[int32]struct{}{},
+	}
+	for i := 0; i < 1; i++ {
+		consumer, err := client.Subscribe(ConsumerOptions{
+			Topic:            topic,
+			Name:             fmt.Sprintf("consumer-%d", i),
+			SubscriptionName: "my-sub",
+			Type:             Failover,
+			EventListener:    listener,
+		})
+		assert.Nil(t, err)
+		defer consumer.Close()
+		consumers = append(consumers, consumer)
+	}
+
+	allConsume(consumers)
+	// first consumer will get 3 partitions
+	assert.Equal(t, 1, listener.getConsumerCount())
+	assert.Equal(t, 3, listener.getPartitionCount(consumers[0].Name()))
+
+	// 1 partition per consumer
+	for i := 1; i < 3; i++ {
+		consumer, err := client.Subscribe(ConsumerOptions{
+			Topic:            topic,
+			Name:             fmt.Sprintf("consumer-%d", i),
+			SubscriptionName: "my-sub",
+			Type:             Failover,
+			EventListener:    listener,
+		})
+		assert.Nil(t, err)
+		defer consumer.Close()
+		consumers = append(consumers, consumer)
+	}
+	allConsume(consumers)
+	assert.Equal(t, 3, listener.getConsumerCount())
+	for _, c := range consumers {
+		assert.Equal(t, 1, listener.getPartitionCount(c.Name()))
+	}
+
+	consumers[0].Close()
+	// wait broker reschedule active consumers
+	time.Sleep(time.Second * 3)
+	allConsume(consumers)
+
+	// close consumer won't get notify
+	assert.Equal(t, 3, listener.getConsumerCount())
+	// residual consumers will cover all partitions
+	assert.Equal(t, 3, listener.getPartitionCount(consumers[1].Name())+listener.getPartitionCount(consumers[2].Name()))
+	for _, c := range consumers {
+		c.Close()
+	}
+}
+
 func TestPartitionTopicsConsumerPubSubEncryption(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
@@ -443,7 +579,7 @@ func TestPartitionTopicsConsumerPubSubEncryption(t *testing.T) {
 		assert.Nil(t, err)
 		msgs = append(msgs, string(msg.Payload()))
 
-		fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+		t.Logf("Received message msgId: %#v -- content: '%s'\n",
 			msg.ID(), string(msg.Payload()))
 
 		consumer.Ack(msg)
@@ -536,7 +672,7 @@ func TestConsumerShared(t *testing.T) {
 			readMsgs++
 			payload := string(cm.Message.Payload())
 			messages[payload] = struct{}{}
-			fmt.Printf("consumer1 msg id is: %v, value is: %s\n", cm.Message.ID(), payload)
+			t.Logf("consumer1 msg id is: %v, value is: %s\n", cm.Message.ID(), payload)
 			consumer1.Ack(cm.Message)
 		case cm, ok := <-consumer2.Chan():
 			if !ok {
@@ -545,7 +681,7 @@ func TestConsumerShared(t *testing.T) {
 			readMsgs++
 			payload := string(cm.Message.Payload())
 			messages[payload] = struct{}{}
-			fmt.Printf("consumer2 msg id is: %v, value is: %s\n", cm.Message.ID(), payload)
+			t.Logf("consumer2 msg id is: %v, value is: %s\n", cm.Message.ID(), payload)
 			consumer2.Ack(cm.Message)
 		}
 	}
@@ -1764,7 +1900,7 @@ func TestConsumerAddTopicPartitions(t *testing.T) {
 		assert.Nil(t, err)
 		msgs = append(msgs, string(msg.Payload()))
 
-		fmt.Printf("Received message msgId: %#v -- content: '%s'\n",
+		t.Logf("Received message msgId: %#v -- content: '%s'\n",
 			msg.ID(), string(msg.Payload()))
 
 		consumer.Ack(msg)
@@ -2105,11 +2241,11 @@ func TestKeyBasedBatchProducerConsumerKeyShared(t *testing.T) {
 	assert.Equal(t, len(consumer1Keys)*MsgBatchCount, receivedConsumer1)
 	assert.Equal(t, len(consumer2Keys)*MsgBatchCount, receivedConsumer2)
 
-	fmt.Printf("TestKeyBasedBatchProducerConsumerKeyShared received messages consumer1: %d consumser2: %d\n",
+	t.Logf("TestKeyBasedBatchProducerConsumerKeyShared received messages consumer1: %d consumser2: %d\n",
 		receivedConsumer1, receivedConsumer2)
 	assert.Equal(t, 300, receivedConsumer1+receivedConsumer2)
 
-	fmt.Printf("TestKeyBasedBatchProducerConsumerKeyShared received messages keys consumer1: %v consumser2: %v\n",
+	t.Logf("TestKeyBasedBatchProducerConsumerKeyShared received messages keys consumer1: %v consumser2: %v\n",
 		consumer1Keys, consumer2Keys)
 }
 
@@ -2287,7 +2423,7 @@ func TestConsumerKeySharedWithOrderingKey(t *testing.T) {
 	assert.NotEqual(t, 0, receivedConsumer1)
 	assert.NotEqual(t, 0, receivedConsumer2)
 
-	fmt.Printf(
+	t.Logf(
 		"TestConsumerKeySharedWithOrderingKey received messages consumer1: %d consumser2: %d\n",
 		receivedConsumer1, receivedConsumer2,
 	)
@@ -2623,7 +2759,7 @@ func TestProducerConsumerRedeliveryOfFailedEncryptedMessages(t *testing.T) {
 			Value: fmt.Sprintf(message, i),
 		})
 		assert.Nil(t, err)
-		fmt.Printf("Sent : %v\n", mid)
+		t.Logf("Sent : %v\n", mid)
 	}
 
 	// Consuming from consumer 2 and 3
@@ -2656,7 +2792,7 @@ func TestProducerConsumerRedeliveryOfFailedEncryptedMessages(t *testing.T) {
 		assert.Nil(t, err)
 		messageMap[*receivedMsg] = struct{}{}
 		cryptoConsumer.Ack(m)
-		fmt.Printf("Received : %v\n", m.ID())
+		t.Logf("Received : %v\n", m.ID())
 	}
 
 	// check if all messages were received
@@ -2904,7 +3040,7 @@ func TestBatchMessageReceiveWithCompressionAndRSAEcnryption(t *testing.T) {
 
 	for i := 0; i < numOfMessages; i++ {
 		msg, err := consumer.Receive(ctx)
-		fmt.Printf("received : %v\n", string(msg.Payload()))
+		t.Logf("received : %v\n", string(msg.Payload()))
 		assert.Nil(t, err)
 		consumer.Ack(msg)
 		count++