Repository: apache/pulsar-client-go
Pull request #944 (2023-01-17 05:57:14)

[feat] Support Exclusive Producer access mode.
Master Issue: #931

### Motivation

#931

### Modifications

- Support the `Exclusive` Producer access mode config in the `ProducerOptions`.

**Note**: `WaitForExclusive` and `ExclusiveWithFencing` will be supported in other PR.


### Verifying this change

- Add `TestExclusiveProducer` to cover it.

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)
  - 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


shibd: @nodece @freeznet Can you help look at ci failed? https://github.com/apache/pulsar-client-go/actions/runs/4101318705/jobs/7072982772

It doesn't seem to have anything to do with this PR.

nodece: > @nodece @freeznet Can you help look at ci failed? https://github.com/apache/pulsar-client-go/actions/runs/4101318705/jobs/7072982772
> 
> It doesn't seem to have anything to do with this PR.

It looks like a `golang-lint@1.51.x` issue,  which was released yesterday. Could you fix the `golang-lint` version to `1.50.1`?


Patch:
diff --git a/pulsar/producer.go b/pulsar/producer.go
index d088fb2..d9ac34b 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -55,6 +55,17 @@ const (
 	Better
 )
 
+type ProducerAccessMode int
+
+const (
+	// ProducerAccessModeShared is default multiple producers can publish on a topic.
+	ProducerAccessModeShared ProducerAccessMode = iota
+
+	// ProducerAccessModeExclusive is required exclusive access for producer.
+	// Fail immediately if there's already a producer connected.
+	ProducerAccessModeExclusive
+)
+
 // TopicMetadata represents a topic metadata.
 type TopicMetadata interface {
 	// NumPartitions returns the number of partitions for a particular topic.
@@ -187,6 +198,12 @@ type ProducerOptions struct {
 	// ChunkMaxMessageSize is the max size of single chunk payload.
 	// It will actually only take effect if it is smaller than the maxMessageSize from the broker.
 	ChunkMaxMessageSize uint
+
+	// The type of access to the topic that the producer requires. (default ProducerAccessModeShared)
+	// Options:
+	// - ProducerAccessModeShared
+	// - ProducerAccessModeExclusive
+	ProducerAccessMode
 }
 
 // Producer is used to publish messages on a topic
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index b0467f5..eece055 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -95,6 +95,7 @@ type partitionProducer struct {
 	metrics          *internal.LeveledMetrics
 	epoch            uint64
 	schemaCache      *schemaCache
+	topicEpoch       *uint64
 }
 
 type schemaCache struct {
@@ -237,6 +238,11 @@ func (p *partitionProducer) grabCnx() error {
 		Schema:                   pbSchema,
 		Epoch:                    proto.Uint64(atomic.LoadUint64(&p.epoch)),
 		UserProvidedProducerName: proto.Bool(p.userProvidedProducerName),
+		ProducerAccessMode:       toProtoProducerAccessMode(p.options.ProducerAccessMode).Enum(),
+	}
+
+	if p.topicEpoch != nil {
+		cmdProducer.TopicEpoch = proto.Uint64(*p.topicEpoch)
 	}
 
 	if p.producerName != "" {
@@ -253,6 +259,8 @@ func (p *partitionProducer) grabCnx() error {
 	}
 
 	p.producerName = res.Response.ProducerSuccess.GetProducerName()
+	nextTopicEpoch := res.Response.ProducerSuccess.GetTopicEpoch()
+	p.topicEpoch = &nextTopicEpoch
 
 	if p.options.Encryption != nil {
 		p.encryptor = internalcrypto.NewProducerEncryptor(p.options.Encryption.Keys,
@@ -1352,3 +1360,14 @@ func (c *chunkRecorder) setFirstChunkID(msgID messageID) {
 func (c *chunkRecorder) setLastChunkID(msgID messageID) {
 	c.chunkedMsgID.messageID = msgID
 }
+
+func toProtoProducerAccessMode(accessMode ProducerAccessMode) pb.ProducerAccessMode {
+	switch accessMode {
+	case ProducerAccessModeShared:
+		return pb.ProducerAccessMode_Shared
+	case ProducerAccessModeExclusive:
+		return pb.ProducerAccessMode_Exclusive
+	}
+
+	return pb.ProducerAccessMode_Shared
+}


Test patch:
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 75fc6db..d7950eb 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"net/http"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -1705,3 +1706,34 @@ func TestProducerWithSchemaAndConsumerSchemaNotFound(t *testing.T) {
 	// should fail with error but not panic
 	assert.Error(t, err)
 }
+
+func TestExclusiveProducer(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:              topicName,
+		ProducerAccessMode: ProducerAccessModeExclusive,
+	})
+
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+	defer producer.Close()
+
+	_, err = client.CreateProducer(ProducerOptions{
+		Topic:              topicName,
+		ProducerAccessMode: ProducerAccessModeExclusive,
+	})
+	assert.Error(t, err, "Producer should be fenced")
+	assert.True(t, strings.Contains(err.Error(), "ProducerFenced"))
+
+	_, err = client.CreateProducer(ProducerOptions{
+		Topic: topicName,
+	})
+	assert.Error(t, err, "Producer should be failed")
+	assert.True(t, strings.Contains(err.Error(), "ProducerBusy"))
+}