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

Fix batch listener error handling (#1007)

Browse Source 
Updates the batch listener error handler code to handle the case where there
is only a single outstanding message in the batch list when retries are expired.

See #998

Signed-off-by: darshimo <takumishimoda7623@gmail.com>
Co-authored-by: darshimo <takumishimoda7623@gmail.com>
This commit is contained in:
Chris Bono
2025-01-19 18:13:51 -06:00
committed by GitHub
parent b493e4ff5c
commit 7881c8f99d
2 changed files with 78 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -593,12 +593,11 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
pulsarBatchListenerFailedException);
handleAck(pulsarMessage);
if (messageList.size() == 1) {
messageList.remove(0);
messagesPendingInBatch.set(false);
}
else {
messageList = messageList.subList(1, messageList.size());
}
if (!messageList.isEmpty()) {
messagesPendingInBatch.set(true);
}
this.pulsarConsumerErrorHandler.clearMessage();

77
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarConsumerErrorHandlerTests.java
View File

@@ -38,6 +38,7 @@ import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.junit.jupiter.api.Test;
import org.springframework.core.log.LogAccessor;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarOperations;
@@ -51,6 +52,8 @@ import org.springframework.util.backoff.FixedBackOff;
*/
public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContainerSupport {
private final LogAccessor logger = new LogAccessor(this.getClass());
@Test
@SuppressWarnings("unchecked")
void happyPathErrorHandlingForRecordMessageListener() throws Exception {
@@ -564,4 +567,78 @@ public class DefaultPulsarConsumerErrorHandlerTests implements PulsarTestContain
pulsarClient.close();
}
@Test
@SuppressWarnings("unchecked")
void whenBatchRecordListenerOneMessageBatchFailsThenSentToDltProperly() throws Exception {
var topicName = "default-error-handler-tests-9";
var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
var pulsarConsumerFactory = new DefaultPulsarConsumerFactory<Integer>(pulsarClient,
List.of((consumerBuilder) -> {
consumerBuilder.topic(topicName);
consumerBuilder.subscriptionName("%s-sub".formatted(topicName));
}));
// Prepare container for batch consume
var pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setSchema(Schema.INT32);
pulsarContainerProperties.setAckMode(AckMode.MANUAL);
pulsarContainerProperties.setBatchListener(true);
pulsarContainerProperties.setMaxNumMessages(1);
pulsarContainerProperties.setBatchTimeoutMillis(60_000);
PulsarBatchAcknowledgingMessageListener<?> pulsarBatchMessageListener = mock();
doAnswer(invocation -> {
List<Message<Integer>> message = invocation.getArgument(1);
Message<Integer> integerMessage = message.get(0);
Integer value = integerMessage.getValue();
if (value == 0) {
throw new PulsarBatchListenerFailedException("failed", integerMessage);
}
Acknowledgement acknowledgment = invocation.getArgument(2);
List<MessageId> messageIds = new ArrayList<>();
for (Message<Integer> integerMessage1 : message) {
messageIds.add(integerMessage1.getMessageId());
}
acknowledgment.acknowledge(messageIds);
return new Object();
}).when(pulsarBatchMessageListener).received(any(Consumer.class), any(List.class), any(Acknowledgement.class));
pulsarContainerProperties.setMessageListener(pulsarBatchMessageListener);
var container = new DefaultPulsarMessageListenerContainer<>(pulsarConsumerFactory, pulsarContainerProperties);
// Set error handler to recover after 2 retries
PulsarTemplate<Integer> mockPulsarTemplate = mock(RETURNS_DEEP_STUBS);
PulsarOperations.SendMessageBuilder<Integer> sendMessageBuilderMock = mock();
when(mockPulsarTemplate.newMessage(any(Integer.class))
.withTopic(any(String.class))
.withMessageCustomizer(any(TypedMessageBuilderCustomizer.class))).thenReturn(sendMessageBuilderMock);
container.setPulsarConsumerErrorHandler(new DefaultPulsarConsumerErrorHandler<>(
new PulsarDeadLetterPublishingRecoverer<>(mockPulsarTemplate), new FixedBackOff(100, 2)));
try {
container.start();
// Send single message in batch
var pulsarProducerFactory = new DefaultPulsarProducerFactory<Integer>(pulsarClient, topicName);
var pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
pulsarTemplate.sendAsync(0);
// Initial call should fail
// Next 2 calls should fail (retries 2)
// No more calls after that - msg should go to DLT
await().atMost(Duration.ofSeconds(30))
.untilAsserted(() -> verify(pulsarBatchMessageListener, times(3)).received(any(Consumer.class),
any(List.class), any(Acknowledgement.class)));
await().atMost(Duration.ofSeconds(30))
.untilAsserted(() -> verify(sendMessageBuilderMock, times(1)).sendAsync());
}
finally {
safeStopContainer(container);
}
pulsarClient.close();
}
private void safeStopContainer(PulsarMessageListenerContainer container) {
try {
container.stop();
}
catch (Exception ex) {
logger.warn(ex, "Failed to stop container %s: %s".formatted(container, ex.getMessage()));
}
}
}