Skip to content

Commit 495de95

Browse files
onobconobc
authored
Allow multiple customizers for several components (#435)
* DefaultPulsarReaderFactory accepts multiple customizers * DefaultPulsarConsumerFactory accepts multiple customizers * DefaultReactivePulsarSenderFactory accepts multiple customizers See #432
1 parent 0b9baf8 commit 495de95

17 files changed

+237
-82
lines changed

‎spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java‎

Lines changed: 34 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -55,23 +55,49 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
5555
@Nullable
5656
private final ReactiveMessageSenderCache reactiveMessageSenderCache;
5757

58+
@Nullable
59+
private final List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers;
60+
5861
private TopicResolver topicResolver;
5962

63+
/**
64+
* Construct an instance.
65+
* @param pulsarClient the pulsar client to adapt into a reactive client
66+
* @param reactiveMessageSenderSpec spec that defines the initial settings on the
67+
* created senders
68+
* @param reactiveMessageSenderCache cache used to cache created senders
69+
* @param defaultSenderBuilderCustomizers optional list of sender builder customizers
70+
* to apply to the created senders
71+
*/
6072
public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient,
6173
@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
62-
@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache) {
74+
@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache,
75+
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers) {
6376
this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec,
64-
reactiveMessageSenderCache, new DefaultTopicResolver());
77+
reactiveMessageSenderCache, defaultSenderBuilderCustomizers, new DefaultTopicResolver());
6578
}
6679

80+
/**
81+
* Construct an instance.
82+
* @param reactivePulsarClient the reactive client to use
83+
* @param reactiveMessageSenderSpec spec that defines the initial settings on the
84+
* created senders
85+
* @param reactiveMessageSenderCache cache used to cache created senders
86+
* @param defaultSenderBuilderCustomizers optional list of sender builder customizers
87+
* to apply to the created senders
88+
* @param topicResolver the topic resolver to use
89+
*/
6790
public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
6891
@Nullable ReactiveMessageSenderSpec reactiveMessageSenderSpec,
69-
@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache, TopicResolver topicResolver) {
92+
@Nullable ReactiveMessageSenderCache reactiveMessageSenderCache,
93+
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> defaultSenderBuilderCustomizers,
94+
TopicResolver topicResolver) {
7095
this.reactivePulsarClient = reactivePulsarClient;
7196
this.reactiveMessageSenderSpec = new ImmutableReactiveMessageSenderSpec(
7297
reactiveMessageSenderSpec != null ? reactiveMessageSenderSpec : new MutableReactiveMessageSenderSpec());
7398
this.reactiveMessageSenderCache = reactiveMessageSenderCache;
7499
this.topicResolver = topicResolver;
100+
this.defaultSenderBuilderCustomizers = defaultSenderBuilderCustomizers;
75101
}
76102

