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

Polish "Listener container properties for batch receive"

Browse Source
This commit is contained in:
Chris Bono
2022-09-16 12:08:59 -05:00
committed by Chris Bono
parent 6a01ad7fe0
commit eef1616aec
4 changed files with 15 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -61,7 +61,7 @@ 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::getBatchTimeoutMillis).to(containerProperties::setBatchTimeout);
map.from(properties::getMaxNumBytes).to(containerProperties::setMaxNumBytes);
map.from(properties::getMaxNumMessages).to(containerProperties::setMaxNumMessages);

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

@@ -1324,19 +1324,20 @@ public class PulsarProperties {
private SchemaType schemaType;
/**
* Max number of messages for batch.
* Max number of messages in a single batch request.
*/
private int maxNumMessages = -1;
/**
* Max number of bytes for batch.
* Max number of bytes in a single batch request.
*/
private int maxNumBytes = 10 * 1024 * 1024;
/**
* Timeout for batch.
* Number of milliseconds to wait for enough message to fill a batch request
* before timing out.
*/
private int batchTimeout = 100;
private int batchTimeoutMillis = 100;
public AckMode getAckMode() {
return this.ackMode;
@@ -1370,12 +1371,12 @@ public class PulsarProperties {
this.maxNumBytes = maxNumBytes;
}
public int getBatchTimeout() {
return this.batchTimeout;
public int getBatchTimeoutMillis() {
return this.batchTimeoutMillis;
}
public void setBatchTimeout(int batchTimeout) {
this.batchTimeout = batchTimeout;
public void setBatchTimeoutMillis(int batchTimeoutMillis) {
this.batchTimeoutMillis = batchTimeoutMillis;
}
}

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

@@ -204,8 +204,8 @@ class PulsarAutoConfigurationTests {
@Test
void consumerBatchPropertiesAreHonored() {
contextRunner
.withPropertyValues("spring.pulsar.listener.maxNumMessages=10",
"spring.pulsar.listener.maxNumBytes=101", "spring.pulsar.listener.batchTimeout=50")
.withPropertyValues("spring.pulsar.listener.max-num-messages=10",
"spring.pulsar.listener.max-num-bytes=101", "spring.pulsar.listener.batch-timeout-millis=50")
.run((context -> assertThat(context).hasNotFailed()
.getBean(ConcurrentPulsarListenerContainerFactory.class).extracting("containerProperties")
.hasFieldOrPropertyWithValue("maxNumMessages", 10)

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

@@ -361,9 +361,9 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.nackableMessages.add(message.getMessageId());
}
else {
throw new IllegalStateException(
"Exception occurred and not negatively acknowledged or handled properly",
e);
throw new IllegalStateException(String.format(
"Exception occurred and message %s was not auto-nacked; switch to AckMode BATCH or RECORD to enable auto-nacks",
message.getMessageId()), e);
}
}
}