[PIP-105] Part-1 Support add subscription properties (#12869)

### Motivation
See #12269

I divided this PIP into two parts, the first part supports setting properties for subscriptions
The second part will support plug-in loading, filtering, etc.

### Modifications
Support add subscription properties

Author: 315157973

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1010,20 +1010,22 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                                 new SubscriptionNotFoundException(
                                                         "Subscription does not exist"));
                             }
-
+                            SubscriptionOption option = SubscriptionOption.builder().cnx(ServerCnx.this)
+                                    .subscriptionName(subscriptionName)
+                                    .consumerId(consumerId).subType(subType).priorityLevel(priorityLevel)
+                                    .consumerName(consumerName).isDurable(isDurable).subType(subType)
+                                    .startMessageId(startMessageId).metadata(metadata).readCompacted(readCompacted)
+                                    .initialPosition(initialPosition)
+                                    .startMessageRollbackDurationSec(startMessageRollbackDurationSec)
+                                    .replicatedSubscriptionStateArg(isReplicated).keySharedMeta(keySharedMeta)
+                                    .subscriptionProperties(SubscriptionOption.getPropertiesMap(
+                                            subscribe.getSubscriptionPropertiesList()))
+                                    .build();
                             if (schema != null) {
                                 return topic.addSchemaIfIdleOrCheckCompatible(schema)
-                                        .thenCompose(v -> topic.subscribe(
-                                                ServerCnx.this, subscriptionName, consumerId,
-                                                subType, priorityLevel, consumerName, isDurable,
-                                                startMessageId, metadata,
-                                                readCompacted, initialPosition, startMessageRollbackDurationSec,
-                                                isReplicated, keySharedMeta));
+                                        .thenCompose(v -> topic.subscribe(option));
                             } else {
-                                return topic.subscribe(ServerCnx.this, subscriptionName, consumerId,
-                                    subType, priorityLevel, consumerName, isDurable,
-                                    startMessageId, metadata, readCompacted, initialPosition,
-                                    startMessageRollbackDurationSec, isReplicated, keySharedMeta);
+                                return topic.subscribe(option);
                             }
                         })
                         .thenAccept(consumer -> {

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SubscriptionOption.java

@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import lombok.Builder;
+import lombok.Getter;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.common.api.proto.CommandSubscribe;
+import org.apache.pulsar.common.api.proto.KeySharedMeta;
+import org.apache.pulsar.common.api.proto.KeyValue;
+
+@Getter
+@Builder
+public class SubscriptionOption {
+    private final TransportCnx cnx;
+    private String subscriptionName;
+    private long consumerId;
+    private CommandSubscribe.SubType subType;
+    private int priorityLevel;
+    private String consumerName;
+    private boolean isDurable;
+    private MessageId startMessageId;
+    private Map<String, String> metadata;
+    private boolean readCompacted;
+    private CommandSubscribe.InitialPosition initialPosition;
+    private long startMessageRollbackDurationSec;
+    private boolean replicatedSubscriptionStateArg;
+    private KeySharedMeta keySharedMeta;
+    private Optional<Map<String, String>> subscriptionProperties;
+
+    public static Optional<Map<String, String>> getPropertiesMap(List<KeyValue> list) {
+        if (list == null) {
+            return Optional.of(Collections.emptyMap());
+        }
+        return Optional.of(list.stream().collect(Collectors.toMap(
+                KeyValue::getKey, KeyValue::getValue, (key1, key2) -> key1)));
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -129,6 +129,7 @@ default boolean isMarkerMessage() {
      */
     void recordAddLatency(long latency, TimeUnit unit);
 
+    @Deprecated
     CompletableFuture<Consumer> subscribe(TransportCnx cnx, String subscriptionName, long consumerId, SubType subType,
                                           int priorityLevel, String consumerName, boolean isDurable,
                                           MessageId startMessageId,
@@ -137,6 +138,13 @@ CompletableFuture<Consumer> subscribe(TransportCnx cnx, String subscriptionName,
                                           long startMessageRollbackDurationSec, boolean replicateSubscriptionState,
                                           KeySharedMeta keySharedMeta);
 
+    /**
+     * Subscribe a topic.
+     * @param option
+     * @return
+     */
+    CompletableFuture<Consumer> subscribe(SubscriptionOption option);
+
     CompletableFuture<Subscription> createSubscription(String subscriptionName, InitialPosition initialPosition,
             boolean replicateSubscriptionState);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -55,6 +55,7 @@
 import org.apache.pulsar.broker.service.Replicator;
 import org.apache.pulsar.broker.service.StreamingStats;
 import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.SubscriptionOption;
 import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.TransportCnx;
 import org.apache.pulsar.broker.stats.ClusterReplicationMetrics;
@@ -224,6 +225,16 @@ public CompletableFuture<Void> checkIfTransactionBufferRecoverCompletely(boolean
         return  CompletableFuture.completedFuture(null);
     }
 
+    @Override
+    public CompletableFuture<Consumer> subscribe(SubscriptionOption option) {
+        return internalSubscribe(option.getCnx(), option.getSubscriptionName(), option.getConsumerId(),
+                option.getSubType(), option.getPriorityLevel(), option.getConsumerName(),
+                option.isDurable(), option.getStartMessageId(), option.getMetadata(),
+                option.isReadCompacted(), option.getInitialPosition(),
+                option.getStartMessageRollbackDurationSec(), option.isReplicatedSubscriptionStateArg(),
+                option.getKeySharedMeta());
+    }
+
     @Override
     public CompletableFuture<Consumer> subscribe(final TransportCnx cnx, String subscriptionName, long consumerId,
                                                  SubType subType, int priorityLevel, String consumerName,
@@ -232,6 +243,19 @@ public CompletableFuture<Consumer> subscribe(final TransportCnx cnx, String subs
                                                  InitialPosition initialPosition,
                                                  long resetStartMessageBackInSec, boolean replicateSubscriptionState,
                                                  KeySharedMeta keySharedMeta) {
+        return internalSubscribe(cnx, subscriptionName, consumerId, subType, priorityLevel, consumerName,
+                isDurable, startMessageId, metadata, readCompacted, initialPosition, resetStartMessageBackInSec,
+                replicateSubscriptionState, keySharedMeta);
+    }
+
+    private CompletableFuture<Consumer> internalSubscribe(final TransportCnx cnx, String subscriptionName,
+                                                          long consumerId, SubType subType, int priorityLevel,
+                                                          String consumerName, boolean isDurable,
+                                                          MessageId startMessageId, Map<String, String> metadata,
+                                                          boolean readCompacted, InitialPosition initialPosition,
+                                                          long resetStartMessageBackInSec,
+                                                          boolean replicateSubscriptionState,
+                                                          KeySharedMeta keySharedMeta) {
 
         return brokerService.checkTopicNsOwnership(getName()).thenCompose(__ -> {
             final CompletableFuture<Consumer> future = new CompletableFuture<>();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -22,6 +22,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -116,6 +117,7 @@ public class PersistentSubscription implements Subscription {
     private volatile ReplicatedSubscriptionSnapshotCache replicatedSubscriptionSnapshotCache;
     private volatile Position lastMarkDeleteForTransactionMarker;
     private final PendingAckHandle pendingAckHandle;
+    private Map<String, String> subscriptionProperties;
 
     private final LongAdder bytesOutFromRemovedConsumers = new LongAdder();
     private final LongAdder msgOutFromRemovedConsumer = new LongAdder();
@@ -143,14 +145,21 @@ static boolean isCursorFromReplicatedSubscription(ManagedCursor cursor) {
     }
 
     public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
-            boolean replicated) {
+                                                         boolean replicated) {
+        this(topic, subscriptionName, cursor, replicated, Collections.emptyMap());
+    }
+
+    public PersistentSubscription(PersistentTopic topic, String subscriptionName, ManagedCursor cursor,
+            boolean replicated, Map<String, String> subscriptionProperties) {
         this.topic = topic;
         this.cursor = cursor;
         this.topicName = topic.getName();
         this.subName = subscriptionName;
         this.fullName = MoreObjects.toStringHelper(this).add("topic", topicName).add("name", subName).toString();
         this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
         this.setReplicated(replicated);
+        this.subscriptionProperties = subscriptionProperties == null
+                ? new HashMap<>() : Collections.unmodifiableMap(subscriptionProperties);
         if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
                 && !checkTopicIsEventsNames(TopicName.get(topicName))) {
             this.pendingAckHandle = new PendingAckHandleImpl(this);
@@ -1141,13 +1150,16 @@ void topicTerminated() {
         }
     }
 
+    public Map<String, String> getSubscriptionProperties() {
+        return subscriptionProperties;
+    }
+
     /**
      * Return a merged map that contains the cursor properties specified by used
      * (eg. when using compaction subscription) and the subscription properties.
      */
     protected Map<String, Long> mergeCursorProperties(Map<String, Long> userProperties) {
-        Map<String, Long> baseProperties = isReplicated() ? REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES
-                : NON_REPLICATED_SUBSCRIPTION_CURSOR_PROPERTIES;
+        Map<String, Long> baseProperties = getBaseCursorProperties(isReplicated());
 
         if (userProperties.isEmpty()) {
             // Use only the static instance in the common case