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

Enabling ackMode on PulsarListener

Browse Source
This commit is contained in:
Soby Chacko
2022-09-12 12:17:26 -04:00
parent 69e431d7ab
commit df00e9b67a
14 changed files with 108 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -38,7 +38,7 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.listener.AckMode;
/**
* Configuration properties for Spring for Apache Pulsar.
@@ -1308,18 +1308,18 @@ public class PulsarProperties {
/**
* AckMode for acknowledgements. Allowed values are RECORD, BATCH, MANUAL.
*/
private PulsarContainerProperties.AckMode ackMode;
private AckMode ackMode;
/**
* SchemaType of the consumed messages.
*/
private SchemaType schemaType;
public PulsarContainerProperties.AckMode getAckMode() {
public AckMode getAckMode() {
return this.ackMode;
}
public void setAckMode(PulsarContainerProperties.AckMode ackMode) {
public void setAckMode(AckMode ackMode) {
this.ackMode = ackMode;
}

7
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListener.java
View File

@@ -27,6 +27,7 @@ import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.pulsar.config.PulsarListenerContainerFactory;
import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.listener.AckMode;
/**
* Annotation that marks a method to be the target of a Pulsar message listener on the
@@ -179,4 +180,10 @@ public @interface PulsarListener {
*/
String deadLetterPolicy() default "";
/**
* Override the container default ack mode of BATCH.
* @return ack mode used by the listener
*/
AckMode ackMode() default AckMode.BATCH;
}

1
spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarListenerAnnotationBeanPostProcessor.java
View File

@@ -349,6 +349,7 @@ public class PulsarListenerAnnotationBeanPostProcessor<K, V>
endpoint.setTopicPattern(topicPattern);
endpoint.setSubscriptionType(getEndpointSubscriptionType(pulsarListener));
endpoint.setSchemaType(pulsarListener.schemaType());
endpoint.setAckMode(pulsarListener.ackMode());
String concurrency = pulsarListener.concurrency();
if (StringUtils.hasText(concurrency)) {

10
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerContainerFactory.java
View File

@@ -28,6 +28,7 @@ import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.log.LogAccessor;
import org.springframework.pulsar.core.PulsarConsumerFactory;
import org.springframework.pulsar.listener.AbstractPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarContainerProperties;
import org.springframework.pulsar.support.JavaUtils;
import org.springframework.pulsar.support.MessageConverter;
@@ -148,16 +149,19 @@ public abstract class AbstractPulsarListenerContainerFactory<C extends AbstractP
properties.setSchemaType(this.containerProperties.getSchemaType());
}
}
if (properties.getSchema() == null) {
properties.setSchema(Schema.BYTES);
}
if (properties.getSubscriptionType() == null) {
properties.setSubscriptionType(this.containerProperties.getSubscriptionType());
}
properties.setAckMode(this.containerProperties.getAckMode());
if (endpoint.getAckMode() != AckMode.BATCH) {
properties.setAckMode(endpoint.getAckMode());
}
else if (this.containerProperties.getAckMode() != AckMode.BATCH) {
properties.setAckMode(this.containerProperties.getAckMode());
}
Boolean autoStart = endpoint.getAutoStartup();
if (autoStart != null) {

11
spring-pulsar/src/main/java/org/springframework/pulsar/config/AbstractPulsarListenerEndpoint.java
View File

@@ -37,6 +37,7 @@ import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.core.log.LogAccessor;
import org.springframework.expression.BeanResolver;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter;
import org.springframework.pulsar.support.MessageConverter;
@@ -83,6 +84,8 @@ public abstract class AbstractPulsarListenerEndpoint<K>
private Integer concurrency;
private AckMode ackMode;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
@@ -244,4 +247,12 @@ public abstract class AbstractPulsarListenerEndpoint<K>
this.concurrency = concurrency;
}
public AckMode getAckMode() {
return this.ackMode;
}
public void setAckMode(AckMode ackMode) {
this.ackMode = ackMode;
}
}

3
spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java
View File

@@ -48,7 +48,6 @@ public class ConcurrentPulsarListenerContainerFactory<T>
@Override
protected ConcurrentPulsarMessageListenerContainer<T> createContainerInstance(PulsarListenerEndpoint endpoint) {
PulsarContainerProperties properties = new PulsarContainerProperties();
Collection<String> topics = endpoint.getTopics();
String topicPattern = endpoint.getTopicPattern();
@@ -57,13 +56,11 @@ public class ConcurrentPulsarListenerContainerFactory<T>
final String[] topics1 = topics.toArray(new String[0]);
properties.setTopics(topics1);
}
if (StringUtils.hasText(topicPattern)) {
properties.setTopicsPattern(topicPattern);
}
final String subscriptionName = endpoint.getSubscriptionName();
if (StringUtils.hasText(subscriptionName)) {
properties.setSubscriptionName(endpoint.getSubscriptionName());
}

3
spring-pulsar/src/main/java/org/springframework/pulsar/config/PulsarListenerEndpoint.java
View File

@@ -23,6 +23,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
import org.springframework.pulsar.support.MessageConverter;
@@ -64,4 +65,6 @@ public interface PulsarListenerEndpoint {
@Nullable
Integer getConcurrency();
AckMode getAckMode();
}

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

@@ -24,6 +24,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.common.schema.SchemaType;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarMessageListenerContainer;
import org.springframework.pulsar.support.MessageConverter;
@@ -92,4 +93,9 @@ public class PulsarListenerEndpointAdapter implements PulsarListenerEndpoint {
return null;
}
@Override
public AckMode getAckMode() {
return null;
}
}

39
spring-pulsar/src/main/java/org/springframework/pulsar/listener/AckMode.java Normal file
View File

