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

Listener container properties for batch receive.

Browse Source 
Acknowledgment API cleanup.
This commit is contained in:
Soby Chacko
2022-09-15 21:57:52 -04:00
parent 9349ce9f85
commit 6a01ad7fe0
4 changed files with 79 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java
View File

@@ -61,6 +61,9 @@ public class PulsarAnnotationDrivenConfiguration {
map.from(properties::getSchemaType).to(containerProperties::setSchemaType);
map.from(properties::getAckMode).to(containerProperties::setAckMode);
map.from(properties::getBatchTimeout).to(containerProperties::setBatchTimeout);
map.from(properties::getMaxNumBytes).to(containerProperties::setMaxNumBytes);
map.from(properties::getMaxNumMessages).to(containerProperties::setMaxNumMessages);
return factory;
}

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

@@ -1323,6 +1323,21 @@ public class PulsarProperties {
*/
private SchemaType schemaType;
/**
* Max number of messages for batch.
*/
private int maxNumMessages = -1;
/**
* Max number of bytes for batch.
*/
private int maxNumBytes = 10 * 1024 * 1024;
/**
* Timeout for batch.
*/
private int batchTimeout = 100;
public AckMode getAckMode() {
return this.ackMode;
}
@@ -1339,6 +1354,30 @@ public class PulsarProperties {
this.schemaType = schemaType;
}
public int getMaxNumMessages() {
return this.maxNumMessages;
}
public void setMaxNumMessages(int maxNumMessages) {
this.maxNumMessages = maxNumMessages;
}
public int getMaxNumBytes() {
return this.maxNumBytes;
}
public void setMaxNumBytes(int maxNumBytes) {
this.maxNumBytes = maxNumBytes;
}
public int getBatchTimeout() {
return this.batchTimeout;
}
public void setBatchTimeout(int batchTimeout) {
this.batchTimeout = batchTimeout;
}
}
public static class Admin {

12
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -201,6 +201,18 @@ class PulsarAutoConfigurationTests {
InterceptorTestConfiguration.interceptorFoo)));
}
@Test
void consumerBatchPropertiesAreHonored() {
contextRunner
.withPropertyValues("spring.pulsar.listener.maxNumMessages=10",
"spring.pulsar.listener.maxNumBytes=101", "spring.pulsar.listener.batchTimeout=50")
.run((context -> assertThat(context).hasNotFailed()
.getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("containerProperties")
.hasFieldOrPropertyWithValue("maxNumMessages", 10)
.hasFieldOrPropertyWithValue("maxNumBytes", 101)
.hasFieldOrPropertyWithValue("batchTimeout", 50)));
}
@Nested
class ClientAutoConfigurationTests {

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

@@ -360,6 +360,11 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
else if (this.containerProperties.getAckMode() == AckMode.BATCH) {
this.nackableMessages.add(message.getMessageId());
}
else {
throw new IllegalStateException(
"Exception occurred and not negatively acknowledged or handled properly",
e);
}
}
}
}
@@ -541,12 +546,29 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
@Override
public void acknowledge(MessageId messageId) {
throw new UnsupportedOperationException();
try {
this.consumer.acknowledge(messageId);
}
catch (PulsarClientException e) {
this.consumer.negativeAcknowledge(messageId);
}
}
@Override
public void acknowledge(List<MessageId> messageIds) {
throw new UnsupportedOperationException();
try {
this.consumer.acknowledge(messageIds);
}
catch (PulsarClientException e) {
for (MessageId messageId : messageIds) {
try {
this.consumer.acknowledge(messageId);
}
catch (PulsarClientException ex) {
this.consumer.negativeAcknowledge(messageId);
}
}
}
}
@Override
@@ -556,7 +578,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
@Override
public void nack(MessageId messageId) {
throw new UnsupportedOperationException();
this.consumer.negativeAcknowledge(messageId);
}
}