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

Rework batch acknowledgment

Browse Source
This commit is contained in:
Soby Chacko
2022-07-22 20:11:59 -04:00
parent 3b588caa49
commit 3faddca84a
2 changed files with 10 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,9 +23,11 @@ import java.util.List;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Messages;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.pulsar.listener.PulsarBatchMessageListener;
import org.springframework.pulsar.listener.Acknowledgement;
import org.springframework.pulsar.listener.PulsarBatchAcknowledgingMessageListener;
import org.springframework.pulsar.support.converter.PulsarBatchMessageConverter;
import org.springframework.pulsar.support.converter.PulsarBatchMessagingMessageConverter;
import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter;
@@ -42,7 +44,7 @@ import org.springframework.util.Assert;
*/
@SuppressWarnings("serial")
public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagingMessageListenerAdapter<V>
implements PulsarBatchMessageListener<V> {
implements PulsarBatchAcknowledgingMessageListener<V> {
private PulsarBatchMessageConverter<V> batchMessageConverter = new PulsarBatchMessagingMessageConverter<V>();
@@ -63,7 +65,8 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagi
return this.batchMessageConverter;
}
public void received(Consumer<V> consumer, Messages<V> msg) {
@Override
public void received(Consumer<V> consumer, Messages<V> msg, @Nullable Acknowledgement acknowledgement) {
Message<?> message;
if (!isConsumerRecordList()) {
if (isMessageList()) {
@@ -81,15 +84,15 @@ public class PulsarBatchMessagingMessageListenerAdapter<V> extends PulsarMessagi
message = null; // optimization since we won't need any conversion to invoke
}
logger.debug(() -> "Processing [" + message + "]");
invoke(msg, consumer, message);
invoke(msg, consumer, message, acknowledgement);
}
protected void invoke(Object records, Consumer<V> consumer,
final Message<?> messageArg) {
final Message<?> messageArg, Acknowledgement acknowledgement) {
Message<?> message = messageArg;
try {
Object result = invokeHandler(records, message, consumer, null);
Object result = invokeHandler(records, message, consumer, acknowledgement);
// if (result != null) {
// handleResult(result, records, 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(1_000);
Thread.sleep(5_000); //Rework this
// 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));