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

Improve acking when there are no errors

Browse Source 
When no shared subcription types are used and no errors from processing,
we can optimize acking by relying on the acknowledgeCumulative on the
Pulsar consumer.
This commit is contained in:
Soby Chacko
2022-08-09 18:29:07 -04:00
parent 749f2dbb5e
commit a0de08e098
3 changed files with 27 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,8 @@ import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;
import org.apache.pulsar.client.api.BatchReceivePolicy;
import org.apache.pulsar.client.api.Consumer;
@@ -256,7 +258,15 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
try {
this.consumer.acknowledge(messages);
if (isSharedSubsriptionType()) {
this.consumer.acknowledge(messages);
}
else {
final Stream<Message<T>> stream = StreamSupport.stream(messages.spliterator(),
true);
Message<T> last = stream.reduce((a, b) -> b).orElse(null);
this.consumer.acknowledgeCumulative(last);
}
}
catch (PulsarClientException pce) {
this.consumer.negativeAcknowledge(messages);
@@ -303,11 +313,23 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
}
private boolean isSharedSubsriptionType() {
return this.containerProperties.getSubscriptionType() == SubscriptionType.Shared
|| this.containerProperties.getSubscriptionType() == SubscriptionType.Key_Shared;
}
private void handleAcks(Messages<T> messages) {
if (this.nackableMessages.isEmpty()) {
try {
if (messages.size() > 0) {
this.consumer.acknowledge(messages);
if (isSharedSubsriptionType()) {
this.consumer.acknowledge(messages);
}
else {
final Stream<Message<T>> stream = StreamSupport.stream(messages.spliterator(), true);
Message<T> last = stream.reduce((a, b) -> b).orElse(null);
this.consumer.acknowledgeCumulative(last);
}
}
}
catch (PulsarClientException pce) {

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

@@ -63,7 +63,7 @@ public class PulsarContainerProperties {
private String subscriptionName;
private SubscriptionType subscriptionType;
private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
private Schema<?> schema;

4
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarMessageListenerContainerTests.java
View File

@@ -128,7 +128,7 @@ class PulsarMessageListenerContainerTests extends AbstractContainerBaseTests {
}
assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
verify(containerConsumer, never()).acknowledge(any(Message.class));
verify(containerConsumer, atLeastOnce()).acknowledge(any(Messages.class));
verify(containerConsumer, atLeastOnce()).acknowledgeCumulative(any(Message.class));
container.stop();
pulsarClient.close();
}
@@ -270,7 +270,7 @@ class PulsarMessageListenerContainerTests extends AbstractContainerBaseTests {
}
assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
verify(pulsarBatchMessageListener, times(1)).received(any(Consumer.class), any(Messages.class));
verify(containerConsumer, times(1)).acknowledge(any(Messages.class));
verify(containerConsumer, times(1)).acknowledgeCumulative(any(Message.class));
container.stop();
pulsarClient.close();
}