[feature][client] PIP-184: Topic specific consumer priorityLevel

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java

@@ -813,4 +813,38 @@ public interface ConsumerBuilder<T> extends Cloneable {
      * @param enabled whether to enable AutoScaledReceiverQueueSize.
      */
     ConsumerBuilder<T> autoScaledReceiverQueueSizeEnabled(boolean enabled);
+
+    /**
+     * Configure topic specific options to override those set at the {@link ConsumerBuilder} level.
+     *
+     * @param topicName a topic name
+     * @return a {@link TopicConsumerBuilder} instance
+     */
+    TopicConsumerBuilder<T> topicConfiguration(String topicName);
+
+    /**
+     * Configure topic specific options to override those set at the {@link ConsumerBuilder} level.
+     *
+     * @param topicName a topic name
+     * @param builderConsumer a consumer to allow the configuration of the {@link TopicConsumerBuilder} instance
+     */
+    ConsumerBuilder<T> topicConfiguration(String topicName,
+                                          java.util.function.Consumer<TopicConsumerBuilder<T>> builderConsumer);
+
+    /**
+     * Configure topic specific options to override those set at the {@link ConsumerBuilder} level.
+     *
+     * @param topicsPattern a regular expression to match a topic name
+     * @return a {@link TopicConsumerBuilder} instance
+     */
+    TopicConsumerBuilder<T> topicConfiguration(Pattern topicsPattern);
+
+    /**
+     * Configure topic specific options to override those set at the {@link ConsumerBuilder} level.
+     *
+     * @param topicsPattern a regular expression to match a topic name
+     * @param builderConsumer a consumer to allow the configuration of the {@link TopicConsumerBuilder} instance
+     */
+    ConsumerBuilder<T> topicConfiguration(Pattern topicsPattern,
+                                          java.util.function.Consumer<TopicConsumerBuilder<T>> builderConsumer);
 }

pulsar-client-api/src/main/java/org/apache/pulsar/client/api/TopicConsumerBuilder.java

@@ -0,0 +1,47 @@
+/**
+ * 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.client.api;
+
+/**
+ * {@link TopicConsumerBuilder} is used to configure topic specific options to override those set at the
+ * {@link ConsumerBuilder} level.
+ *
+ * @see ConsumerBuilder#topicConfiguration(String)
+ *
+ * @param <T> the type of the value in the {@link ConsumerBuilder}
+ */
+public interface TopicConsumerBuilder<T> {
+    /**
+     * Configure the priority level of this topic.
+     *
+     * @see ConsumerBuilder#priorityLevel(int)
+     *
+     * @param priorityLevel the priority of this topic
+     * @return the {@link TopicConsumerBuilder} instance
+     */
+    TopicConsumerBuilder<T> priorityLevel(int priorityLevel);
+
+    /**
+     * Complete the configuration of the topic specific options and return control back to the
+     * {@link ConsumerBuilder} instance.
+     *
+     * @return the {@link ConsumerBuilder} instance
+     */
+    ConsumerBuilder<T> build();
+}

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java

@@ -52,8 +52,10 @@
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionMode;
 import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.TopicConsumerBuilder;
 import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.conf.TopicConsumerConfigurationData;
 import org.apache.pulsar.client.util.RetryMessageUtil;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
@@ -537,4 +539,32 @@ public ConsumerBuilder<T> autoScaledReceiverQueueSizeEnabled(boolean enabled) {
         conf.setAutoScaledReceiverQueueSizeEnabled(enabled);
         return this;
     }
