Skip to content

Add support for configuring Pulsar listener container concurrency #42062

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

Add support for configuring Pulsar listener container concurrency #42062

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion .../src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarAutoConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@
* @author Alexander Preuß
* @author Phillip Webb
* @author Jonas Geiregat
* @author Vedran Pavic
* @since 3.2.0
*/
@AutoConfiguration
Expand Down Expand Up @@ -187,7 +188,10 @@ ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
}
pulsarTransactionManager.ifUnique(containerProperties.transactions()::setTransactionManager);
this.propertiesMapper.customizeContainerProperties(containerProperties);
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProperties);
ConcurrentPulsarListenerContainerFactory<Object> listenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>(
pulsarConsumerFactory, containerProperties);
this.propertiesMapper.customizeConcurrentPulsarListenerContainerFactory(listenerContainerFactory);
return listenerContainerFactory;
}

@Bean
Expand Down
13 changes: 13 additions & 0 deletions ...nfigure/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -811,6 +811,11 @@ public static class Listener {
*/
private SchemaType schemaType;

/**
* Number of threads used by listener container.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It may be helpful to elaborate on this one just a bit, maybe something like:

Number of listener threads used by the container to process messages concurrently

Wdyt?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I tried to aligned property description with these:

As a user I'd probably prefer consistency in describing the same concept in different parts of auto-configuration, but whatever Spring Boot team prefers here is fine with me.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

tried to aligned property description with these

Yep, I saw those descriptions as well. I think there is room for improvement in the spring-kafka properties. This is really a nit and consistency is a good thing. As you said, let's see what Boot peepz think.

*/
private Integer concurrency;

/**
* Whether to record observations for when the Observations API is available and
* the client supports it.
Expand All @@ -825,6 +830,14 @@ public void setSchemaType(SchemaType schemaType) {
this.schemaType = schemaType;
}

public Integer getConcurrency() {
return this.concurrency;
}

public void setConcurrency(Integer concurrency) {
this.concurrency = concurrency;
}

public boolean isObservationEnabled() {
return this.observationEnabled;
}
Expand Down
8 changes: 8 additions & 0 deletions ...e/src/main/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.boot.json.JsonWriter;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
Expand Down Expand Up @@ -198,6 +199,13 @@ private void customizePulsarContainerListenerProperties(PulsarContainerPropertie
map.from(properties::isObservationEnabled).to(containerProperties::setObservationEnabled);
}

<T> void customizeConcurrentPulsarListenerContainerFactory(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just noticed that the listener containers on the framework side are inconsistent w/ the concurrency attribute between reactive and imperative. I created this issue to simplify the latter.
If you wanted to handle that issue first it would then simplify this PR so that we would not need this new customizeConcurrentPulsarListenerContainerFactory method and could simply map the property in customizePulsarContainerListenerProperties instead.

Copy link
Contributor Author

@vpavic vpavic Sep 2, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, I can pick up that issue on Spring Pulsar side. So basically this PR is then blocked until spring-projects/spring-pulsar#820 is resolved.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, I can pick up that issue on Spring Pulsar side. So basically this PR is then blocker until spring-projects/spring-pulsar#820 is resolved.

Thanks @vpavic . Yeh, I think we should do the spring-pulsar one first.

ConcurrentPulsarListenerContainerFactory<T> listenerContainerFactory) {
PulsarProperties.Listener properties = this.properties.getListener();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getConcurrency).to(listenerContainerFactory::setConcurrency);
}

<T> void customizeReaderBuilder(ReaderBuilder<T> readerBuilder) {
PulsarProperties.Reader properties = this.properties.getReader();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
Expand Down
4 changes: 3 additions & 1 deletion ...in/java/org/springframework/boot/autoconfigure/pulsar/PulsarReactivePropertiesMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2012-2023 the original author or authors.
* Copyright 2012-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -31,6 +31,7 @@
*
* @author Chris Bono
* @author Phillip Webb
* @author Vedran Pavic
*/
final class PulsarReactivePropertiesMapper {

Expand Down Expand Up @@ -93,6 +94,7 @@ private void customizePulsarContainerListenerProperties(ReactivePulsarContainerP
PulsarProperties.Listener properties = this.properties.getListener();
PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
map.from(properties::getSchemaType).to(containerProperties::setSchemaType);
map.from(properties::getConcurrency).to(containerProperties::setConcurrency);
}

void customizeMessageReaderBuilder(ReactiveMessageReaderBuilder<?> builder) {
Expand Down
12 changes: 12 additions & 0 deletions .../test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesMapperTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@

import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Failover.BackupCluster;
import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory;
import org.springframework.pulsar.core.PulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.listener.PulsarContainerProperties;
Expand Down Expand Up @@ -272,6 +273,17 @@ void customizeContainerProperties() {
assertThat(containerProperties.transactions().isEnabled()).isTrue();
}

@Test
void customizeConcurrentPulsarListenerContainerFactory() {
PulsarProperties properties = new PulsarProperties();
properties.getListener().setConcurrency(10);
ConcurrentPulsarListenerContainerFactory<?> listenerContainerFactory = mock(
ConcurrentPulsarListenerContainerFactory.class);
new PulsarPropertiesMapper(properties)
.customizeConcurrentPulsarListenerContainerFactory(listenerContainerFactory);
then(listenerContainerFactory).should().setConcurrency(10);
}

@Test
@SuppressWarnings("unchecked")
void customizeReaderBuilder() {
Expand Down
2 changes: 2 additions & 0 deletions ...re/src/test/java/org/springframework/boot/autoconfigure/pulsar/PulsarPropertiesTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,9 +393,11 @@ class ListenerProperties {
void bind() {
Map<String, String> map = new HashMap<>();
map.put("spring.pulsar.listener.schema-type", "avro");
map.put("spring.pulsar.listener.concurrency", "10");
map.put("spring.pulsar.listener.observation-enabled", "true");
PulsarProperties.Listener properties = bindProperties(map).getListener();
assertThat(properties.getSchemaType()).isEqualTo(SchemaType.AVRO);
assertThat(properties.getConcurrency()).isEqualTo(10);
assertThat(properties.isObservationEnabled()).isTrue();
}

Expand Down
5 changes: 4 additions & 1 deletion ...va/org/springframework/boot/autoconfigure/pulsar/PulsarReactivePropertiesMapperTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2012-2023 the original author or authors.
* Copyright 2012-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -48,6 +48,7 @@
*
* @author Chris Bono
* @author Phillip Webb
* @author Vedran Pavic
*/
class PulsarReactivePropertiesMapperTests {

Expand Down Expand Up @@ -120,10 +121,12 @@ void customizeContainerProperties() {
PulsarProperties properties = new PulsarProperties();
properties.getConsumer().getSubscription().setType(SubscriptionType.Shared);
properties.getListener().setSchemaType(SchemaType.AVRO);
properties.getListener().setConcurrency(10);
ReactivePulsarContainerProperties<Object> containerProperties = new ReactivePulsarContainerProperties<>();
new PulsarReactivePropertiesMapper(properties).customizeContainerProperties(containerProperties);
assertThat(containerProperties.getSubscriptionType()).isEqualTo(SubscriptionType.Shared);
assertThat(containerProperties.getSchemaType()).isEqualTo(SchemaType.AVRO);
assertThat(containerProperties.getConcurrency()).isEqualTo(10);
}

@Test
Expand Down