Add subscription properties for ConsumerOptions (apache#671)

Signed-off-by: xiaolongran <xiaolongran@tencent.com>

### Motivation

In apache/pulsar#12869, we introduce pluggable entry filter in Dispatcher, the pull request is the Go SDK implementation of this PIP


### Modifications

*Describe the modifications you've done.*

- Add subscription properties for ConsumerOptions 
- Update `pulsarApi.proto` file

pulsar/consumer.go

@@ -101,6 +101,12 @@ type ConsumerOptions struct {
 	// Those properties will be visible in the topic stats
 	Properties map[string]string
 
+	// SubscriptionProperties specify the subscription properties for this subscription.
+	//
+	// > Notice: SubscriptionProperties are immutable, and consumers under the same subscription will fail to create a
+	// > subscription if they use different properties.
+	SubscriptionProperties map[string]string
+
 	// Type specifies the subscription type to be used when subscribing to a topic.
 	// Default is `Exclusive`
 	Type SubscriptionType

pulsar/consumer_impl.go

@@ -297,6 +297,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 
 	receiverQueueSize := c.options.ReceiverQueueSize
 	metadata := c.options.Properties
+	subProperties := c.options.SubscriptionProperties
 
 	startPartition := oldNumPartitions
 	partitionsToAdd := newNumPartitions - oldNumPartitions
@@ -333,6 +334,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				nackRedeliveryDelay:        nackRedeliveryDelay,
 				nackBackoffPolicy:          c.options.NackBackoffPolicy,
 				metadata:                   metadata,
+				subProperties:              subProperties,
 				replicateSubscriptionState: c.options.ReplicateSubscriptionState,
 				startMessageID:             trackingMessageID{},
 				subscriptionMode:           durable,

pulsar/consumer_partition.go

@@ -91,6 +91,7 @@ type partitionConsumerOpts struct {
 	nackRedeliveryDelay        time.Duration
 	nackBackoffPolicy          NackBackoffPolicy
 	metadata                   map[string]string
+	subProperties              map[string]string
 	replicateSubscriptionState bool
 	startMessageID             trackingMessageID
 	startMessageIDInclusive    bool
@@ -1058,6 +1059,7 @@ func (pc *partitionConsumer) grabConn() error {
 		PriorityLevel:              nil,
 		Durable:                    proto.Bool(pc.options.subscriptionMode == durable),
 		Metadata:                   internal.ConvertFromStringMap(pc.options.metadata),
+		SubscriptionProperties:     internal.ConvertFromStringMap(pc.options.subProperties),
 		ReadCompacted:              proto.Bool(pc.options.readCompacted),
 		Schema:                     pbSchema,
 		InitialPosition:            initialPosition.Enum(),
@@ -1075,6 +1077,10 @@ func (pc *partitionConsumer) grabConn() error {
 		cmdSubscribe.Metadata = toKeyValues(pc.options.metadata)
 	}
 
+	if len(pc.options.subProperties) > 0 {
+		cmdSubscribe.SubscriptionProperties = toKeyValues(pc.options.subProperties)
+	}
+
 	// force topic creation is enabled by default so
 	// we only need to set the flag when disabling it
 	if pc.options.disableForceTopicCreation {