+
+    @Override
+    public TopicConsumerBuilder<T> topicConfiguration(String topicName) {
+        TopicConsumerConfigurationData topicConf = TopicConsumerConfigurationData.ofTopicName(topicName, conf);
+        conf.getTopicConfigurations().add(topicConf);
+        return new TopicConsumerBuilderImpl<>(this, topicConf);
+    }
+
+    @Override
+    public ConsumerBuilder<T> topicConfiguration(String topicName,
+                                                 java.util.function.Consumer<TopicConsumerBuilder<T>> builderConsumer) {
+        builderConsumer.accept(topicConfiguration(topicName));
+        return this;
+    }
+
+    @Override
+    public TopicConsumerBuilder<T> topicConfiguration(Pattern topicsPattern) {
+        TopicConsumerConfigurationData topicConf = TopicConsumerConfigurationData.ofTopicsPattern(topicsPattern, conf);
+        conf.getTopicConfigurations().add(topicConf);
+        return new TopicConsumerBuilderImpl<>(this, topicConf);
+    }
+
+    @Override
+    public ConsumerBuilder<T> topicConfiguration(Pattern topicsPattern,
+                                                 java.util.function.Consumer<TopicConsumerBuilder<T>> builderConsumer) {
+        builderConsumer.accept(topicConfiguration(topicsPattern));
+        return this;
+    }
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -61,6 +61,8 @@
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import lombok.AccessLevel;
+import lombok.Getter;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
@@ -146,6 +148,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
     private final NegativeAcksTracker negativeAcksTracker;
 
     protected final ConsumerStatsRecorder stats;
+    @Getter(AccessLevel.PACKAGE)
     private final int priorityLevel;
     private final SubscriptionMode subscriptionMode;
     private volatile BatchMessageIdImpl startMessageId;
@@ -266,7 +269,7 @@ protected ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurat
         this.partitionIndex = partitionIndex;
         this.hasParentConsumer = hasParentConsumer;
         this.parentConsumerHasListener = parentConsumerHasListener;
-        this.priorityLevel = conf.getPriorityLevel();
+        this.priorityLevel = conf.getMatchingTopicConfiguration(topic).getPriorityLevel();
         this.readCompacted = conf.isReadCompacted();
         this.subscriptionInitialPosition = conf.getSubscriptionInitialPosition();
         this.negativeAcksTracker = new NegativeAcksTracker(this, conf);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicConsumerBuilderImpl.java