77103
@Override
@@ -102,6 +128,11 @@ private ReactiveMessageSender<T> doCreateReactiveMessageSender(Schema<T> schema,
102128

103129
ReactiveMessageSenderBuilder<T> sender = this.reactivePulsarClient.messageSender(schema);
104130
sender.applySpec(this.reactiveMessageSenderSpec);
131+
132+
// Apply the default config customizer (preserve the topic)
133+
if (!CollectionUtils.isEmpty(this.defaultSenderBuilderCustomizers)) {
134+
this.defaultSenderBuilderCustomizers.forEach((customizer -> customizer.customize(sender)));
135+
}
105136
sender.topic(resolvedTopic);
106137
if (this.reactiveMessageSenderCache != null) {
107138
sender.cache(this.reactiveMessageSenderCache);

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageConsumerFactoryTests.java renamed to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarConsumerFactoryTests.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,7 +37,7 @@
3737
* @author Christophe Bornet
3838
* @author Chris Bono
3939
*/
40-
class DefaultReactiveMessageConsumerFactoryTests {
40+
class DefaultReactivePulsarConsumerFactoryTests {
4141

4242
private static final Schema<String> SCHEMA = Schema.STRING;
4343

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageReaderFactoryTests.java renamed to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarReaderFactoryTests.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,9 @@
3333
* Tests for {@link DefaultReactivePulsarReaderFactory}.
3434
*
3535
* @author Christophe Bornet
36+
* @author Chris Bono
3637
*/
37-
class DefaultReactiveMessageReaderFactoryTests {
38+
class DefaultReactivePulsarReaderFactoryTests {
3839

3940
private static final Schema<String> schema = Schema.STRING;
4041

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactiveMessageSenderFactoryTests.java renamed to spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java

Lines changed: 50 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,30 +19,35 @@
1919
import static org.assertj.core.api.Assertions.assertThat;
2020
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
2121
import static org.assertj.core.api.Assertions.assertThatNullPointerException;
22+
import static org.mockito.ArgumentMatchers.any;
23+
import static org.mockito.Mockito.inOrder;
24+
import static org.mockito.Mockito.mock;
2225

2326
import java.util.Arrays;
2427
import java.util.Collections;
28+
import java.util.List;
2529

2630
import org.apache.pulsar.client.api.CompressionType;
27-
import org.apache.pulsar.client.api.PulsarClient;
2831
import org.apache.pulsar.client.api.Schema;
2932
import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
3033
import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
3134
import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
35+
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
3236
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
3337
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
3438
import org.assertj.core.api.InstanceOfAssertFactories;
3539
import org.assertj.core.api.ThrowingConsumer;
3640
import org.junit.jupiter.api.Nested;
3741
import org.junit.jupiter.api.Test;
42+
import org.mockito.InOrder;
3843

3944
/**
4045
* Unit tests for {@link DefaultReactivePulsarSenderFactory}.
4146
*
4247
* @author Christophe Bornet
4348
* @author Chris Bono
4449
*/
45-
class DefaultReactiveMessageSenderFactoryTests {
50+
class DefaultReactivePulsarSenderFactoryTests {
4651

4752
protected final Schema<String> schema = Schema.STRING;
4853

@@ -66,17 +71,17 @@ private void assertThatSenderSpecSatisfies(ReactiveMessageSender<String> sender,
6671
}
6772

6873
private ReactivePulsarSenderFactory<String> newSenderFactory() {
69-
return new DefaultReactivePulsarSenderFactory<>((PulsarClient) null, null, null);
74+
return new DefaultReactivePulsarSenderFactory<>(null, null, null, null);
7075
}
7176

7277
private ReactivePulsarSenderFactory<String> newSenderFactoryWithDefaultTopic(String defaultTopic) {
7378
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
7479
senderSpec.setTopicName(defaultTopic);
75-
return new DefaultReactivePulsarSenderFactory<>((PulsarClient) null, senderSpec, null);
80+
return new DefaultReactivePulsarSenderFactory<>(null, senderSpec, null, null);
7681
}
7782

7883
private ReactivePulsarSenderFactory<String> newSenderFactoryWithCache(ReactiveMessageSenderCache cache) {
79-
return new DefaultReactivePulsarSenderFactory<>((PulsarClient) null, null, cache);
84+
return new DefaultReactivePulsarSenderFactory<>(null, null, cache, null);
8085
}
8186

8287
@Nested
@@ -150,4 +155,44 @@ void customizerThatSetsTopicHasNoEffect() {
150155

151156
}
152157

158+
@Nested
159+
@SuppressWarnings("unchecked")
160+
class DefaultConfigCustomizerApi {
161+
162+
private ReactiveMessageSenderBuilderCustomizer<String> configCustomizer1 = mock(
163+
ReactiveMessageSenderBuilderCustomizer.class);
164+
165+
private ReactiveMessageSenderBuilderCustomizer<String> configCustomizer2 = mock(
166+
ReactiveMessageSenderBuilderCustomizer.class);
167+
168+
private ReactiveMessageSenderBuilderCustomizer<String> createSenderCustomizer = mock(
169+
ReactiveMessageSenderBuilderCustomizer.class);
170+
171+
@Test
172+
void singleConfigCustomizer() {
173+
newSenderFactoryWithCustomizers(List.of(configCustomizer1)).createSender(schema, "topic1",
174+
List.of(createSenderCustomizer));
175+
InOrder inOrder = inOrder(configCustomizer1, createSenderCustomizer);
176+
inOrder.verify(configCustomizer1).customize(any(ReactiveMessageSenderBuilder.class));
177+
inOrder.verify(createSenderCustomizer).customize(any(ReactiveMessageSenderBuilder.class));
178+
}
179+
180+
@Test
181+
void multipleConfigCustomizers() {
182+
newSenderFactoryWithCustomizers(List.of(configCustomizer2, configCustomizer1)).createSender(schema,
183+
"topic1", List.of(createSenderCustomizer));
184+
InOrder inOrder = inOrder(configCustomizer1, configCustomizer2, createSenderCustomizer);
185+
inOrder.verify(configCustomizer2).customize(any(ReactiveMessageSenderBuilder.class));
186+
inOrder.verify(configCustomizer1).customize(any(ReactiveMessageSenderBuilder.class));
187+
inOrder.verify(createSenderCustomizer).customize(any(ReactiveMessageSenderBuilder.class));
188+
}
189+
190+
private ReactivePulsarSenderFactory<String> newSenderFactoryWithCustomizers(
191+
List<ReactiveMessageSenderBuilderCustomizer<String>> customizers) {
192+
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
193+
return new DefaultReactivePulsarSenderFactory<>(null, senderSpec, null, customizers);
194+
}
195+
196+
}
197+
153198
}

‎spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java‎

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,8 @@ void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaul
182182
if (producerFactoryHasDefaultTopic) {
183183
spec.setTopicName("fake-topic");
184184
}
185-
ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client, spec, null);
185+
ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client, spec, null,
186+
null);
186187
// Topic mappings allows not specifying the topic when sending (nor having
187188
// default on producer)
188189
DefaultTopicResolver topicResolver = new DefaultTopicResolver();
@@ -198,7 +199,7 @@ void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaul
198199
@Test
199200
void sendMessageWithoutTopicFails() {
200201
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
201-
new MutableReactiveMessageSenderSpec(), null);
202+
new MutableReactiveMessageSenderSpec(), null, null);
202203
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
203204
assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send("test-message").subscribe())
204205
.withMessage("Topic must be specified when no default topic is configured");
@@ -211,7 +212,7 @@ private <T> Message<T> sendAndConsume(Consumer<ReactivePulsarTemplate<T>> sendFu
211212
senderSpec.setTopicName(topic);
212213
}
213214
ReactivePulsarSenderFactory<T> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, senderSpec,
214-
null);
215+
null, null);
215216

216217
ReactivePulsarTemplate<T> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
217218

@@ -260,7 +261,7 @@ void withSchemaInferredByTypeMappings() throws Exception {
260261
MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
261262
spec.setTopicName(topic);
262263
ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client, spec,
263-
null);
264+
null, null);
264265
// Custom schema resolver allows not specifying the schema when sending
265266
DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
266267
schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class));
@@ -282,7 +283,7 @@ void sendNullWithDefaultTopicFails() {
282283
MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
283284
spec.setTopicName("sendNullWithDefaultTopicFails");
284285
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, spec,
285-
null);
286+
null, null);
286287
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
287288
assertThatIllegalArgumentException()
288289
.isThrownBy(() -> pulsarTemplate.send((String) null, Schema.STRING).subscribe())
@@ -292,7 +293,7 @@ void sendNullWithDefaultTopicFails() {
292293
@Test
293294
void sendNullWithoutSchemaFails() {
294295
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
295-
new MutableReactiveMessageSenderSpec(), null);
296+
new MutableReactiveMessageSenderSpec(), null, null);
296297
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
297298
assertThatIllegalArgumentException()
298299
.isThrownBy(() -> pulsarTemplate.send("sendNullWithoutSchemaFails", (String) null, null).subscribe())

