Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove subscription name from PulsarReader (#484)

Browse Source 
This commit removes the subscription name from all PulsarReader
areas. Readers do not have the concept of subscriptions and this was
just a holdover from the PulsarListener codebase.
This commit is contained in:
Chris Bono
2023-11-11 23:06:04 -06:00
committed by GitHub
parent 72c243892a
commit baecaaaa9e
10 changed files with 18 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar.adoc
View File

@@ -1086,22 +1086,20 @@ Spring Boot provides this reader factory which you can further configure by spec
==== PulsarReader Annotation
While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides a convenient annotation called `PulsarReader` that you can use to quickly read from a topic without setting up any reader factories yourselves.
While it is possible to use `PulsarReaderFactory` directly, Spring for Apache Pulsar provides the `PulsarReader` annotation that you can use to quickly read from a topic without setting up any reader factories yourselves.
This is similar to the same ideas behind `PulsarListener.`
Here is a quick example.
[source, java]
----
@PulsarReader(id = "pulsar-reader-demo-id", subscriptionName = "pulsar-reader-demo-subscription",
topics = "pulsar-reader-demo-topic", startMessageId = "earliest")
@PulsarReader(id = "reader-demo-id", topics = "reader-demo-topic", startMessageId = "earliest")
void read(String message) {
//...
}
----
As you can see from the above example, `PulsarReader` is a quick way to read from a Pulsar topic.
The `id` and `subscriptionName` attributes are optional, but always a best practice to provide them.
`PulsarReader` requires `topics` and the `startMessageId` as mandatory attributes.
The `id` attribute is optional, but it is a best practice to provide a value that is meaningful to your application.
When not specified an auto-generated id will be used.
On the other hand, the `topics` and `startMessageId` attributes are mandatory.
The `topics` attribute can be a single topic or a comma-separated list of topics.
The `startMessageId` attribute instructs the reader to start from a particular message in the topic.
The valid values for `startMessageId` are `earliest` or `latest.`
@@ -1114,8 +1112,8 @@ You can provide a `@Bean` from `ReaderBuilderCustomizer` and then make it availa
[source, java]
----
@PulsarReader(id = "with-customizer-reader", subscriptionName = "with-customizer-reader-subscription",
topics = "with-customizer-reader-topic", readerCustomizer = "myCustomizer")
@PulsarReader(id = "reader-customizer-demo-id", topics = "reader-customizer-demo-topic",
readerCustomizer = "myCustomizer")
void read(String message) {
//...
}

6
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReader.java
View File

@@ -46,12 +46,6 @@ public @interface PulsarReader {
*/
String id() default "";
/**
* Pulsar subscription name associated with this listener.
* @return the {@code subscriptionName} for this Pulsar listener endpoint.
*/
String subscriptionName() default "";
/**
* Pulsar schema type for this listener.
* @return the {@code schemaType} for this listener

8
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarReaderAnnotationBeanPostProcessor.java
View File

@@ -210,7 +210,6 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
Object bean, String[] topics) {
endpoint.setBean(bean);
endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
endpoint.setSubscriptionName(getEndpointSubscriptionName(pulsarReader));
endpoint.setId(getEndpointId(pulsarReader));
endpoint.setTopics(topics);
endpoint.setSchemaType(pulsarReader.schemaType());
@@ -248,13 +247,6 @@ public class PulsarReaderAnnotationBeanPostProcessor<V> extends AbstractPulsarAn
}
}
private String getEndpointSubscriptionName(PulsarReader pulsarReader) {
if (StringUtils.hasText(pulsarReader.subscriptionName())) {
return resolveExpressionAsString(pulsarReader.subscriptionName(), "subscriptionName");
}
return GENERATED_ID_PREFIX + this.counter.getAndIncrement();
}
private String getEndpointId(PulsarReader pulsarReader) {
if (StringUtils.hasText(pulsarReader.id())) {
return resolveExpressionAsString(pulsarReader.id(), "id");

6
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarReaderEndpoint.java
View File

@@ -109,12 +109,6 @@ public abstract class AbstractPulsarReaderEndpoint<K>
this.subscriptionName = subscriptionName;
}
@Nullable
@Override
public String getSubscriptionName() {
return this.subscriptionName;
}
public void setId(String id) {
this.id = id;
}

9
spring-pulsar/src/main/java/org/springframework/pulsar/config/DefaultPulsarReaderContainerFactory.java
View File

@@ -21,7 +21,6 @@ import org.springframework.pulsar.reader.DefaultPulsarMessageReaderContainer;
import org.springframework.pulsar.reader.PulsarMessageReaderContainer;
import org.springframework.pulsar.reader.PulsarReaderContainerProperties;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
* Concrete implementation for {@link PulsarReaderContainerFactory}.
@@ -40,21 +39,13 @@ public class DefaultPulsarReaderContainerFactory<T>
@Override
protected DefaultPulsarMessageReaderContainer<T> createContainerInstance(
PulsarReaderEndpoint<PulsarMessageReaderContainer> endpoint) {
PulsarReaderContainerProperties properties = new PulsarReaderContainerProperties();
properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver());
if (!CollectionUtils.isEmpty(endpoint.getTopics())) {
properties.setTopics(endpoint.getTopics());
}
if (StringUtils.hasText(endpoint.getSubscriptionName())) {
properties.setSubscriptionName(endpoint.getSubscriptionName());
}
properties.setSchemaType(endpoint.getSchemaType());
properties.setStartMessageId(endpoint.getStartMessageId());
return new DefaultPulsarMessageReaderContainer<>(this.getReaderFactory(), properties);
}

12
spring-pulsar/src/main/java/org/springframework/pulsar/config/GenericReaderEndpointRegistry.java
View File

@@ -121,16 +121,12 @@ public class GenericReaderEndpointRegistry<C extends PulsarMessageReaderContaine
boolean startImmediately) {
Assert.notNull(endpoint, "Endpoint must not be null");
Assert.notNull(factory, "Factory must not be null");
String subscriptionName = endpoint.getSubscriptionName();
String id = endpoint.getId();
Assert.hasText(subscriptionName, "Endpoint id must not be empty");
Assert.hasText(id, "Endpoint id must not be empty");
this.containersLock.lock();
try {
Assert.state(!this.readerContainers.containsKey(id),
"Another endpoint is already registered with id '" + subscriptionName + "'");
"Another endpoint is already registered with id '" + id + "'");
C container = createReaderContainer(endpoint, factory);
this.readerContainers.put(id, container);
}
@@ -140,9 +136,7 @@ public class GenericReaderEndpointRegistry<C extends PulsarMessageReaderContaine
}
protected C createReaderContainer(E endpoint, ReaderContainerFactory<? extends C, E> factory) {
C readerContainer = factory.createReaderContainer(endpoint);
if (readerContainer instanceof InitializingBean) {
try {
((InitializingBean) readerContainer).afterPropertiesSet();
@@ -151,7 +145,6 @@ public class GenericReaderEndpointRegistry<C extends PulsarMessageReaderContaine
throw new BeanInitializationException("Failed to initialize message listener container", ex);
}
}
int containerPhase = readerContainer.getPhase();
if (readerContainer.isAutoStartup() && containerPhase != C.DEFAULT_PHASE) {
if (this.phase != C.DEFAULT_PHASE && this.phase != containerPhase) {
@@ -160,7 +153,6 @@ public class GenericReaderEndpointRegistry<C extends PulsarMessageReaderContaine
}
this.phase = readerContainer.getPhase();
}
return readerContainer;
}

7
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpoint.java
View File

@@ -44,13 +44,6 @@ public interface PulsarReaderEndpoint<C extends PulsarMessageReaderContainer> {
@Nullable
String getId();
/**
* Return the subscription name for this endpoint's container.
* @return the subscription name.
*/
@Nullable
String getSubscriptionName();
/**
* Return the topics for this endpoint's container.
* @return the topics.

2
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarReaderEndpointRegistrar.java
View File

@@ -117,7 +117,7 @@ public class PulsarReaderEndpointRegistrar implements BeanFactoryAware, Initiali
public void registerEndpoint(PulsarReaderEndpoint endpoint, @Nullable ReaderContainerFactory<?, ?> factory) {
Assert.notNull(endpoint, "Endpoint must be set");
Assert.hasText(endpoint.getSubscriptionName(), "Endpoint id must be set");
Assert.hasText(endpoint.getId(), "Endpoint id must be set");
// Factory may be null, we defer the resolution right before actually creating the
// container
PulsarReaderEndpointDescriptor descriptor = new PulsarReaderEndpointDescriptor(endpoint, factory);

10
spring-pulsar/src/main/java/org/springframework/pulsar/reader/PulsarReaderContainerProperties.java
View File

@@ -54,8 +54,6 @@ public class PulsarReaderContainerProperties {
private SchemaResolver schemaResolver;
private String subscriptionName;
public Object getReaderListener() {
return this.readerListener;
}
@@ -114,14 +112,6 @@ public class PulsarReaderContainerProperties {
this.schema = schema;
}
public String getSubscriptionName() {
return this.subscriptionName;
}
public void setSubscriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
public SchemaType getSchemaType() {
return this.schemaType;
}

20
spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderTests.java
View File

@@ -50,7 +50,7 @@ import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
/**
* {@link PulsarReader} integration tests.
* Tests for {@link PulsarReader}.
*
* @author Soby Chacko
* @author Chris Bono
@@ -113,9 +113,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
@Configuration
static class PulsarReaderStartMessageIdEarliest {
@PulsarReader(id = "pulsarReaderBasicScenario-id-1",
subscriptionName = "pulsarReaderBasicScenario-subscription-1",
topics = "pulsarReaderBasicScenario-topic-1", startMessageId = "earliest")
@PulsarReader(id = "pulsarReaderBasicScenario-id-1", topics = "pulsarReaderBasicScenario-topic-1",
startMessageId = "earliest")
void read(String ignored) {
latch.countDown();
}
@@ -140,9 +139,7 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
@EnablePulsar
static class PulsarReaderStartMessageIdMissing {
@PulsarReader(id = "pulsarReaderBasicScenario-id-2",
subscriptionName = "pulsarReaderBasicScenario-subscription-2",
topics = "pulsarReaderBasicScenario-topic-2")
@PulsarReader(id = "pulsarReaderBasicScenario-id-2", topics = "pulsarReaderBasicScenario-topic-2")
void readWithoutStartMessageId(String ignored) {
}
@@ -169,9 +166,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
@Configuration
static class PulsarReaderStartMessageIdLatest {
@PulsarReader(id = "pulsarReaderBasicScenario-id-3",
subscriptionName = "pulsarReaderBasicScenario-subscription-3",
topics = "pulsarReaderBasicScenario-topic-3", startMessageId = "latest")
@PulsarReader(id = "pulsarReaderBasicScenario-id-3", topics = "pulsarReaderBasicScenario-topic-3",
startMessageId = "latest")
void read(String msg) {
latch.countDown();
assertThat(msg).isEqualTo("hello foobar");
@@ -200,8 +196,8 @@ public class PulsarReaderTests implements PulsarTestContainerSupport {
MessageId[] messageIds = new MessageId[10];
@PulsarReader(id = "with-customizer-reader", subscriptionName = "with-customizer-reader-subscription",
topics = "with-customizer-reader-topic", readerCustomizer = "myCustomizer")
@PulsarReader(id = "with-customizer-reader", topics = "with-customizer-reader-topic",
readerCustomizer = "myCustomizer")
void listen(Message<String> message) {
assertThat(message.getMessageId()).isEqualTo(messageIds[currentIndex++]);
latch.countDown();