|
17 | 17 | package org.springframework.pulsar.core; |
18 | 18 |
|
19 | 19 | import java.util.ArrayList; |
20 | | -import java.util.HashMap; |
| 20 | +import java.util.Collection; |
| 21 | +import java.util.Collections; |
| 22 | +import java.util.HashSet; |
21 | 23 | import java.util.List; |
22 | 24 | import java.util.Map; |
| 25 | +import java.util.TreeMap; |
| 26 | +import java.util.concurrent.TimeUnit; |
23 | 27 |
|
24 | | -import org.apache.pulsar.client.api.BatchReceivePolicy; |
25 | 28 | import org.apache.pulsar.client.api.Consumer; |
26 | 29 | import org.apache.pulsar.client.api.ConsumerBuilder; |
27 | | -import org.apache.pulsar.client.api.DeadLetterPolicy; |
28 | 30 | import org.apache.pulsar.client.api.PulsarClient; |
29 | 31 | import org.apache.pulsar.client.api.PulsarClientException; |
30 | | -import org.apache.pulsar.client.api.RedeliveryBackoff; |
31 | 32 | import org.apache.pulsar.client.api.Schema; |
| 33 | +import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; |
32 | 34 |
|
| 35 | +import org.springframework.lang.Nullable; |
33 | 36 | import org.springframework.util.CollectionUtils; |
34 | 37 |
|
35 | 38 | /** |
|
38 | 41 | * @param <T> underlying payload type for the consumer. |
39 | 42 | * @author Soby Chacko |
40 | 43 | * @author Alexander Preuß |
| 44 | + * @author Christophe Bornet |
41 | 45 | */ |
42 | 46 | public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T> { |
43 | 47 |
|
44 | | - private final Map<String, Object> consumerConfig = new HashMap<>(); |
| 48 | + private final ConsumerConfigurationData<T> consumerConfig; |
45 | 49 |
|
46 | 50 | private final List<Consumer<T>> consumers = new ArrayList<>(); |
47 | 51 |
|
48 | | - private PulsarClient pulsarClient; |
| 52 | + private final PulsarClient pulsarClient; |
49 | 53 |
|
50 | | - public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, Map<String, Object> consumerConfig) { |
| 54 | + public DefaultPulsarConsumerFactory(PulsarClient pulsarClient, ConsumerConfigurationData<T> consumerConfig) { |
51 | 55 | this.pulsarClient = pulsarClient; |
52 | | - if (!CollectionUtils.isEmpty(consumerConfig)) { |
53 | | - this.consumerConfig.putAll(consumerConfig); |
54 | | - } |
| 56 | + this.consumerConfig = consumerConfig != null ? consumerConfig : new ConsumerConfigurationData<>(); |
55 | 57 | } |
56 | 58 |
|
57 | 59 | @Override |
58 | | - public Consumer<T> createConsumer(Schema<T> schema, Map<String, Object> propertiesToOverride) |
59 | | - throws PulsarClientException { |
60 | | - |
61 | | - final ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema); |
| 60 | + public Consumer<T> createConsumer(Schema<T> schema) throws PulsarClientException { |
| 61 | + return createConsumer(schema, Collections.emptyList()); |
| 62 | + } |
62 | 63 |
|
63 | | - final Map<String, Object> properties = new HashMap<>(this.consumerConfig); |
64 | | - properties.putAll(propertiesToOverride); |
| 64 | + @Override |
| 65 | + public Consumer<T> createConsumer(Schema<T> schema, List<ConsumerBuilderCustomizer<T>> customizers) |
| 66 | + throws PulsarClientException { |
65 | 67 |
|
66 | | - if (!CollectionUtils.isEmpty(properties)) { |
67 | | - consumerBuilder.loadConf(properties); |
68 | | - } |
69 | | - Consumer<T> consumer = consumerBuilder.subscribe(); |
70 | | - this.consumers.add(consumer); |
71 | | - return consumer; |
| 68 | + return createConsumer(schema, null, null, Collections.emptyList()); |
72 | 69 | } |
73 | 70 |
|
74 | 71 | @Override |
75 | | - public Consumer<T> createConsumer(Schema<T> schema, BatchReceivePolicy batchReceivePolicy, |
76 | | - Map<String, Object> propertiesToOverride) throws PulsarClientException { |
77 | | - |
| 72 | + public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String> topics, |
| 73 | + @Nullable Map<String, String> properties, @Nullable List<ConsumerBuilderCustomizer<T>> customizers) |
| 74 | + throws PulsarClientException { |
78 | 75 | final ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema); |
79 | | - final Map<String, Object> properties = new HashMap<>(this.consumerConfig); |
80 | | - properties.putAll(propertiesToOverride); |
81 | 76 |
|
82 | | - // Remove deadLetterPolicy from the properties here and save it to re-apply after |
83 | | - // calling `loadConf` (https://github.com/apache/pulsar/issues/11646) |
84 | | - DeadLetterPolicy deadLetterPolicy = null; |
85 | | - if (properties.containsKey("deadLetterPolicy")) { |
86 | | - deadLetterPolicy = (DeadLetterPolicy) properties.remove("deadLetterPolicy"); |
| 77 | + ConsumerConfigurationData<T> config = this.consumerConfig.clone(); |
| 78 | + if (topics != null) { |
| 79 | + config.setTopicNames(new HashSet<>(topics)); |
87 | 80 | } |
88 | | - |
89 | | - if (!CollectionUtils.isEmpty(properties)) { |
90 | | - consumerBuilder.loadConf(properties); |
91 | | - } |
92 | | - |
93 | | - if (deadLetterPolicy != null) { |
94 | | - consumerBuilder.deadLetterPolicy(deadLetterPolicy); |
| 81 | + if (properties != null) { |
| 82 | + config.setProperties(new TreeMap<>(properties)); |
95 | 83 | } |
96 | 84 |
|
97 | | - if (properties.containsKey("negativeAckRedeliveryBackoff")) { |
98 | | - final RedeliveryBackoff negativeAckRedeliveryBackoff = (RedeliveryBackoff) properties |
99 | | - .get("negativeAckRedeliveryBackoff"); |
100 | | - consumerBuilder.negativeAckRedeliveryBackoff(negativeAckRedeliveryBackoff); |
101 | | - } |
| 85 | + loadConf(consumerBuilder, config); |
102 | 86 |
|
103 | | - if (properties.containsKey("ackTimeoutRedeliveryBackoff")) { |
104 | | - final RedeliveryBackoff ackTimeoutRedeliveryBackoff = (RedeliveryBackoff) properties |
105 | | - .get("ackTimeoutRedeliveryBackoff"); |
106 | | - consumerBuilder.ackTimeoutRedeliveryBackoff(ackTimeoutRedeliveryBackoff); |
| 87 | + if (!CollectionUtils.isEmpty(customizers)) { |
| 88 | + customizers.forEach(customizer -> customizer.customize(consumerBuilder)); |
107 | 89 | } |
108 | 90 |
|
109 | | - consumerBuilder.batchReceivePolicy(batchReceivePolicy); |
110 | 91 | Consumer<T> consumer = consumerBuilder.subscribe(); |
111 | 92 | this.consumers.add(consumer); |
112 | 93 | return consumer; |
113 | 94 | } |
114 | 95 |
|
115 | | - public Map<String, Object> getConsumerConfig() { |
| 96 | + public ConsumerConfigurationData<T> getConsumerConfig() { |
116 | 97 | return this.consumerConfig; |
117 | 98 | } |
118 | 99 |
|
| 100 | + private static <T> void loadConf(ConsumerBuilder<T> consumer, ConsumerConfigurationData<T> conf) { |
| 101 | + if (!CollectionUtils.isEmpty(conf.getTopicNames())) { |
| 102 | + consumer.topics(conf.getTopicNames().stream().toList()); |
| 103 | + } |
| 104 | + if (conf.getTopicsPattern() != null) { |
| 105 | + consumer.topicsPattern(conf.getTopicsPattern()); |
| 106 | + } |
| 107 | + if (conf.getSubscriptionName() != null) { |
| 108 | + consumer.subscriptionName(conf.getSubscriptionName()); |
| 109 | + } |
| 110 | + if (conf.getSubscriptionType() != null) { |
| 111 | + consumer.subscriptionType(conf.getSubscriptionType()); |
| 112 | + } |
| 113 | + if (!CollectionUtils.isEmpty(conf.getSubscriptionProperties())) { |
| 114 | + consumer.subscriptionProperties(conf.getSubscriptionProperties()); |
| 115 | + } |
| 116 | + if (conf.getSubscriptionMode() != null) { |
| 117 | + consumer.subscriptionMode(conf.getSubscriptionMode()); |
| 118 | + } |
| 119 | + if (conf.getMessageListener() != null) { |
| 120 | + consumer.messageListener(conf.getMessageListener()); |
| 121 | + } |
| 122 | + if (conf.getConsumerEventListener() != null) { |
| 123 | + consumer.consumerEventListener(conf.getConsumerEventListener()); |
| 124 | + } |
| 125 | + if (conf.getNegativeAckRedeliveryBackoff() != null) { |
| 126 | + consumer.negativeAckRedeliveryBackoff(conf.getNegativeAckRedeliveryBackoff()); |
| 127 | + } |
| 128 | + if (conf.getAckTimeoutRedeliveryBackoff() != null) { |
| 129 | + consumer.ackTimeoutRedeliveryBackoff(conf.getAckTimeoutRedeliveryBackoff()); |
| 130 | + } |
| 131 | + if (conf.getReceiverQueueSize() >= 0) { |
| 132 | + consumer.receiverQueueSize(conf.getReceiverQueueSize()); |
| 133 | + } |
| 134 | + if (conf.getAcknowledgementsGroupTimeMicros() >= 0) { |
| 135 | + consumer.acknowledgmentGroupTime(conf.getAcknowledgementsGroupTimeMicros(), TimeUnit.MICROSECONDS); |
| 136 | + } |
| 137 | + if (conf.getNegativeAckRedeliveryDelayMicros() >= 0) { |
| 138 | + consumer.negativeAckRedeliveryDelay(conf.getNegativeAckRedeliveryDelayMicros(), TimeUnit.MICROSECONDS); |
| 139 | + } |
| 140 | + if (conf.getMaxTotalReceiverQueueSizeAcrossPartitions() >= 0) { |
| 141 | + consumer.maxTotalReceiverQueueSizeAcrossPartitions(conf.getMaxTotalReceiverQueueSizeAcrossPartitions()); |
| 142 | + } |
| 143 | + if (conf.getConsumerName() != null) { |
| 144 | + consumer.consumerName(conf.getConsumerName()); |
| 145 | + } |
| 146 | + if (conf.getAckTimeoutMillis() >= 0) { |
| 147 | + consumer.ackTimeout(conf.getAckTimeoutMillis(), TimeUnit.MILLISECONDS); |
| 148 | + } |
| 149 | + if (conf.getTickDurationMillis() >= 0) { |
| 150 | + consumer.ackTimeoutTickTime(conf.getTickDurationMillis(), TimeUnit.MILLISECONDS); |
| 151 | + } |
| 152 | + if (conf.getPriorityLevel() >= 0) { |
| 153 | + consumer.priorityLevel(conf.getPriorityLevel()); |
| 154 | + } |
| 155 | + if (conf.getMaxPendingChunkedMessage() >= 0) { |
| 156 | + consumer.maxPendingChunkedMessage(conf.getMaxPendingChunkedMessage()); |
| 157 | + } |
| 158 | + consumer.autoAckOldestChunkedMessageOnQueueFull(conf.isAutoAckOldestChunkedMessageOnQueueFull()); |
| 159 | + if (conf.getExpireTimeOfIncompleteChunkedMessageMillis() >= 0) { |
| 160 | + consumer.expireTimeOfIncompleteChunkedMessage(conf.getExpireTimeOfIncompleteChunkedMessageMillis(), |
| 161 | + TimeUnit.MILLISECONDS); |
| 162 | + } |
| 163 | + if (conf.getCryptoKeyReader() != null) { |
| 164 | + consumer.cryptoKeyReader(conf.getCryptoKeyReader()); |
| 165 | + } |
| 166 | + if (conf.getMessageCrypto() != null) { |
| 167 | + consumer.messageCrypto(conf.getMessageCrypto()); |
| 168 | + } |
| 169 | + if (conf.getCryptoFailureAction() != null) { |
| 170 | + consumer.cryptoFailureAction(conf.getCryptoFailureAction()); |
| 171 | + } |
| 172 | + if (!CollectionUtils.isEmpty(conf.getProperties())) { |
| 173 | + consumer.properties(conf.getProperties()); |
| 174 | + } |
| 175 | + consumer.readCompacted(conf.isReadCompacted()); |
| 176 | + if (conf.getSubscriptionInitialPosition() != null) { |
| 177 | + consumer.subscriptionInitialPosition(conf.getSubscriptionInitialPosition()); |
| 178 | + } |
| 179 | + if (conf.getPatternAutoDiscoveryPeriod() >= 0) { |
| 180 | + consumer.patternAutoDiscoveryPeriod(conf.getPatternAutoDiscoveryPeriod(), TimeUnit.SECONDS); |
| 181 | + } |
| 182 | + if (conf.getRegexSubscriptionMode() != null) { |
| 183 | + consumer.subscriptionTopicsMode(conf.getRegexSubscriptionMode()); |
| 184 | + } |
| 185 | + if (conf.getDeadLetterPolicy() != null) { |
| 186 | + consumer.deadLetterPolicy(conf.getDeadLetterPolicy()); |
| 187 | + } |
| 188 | + consumer.enableRetry(conf.isRetryEnable()); |
| 189 | + if (conf.getBatchReceivePolicy() != null) { |
| 190 | + consumer.batchReceivePolicy(conf.getBatchReceivePolicy()); |
| 191 | + } |
| 192 | + consumer.autoUpdatePartitions(conf.isAutoUpdatePartitions()); |
| 193 | + if (conf.getAutoUpdatePartitionsIntervalSeconds() >= 0) { |
| 194 | + consumer.autoUpdatePartitionsInterval((int) conf.getAutoUpdatePartitionsIntervalSeconds(), |
| 195 | + TimeUnit.SECONDS); |
| 196 | + } |
| 197 | + consumer.replicateSubscriptionState(conf.isReplicateSubscriptionState()); |
| 198 | + if (conf.isResetIncludeHead()) { |
| 199 | + consumer.startMessageIdInclusive(); |
| 200 | + } |
| 201 | + if (conf.getKeySharedPolicy() != null) { |
| 202 | + consumer.keySharedPolicy(conf.getKeySharedPolicy()); |
| 203 | + } |
| 204 | + consumer.enableBatchIndexAcknowledgment(conf.isBatchIndexAckEnabled()); |
| 205 | + consumer.isAckReceiptEnabled(conf.isAckReceiptEnabled()); |
| 206 | + consumer.poolMessages(conf.isPoolMessages()); |
| 207 | + if (conf.getPayloadProcessor() != null) { |
| 208 | + consumer.messagePayloadProcessor(conf.getPayloadProcessor()); |
| 209 | + } |
| 210 | + consumer.startPaused(conf.isStartPaused()); |
| 211 | + } |
| 212 | + |
119 | 213 | } |
0 commit comments