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

Nacking messages in record mode improvements

Browse Source
This commit is contained in:
Soby Chacko
2022-07-23 17:26:25 -04:00
parent 33fba1d450
commit fde611328c
2 changed files with 6 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,7 +21,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -167,7 +166,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
private Consumer<T> consumer;
private final Set<Message<T>> nackableMessages = new HashSet<>();
private final Set<MessageId> nackableMessages = new HashSet<>();
private final PulsarContainerProperties containerProperties = getPulsarContainerProperties();
@@ -291,7 +290,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.consumer.negativeAcknowledge(message);
}
else if (this.containerProperties.getAckMode() == PulsarContainerProperties.AckMode.BATCH) {
this.nackableMessages.add(message);
this.nackableMessages.add(message.getMessageId());
}
}
}
@@ -316,13 +315,9 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
}
else {
for (Message<T> message : messages) {
final Optional<Message<T>> foundMessage = this.nackableMessages.stream()
.filter(m -> m.getMessageId().compareTo(message.getMessageId()) == 0)
.findFirst();
if (foundMessage.isPresent()) {
final Message<T> msg = foundMessage.get();
this.consumer.negativeAcknowledge(msg);
this.nackableMessages.remove(msg);
if (this.nackableMessages.contains(message.getMessageId())) {
this.consumer.negativeAcknowledge(message);
this.nackableMessages.remove(message.getMessageId());
}
else {
handleAck(message);

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

@@ -172,7 +172,7 @@ class PulsarMessageListenerContainerTests extends AbstractContainerBaseTests {
pulsarTemplate.sendAsync("hello john doe");
}
assertThat(latch.await(30, TimeUnit.SECONDS)).isTrue();
Thread.sleep(5_000); //Rework this
Thread.sleep(1_000);
// Half of the message get acknowledged, and the other half gets negatively acknowledged.
verify(containerConsumer, times(5)).acknowledge(any(Message.class));
verify(containerConsumer, times(5)).negativeAcknowledge(any(Message.class));