@@ -0,0 +1,39 @@
/*
* Copyright 2022 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.
* You may obtain a copy of the License at
*
* https://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.springframework.pulsar.listener;
/**
* Enumeration for ack mode.
*
* @author Soby Chacko
*/
public enum AckMode {
/**
* Batch ack mode.
*/
BATCH,
/**
* Record ack mode.
*/
RECORD,
/**
* Manual ack mode.
*/
MANUAL;
}

25
spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java
View File

@@ -286,14 +286,13 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
if (messageList != null && messageList.size() > 0) {
if (this.batchMessageListener instanceof PulsarBatchAcknowledgingMessageListener) {
this.batchMessageListener.received(this.consumer, messageList,
this.containerProperties
.getAckMode() == PulsarContainerProperties.AckMode.MANUAL
? new ConsumerBatchAcknowledgment(this.consumer) : null);
this.containerProperties.getAckMode() == AckMode.MANUAL
? new ConsumerBatchAcknowledgment(this.consumer) : null);
}
else {
this.batchMessageListener.received(this.consumer, messageList);
}
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
if (this.containerProperties.getAckMode() == AckMode.BATCH) {
try {
if (isSharedSubscriptionType()) {
this.consumer.acknowledge(messages);
@@ -332,15 +331,13 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
try {
if (this.listener instanceof PulsarAcknowledgingMessageListener) {
this.listener.received(this.consumer, message,
this.containerProperties
.getAckMode() == PulsarContainerProperties.AckMode.MANUAL
? new ConsumerAcknowledgment(this.consumer, message)
: null);
this.containerProperties.getAckMode() == AckMode.MANUAL
? new ConsumerAcknowledgment(this.consumer, message) : null);
}
else if (this.listener != null) {
this.listener.received(this.consumer, message);
}
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) {
if (this.containerProperties.getAckMode() == AckMode.RECORD) {
handleAck(message);
}
if (inRetryMode.get()) {
@@ -352,12 +349,10 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
invokeRecordListenerErrorHandler(inRetryMode, message, e);
}
else {
if (this.containerProperties
.getAckMode() == PulsarContainerProperties.AckMode.RECORD) {
if (this.containerProperties.getAckMode() == AckMode.RECORD) {
this.consumer.negativeAcknowledge(message);
}
else if (this.containerProperties
.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
else if (this.containerProperties.getAckMode() == AckMode.BATCH) {
this.nackableMessages.add(message.getMessageId());
}
}
@@ -366,7 +361,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
while (inRetryMode.get());
}
// All the records are processed at this point. Handle acks.
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
if (this.containerProperties.getAckMode() == AckMode.BATCH) {
handleAcks(messages);
}
}
@@ -449,7 +444,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.pulsarConsumerErrorHandler.recoverMessage(this.consumer, message, e);
// retries exhausted - if record ackmode, acknowledge, otherwise normal
// batch ack at the end
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.RECORD) {
if (this.containerProperties.getAckMode() == AckMode.RECORD) {
handleAck(message);
}
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java
View File

@@ -38,26 +38,6 @@ public class PulsarContainerProperties {
private Duration consumerStartTimeout = DEFAULT_CONSUMER_START_TIMEOUT;
/**
* Enumeration for ack mode.
*/
public enum AckMode {
/**
* Batch ack mode.
*/
BATCH,
/**
* Record ack mode.
*/
RECORD,
/**
* Manual ack mode.
*/
MANUAL;
}
private String[] topics;
private String topicsPattern;

5
spring-pulsar/src/test/java/org/springframework/pulsar/core/ConsumerAcknowledgmentTests.java
View File

@@ -46,6 +46,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Test;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.Acknowledgement;
import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer;
import org.springframework.pulsar.listener.PulsarAcknowledgingMessageListener;
@@ -73,7 +74,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests {
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<?>) (consumer, msg) -> {
});
pulsarContainerProperties.setSchema(Schema.STRING);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.RECORD);
pulsarContainerProperties.setAckMode(AckMode.RECORD);
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.start();
@@ -235,7 +236,7 @@ class ConsumerAcknowledgmentTests extends AbstractContainerBaseTests {
pulsarContainerProperties.setMessageListener(pulsarAcknowledgingMessageListener);
pulsarContainerProperties.setSchema(Schema.STRING);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<String> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
container.start();

10
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java
View File

@@ -246,7 +246,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
@@ -316,7 +316,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
@@ -385,7 +385,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
@@ -460,7 +460,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);
@@ -531,7 +531,7 @@ public class DefaultPulsarConsumerErrorHandlerTests extends AbstractContainerBas
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(PulsarContainerProperties.AckMode.MANUAL);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
DefaultPulsarMessageListenerContainer<Integer> container = new DefaultPulsarMessageListenerContainer<>(
pulsarConsumerFactory, pulsarContainerProperties);
PulsarTemplate<Integer> mockPulsarTemplate = mock(PulsarTemplate.class, RETURNS_DEEP_STUBS);

12
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -206,6 +206,13 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
void ackModeAppliedToContainerFromListener(@Autowired PulsarListenerEndpointRegistry registry) {
final PulsarContainerProperties pulsarContainerProperties = registry.getListenerContainer("ackMode-test-id")
.getContainerProperties();
assertThat(pulsarContainerProperties.getAckMode()).isEqualTo(AckMode.RECORD);
}
@EnablePulsar
@Configuration
static class TestPulsarListenersForBasicScenario {
@@ -228,6 +235,11 @@ public class PulsarListenerTests extends AbstractContainerBaseTests {
latch2.countDown();
}
@PulsarListener(id = "ackMode-test-id", subscriptionName = "ackModeTest-sub", topics = "ackModeTest-topic",
ackMode = AckMode.RECORD)
void ackModeTestListener(String message) {
}
}
}