‎spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -82,7 +82,7 @@ void messageHandlerListener() throws Exception {
8282
MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
8383
prodConfig.setTopicName(topic);
8484
DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
85-
reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
85+
reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
8686
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
8787
pulsarTemplate.send("hello john doe").subscribe();
8888
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -116,7 +116,7 @@ void streamingHandlerListener() throws Exception {
116116
MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
117117
prodConfig.setTopicName(topic);
118118
DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
119-
reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
119+
reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
120120
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
121121
Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)).as(pulsarTemplate::send).subscribe();
122122
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -157,7 +157,7 @@ void containerProperties() throws Exception {
157157
MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
158158
prodConfig.setTopicName(topic);
159159
DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
160-
reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
160+
reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
161161
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
162162
pulsarTemplate.send("hello john doe").subscribe();
163163
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -271,7 +271,7 @@ void containerTopicsPattern() throws Exception {
271271
MutableReactiveMessageSenderSpec prodConfig = new MutableReactiveMessageSenderSpec();
272272
prodConfig.setTopicName(topic);
273273
DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
274-
reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
274+
reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
275275
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
276276
pulsarTemplate.send("hello john doe").subscribe();
277277
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -321,7 +321,7 @@ void consumerCustomizer() throws Exception {
321321
prodConfig.setBatchingEnabled(false);
322322
prodConfig.setTopicName(topic);
323323
DefaultReactivePulsarSenderFactory<String> pulsarProducerFactory = new DefaultReactivePulsarSenderFactory<>(
324-
reactivePulsarClient, prodConfig, null, new DefaultTopicResolver());
324+
reactivePulsarClient, prodConfig, null, null, new DefaultTopicResolver());
325325
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(pulsarProducerFactory);
326326
Flux.range(0, 5).map(i -> MessageSpec.of("hello john doe" + i)).as(pulsarTemplate::send).subscribe();
327327
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();

‎spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarConsumerFactory.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -48,18 +48,18 @@ public class DefaultPulsarConsumerFactory<T> implements PulsarConsumerFactory<T>
4848
private final PulsarClient pulsarClient;
4949

5050
@Nullable
51-
private final ConsumerBuilderCustomizer<T> defaultConfigCustomizer;
51+
private final List<ConsumerBuilderCustomizer<T>> defaultConfigCustomizers;
5252

5353
/**
5454
* Construct a consumer factory instance.
5555
* @param pulsarClient the client used to consume
56-
* @param defaultConfigCustomizer the default configuration to apply to the consumers
57-
* or null to use no default configuration
56+
* @param defaultConfigCustomizers the optional list of customizers to apply to the
57+
* created consumers or null to use no default configuration
5858
*/
5959
public DefaultPulsarConsumerFactory(PulsarClient pulsarClient,
60-
ConsumerBuilderCustomizer<T> defaultConfigCustomizer) {
60+
List<ConsumerBuilderCustomizer<T>> defaultConfigCustomizers) {
6161
this.pulsarClient = pulsarClient;
62-
this.defaultConfigCustomizer = defaultConfigCustomizer;
62+
this.defaultConfigCustomizers = defaultConfigCustomizers;
6363
}
6464

6565
@Override
@@ -77,8 +77,8 @@ public Consumer<T> createConsumer(Schema<T> schema, @Nullable Collection<String>
7777
ConsumerBuilder<T> consumerBuilder = this.pulsarClient.newConsumer(schema);
7878

7979
// Apply the default config customizer (preserve the topic)
80-
if (this.defaultConfigCustomizer != null) {
81-
this.defaultConfigCustomizer.customize(consumerBuilder);
80+
if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
81+
this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(consumerBuilder)));
8282
}
8383
if (topics != null) {
8484
replaceTopicsOnBuilder(consumerBuilder, topics);

‎spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ protected Producer<T> doCreateProducer(Schema<T> schema, @Nullable String topic,
142142
var producerBuilder = this.pulsarClient.newProducer(schema);
143143

144144
// Apply the default config customizer (preserve the topic)
145-
if (this.defaultConfigCustomizers != null) {
145+
if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
146146
this.defaultConfigCustomizers.forEach((customizer) -> customizer.customize(producerBuilder));
147147
}
148148
producerBuilder.topic(resolvedTopic);

‎spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ public class DefaultPulsarReaderFactory<T> implements PulsarReaderFactory<T> {
4343
private final PulsarClient pulsarClient;
4444

4545
@Nullable
46-
private final ReaderBuilderCustomizer<T> defaultConfigCustomizer;
46+
private final List<ReaderBuilderCustomizer<T>> defaultConfigCustomizers;
4747

4848
/**
4949
* Construct a reader factory instance with no default configuration.
@@ -56,13 +56,13 @@ public DefaultPulsarReaderFactory(PulsarClient pulsarClient) {
5656
/**
5757
* Construct a reader factory instance.
5858
* @param pulsarClient the client used to consume
59-
* @param defaultConfigCustomizer the default configuration to apply to the readers or
60-
* null to use no default configuration
59+
* @param defaultConfigCustomizers the optional list of customizers to apply to the
60+
* readers or null to use no default configuration
6161
*/
6262
public DefaultPulsarReaderFactory(PulsarClient pulsarClient,
63-
@Nullable ReaderBuilderCustomizer<T> defaultConfigCustomizer) {
63+
@Nullable List<ReaderBuilderCustomizer<T>> defaultConfigCustomizers) {
6464
this.pulsarClient = pulsarClient;
65-
this.defaultConfigCustomizer = defaultConfigCustomizer;
65+
this.defaultConfigCustomizers = defaultConfigCustomizers;
6666
}
6767

6868
@Override
@@ -72,8 +72,8 @@ public Reader<T> createReader(@Nullable List<String> topics, @Nullable MessageId
7272
ReaderBuilder<T> readerBuilder = this.pulsarClient.newReader(schema);
7373

7474
// Apply the default config customizer (preserve the topics)
75-
if (this.defaultConfigCustomizer != null) {
76-
this.defaultConfigCustomizer.customize(readerBuilder);
75+
if (!CollectionUtils.isEmpty(this.defaultConfigCustomizers)) {
76+
this.defaultConfigCustomizers.forEach((customizer -> customizer.customize(readerBuilder)));
7777
}
7878

7979
if (!CollectionUtils.isEmpty(topics)) {

‎spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -358,11 +358,11 @@ void messagesAreProperlyAckdOnContainerStopBeforeExitingListenerThread() throws
358358
pulsarClient.close();
359359
}
360360

361-
private <T> ConsumerBuilderCustomizer<T> defaultConfig(String topicName, String subscriptionName) {
362-
return (consumerBuilder) -> {
361+
private <T> List<ConsumerBuilderCustomizer<T>> defaultConfig(String topicName, String subscriptionName) {
362+
return List.of((consumerBuilder) -> {
363363
consumerBuilder.topic(topicName);
364364
consumerBuilder.subscriptionName(subscriptionName);
365-
};
365+
});
366366
}
367367

368368
}

0 commit comments

Comments
 (0)