@@ -0,0 +1,43 @@
+/**
+ * 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.client.impl;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import lombok.RequiredArgsConstructor;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.TopicConsumerBuilder;
+import org.apache.pulsar.client.impl.conf.TopicConsumerConfigurationData;
+
+@RequiredArgsConstructor
+class TopicConsumerBuilderImpl<T> implements TopicConsumerBuilder<T> {
+    private final ConsumerBuilder<T> consumerBuilder;
+    private final TopicConsumerConfigurationData topicConf;
+
+    @Override
+    public TopicConsumerBuilder<T> priorityLevel(int priorityLevel) {
+        checkArgument(priorityLevel >= 0, "priorityLevel needs to be >= 0");
+        topicConf.setPriorityLevel(priorityLevel);
+        return this;
+    }
+
+    @Override
+    public ConsumerBuilder<T> build() {
+        return consumerBuilder;
+    }
+}

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java

@@ -22,6 +22,8 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.google.common.collect.Sets;
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.SortedMap;
@@ -164,6 +166,20 @@ public int getMaxPendingChuckedMessage() {
 
     private boolean autoScaledReceiverQueueSizeEnabled = false;
 
+    private List<TopicConsumerConfigurationData> topicConfigurations = new ArrayList<>();
+
+    public TopicConsumerConfigurationData getMatchingTopicConfiguration(String topicName) {
+        return topicConfigurations.stream()
+                .filter(topicConf -> topicConf.getTopicNameMatcher().matches(topicName))
+                .findFirst()
+                .orElseGet(() -> TopicConsumerConfigurationData.ofTopicName(topicName, this));
+    }
+
+    public void setTopicConfigurations(List<TopicConsumerConfigurationData> topicConfigurations) {
+        checkArgument(topicConfigurations != null, "topicConfigurations should not be null.");
+        this.topicConfigurations = topicConfigurations;
+    }
+
     public void setAutoUpdatePartitionsIntervalSeconds(int interval, TimeUnit timeUnit) {
         checkArgument(interval > 0, "interval needs to be > 0");
         this.autoUpdatePartitionsIntervalSeconds = timeUnit.toSeconds(interval);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/TopicConsumerConfigurationData.java

@@ -0,0 +1,85 @@
+/**
+ * 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.client.impl.conf;
+
+import java.io.Serializable;
+import java.util.regex.Pattern;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class TopicConsumerConfigurationData implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+    private TopicNameMatcher topicNameMatcher;
+    private int priorityLevel;
+
+    public static TopicConsumerConfigurationData ofTopicsPattern(@NonNull Pattern topicsPattern, int priorityLevel) {
+        return of(new TopicNameMatcher.TopicsPattern(topicsPattern), priorityLevel);
+    }
+
+    public static TopicConsumerConfigurationData ofTopicsPattern(@NonNull Pattern topicsPattern,
+                                                                 ConsumerConfigurationData<?> conf) {
+        return ofTopicsPattern(topicsPattern, conf.getPriorityLevel());
+    }
+
+    public static TopicConsumerConfigurationData ofTopicName(@NonNull String topicName, int priorityLevel) {
+        return of(new TopicNameMatcher.TopicName(topicName), priorityLevel);
+    }
+
+    public static TopicConsumerConfigurationData ofTopicName(@NonNull String topicName,
+                                                             ConsumerConfigurationData<?> conf) {
+        return ofTopicName(topicName, conf.getPriorityLevel());
+    }
+
+    static TopicConsumerConfigurationData of(@NonNull TopicNameMatcher topicNameMatcher, int priorityLevel) {
+        return new TopicConsumerConfigurationData(topicNameMatcher, priorityLevel);
+    }
+
+    public interface TopicNameMatcher extends Serializable {
+        boolean matches(String topicName);
+
+        @RequiredArgsConstructor
+        class TopicsPattern implements TopicNameMatcher {
+            @NonNull
+            private final Pattern topicsPattern;
+
+            @Override
+            public boolean matches(String topicName) {
+                return topicsPattern.matcher(topicName).matches();
+            }
+        }
+
+        @RequiredArgsConstructor
+        class TopicName implements TopicNameMatcher {
+            @NonNull
+            private final String topicName;
+
+            @Override
+            public boolean matches(String topicName) {
+                return this.topicName.equals(topicName);
+            }
+        }
+    }
+}

pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerBuilderImplTest.java

@@ -18,10 +18,12 @@
  */
 package org.apache.pulsar.client.impl;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import static org.testng.Assert.assertNotNull;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -31,12 +33,14 @@
 import java.util.regex.Pattern;
 import org.apache.pulsar.client.api.BatchReceivePolicy;
 import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
 import org.apache.pulsar.client.api.DeadLetterPolicy;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
 import org.apache.pulsar.client.api.SubscriptionMode;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
+import org.apache.pulsar.client.impl.conf.TopicConsumerConfigurationData;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
@@ -338,4 +342,18 @@ public void testStartPaused() {
         consumerBuilderImpl.startPaused(true);
         verify(consumerBuilderImpl.getConf()).setStartPaused(true);
     }
+
+    @Test
+    public void testTopicConsumerBuilder() {
+        List<TopicConsumerConfigurationData> topicConsumerConfigurationDataList = new ArrayList<>();
+        when(consumerBuilderImpl.getConf().getTopicConfigurations()).thenReturn(topicConsumerConfigurationDataList);
+
+        ConsumerBuilder<?> consumerBuilder = consumerBuilderImpl.topicConfiguration(Pattern.compile("foo")).priorityLevel(1).build();
+
+        assertThat(consumerBuilder).isSameAs(consumerBuilderImpl);
+        assertThat(topicConsumerConfigurationDataList).hasSize(1);
+        TopicConsumerConfigurationData topicConsumerConfigurationData = topicConsumerConfigurationDataList.get(0);
+        assertThat(topicConsumerConfigurationData.getTopicNameMatcher().matches("foo")).isTrue();
+        assertThat(topicConsumerConfigurationData.getPriorityLevel()).isEqualTo(1);
+    }
 }