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

Add test for batch listener w/ error handler

Browse Source 
See #661
This commit is contained in:
Chris Bono
2024-04-21 12:08:57 -05:00
parent 01e96ddfa3
commit e309277d0d
2 changed files with 24 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -256,6 +256,7 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.isBatchListener = this.containerProperties.isBatchListener();
this.ackMode = this.containerProperties.getAckMode();
this.subscriptionType = this.containerProperties.getSubscriptionType();
this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler();
validateTransactionSettings(this.containerProperties.transactions());
this.transactionManager = this.containerProperties.transactions().getTransactionManager();
this.transactionTemplate = determineTransactionTemplate();
@@ -271,7 +272,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
this.listener = null;
this.batchMessageListener = null;
}
this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler();
this.consumerBuilderCustomizer = getConsumerBuilderCustomizer();
try {
Map<String, Object> propertiesToConsumer = extractDirectConsumerProperties();
@@ -319,19 +319,19 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
if (!txnProps.isEnabled()) {
return;
}
var missingRequiredTxnMgr = txnProps.isRequired() && txnProps.getTransactionManager() == null;
Assert.state(!missingRequiredTxnMgr, "Transactions are required but txn manager is null");
var txnRecordListenerWithBatchAckMode = (txnProps.getTransactionManager() != null && !this.isBatchListener
&& this.containerProperties.getAckMode() == AckMode.BATCH);
Assert.state(txnProps.getTransactionManager() != null,
"Transactions are enabled but txn manager is not set");
var txnRecordListenerWithBatchAckMode = !this.isBatchListener
&& this.containerProperties.getAckMode() == AckMode.BATCH;
Assert.state(!(txnRecordListenerWithBatchAckMode),
"Transactional record listeners can not use batch ack mode");
var batchListenerWithRecordAckMode = (this.isBatchListener
&& this.containerProperties.getAckMode() == AckMode.RECORD);
Assert.state(!(batchListenerWithRecordAckMode), "Batch record listeners do not support AckMode.RECORD");
// TODO custom errorHandler w/ transactions not supported
var txnBatchListenerWithRecordAckMode = this.isBatchListener
&& this.containerProperties.getAckMode() == AckMode.RECORD;
Assert.state(!(txnBatchListenerWithRecordAckMode),
"Transactional batch listeners do not support AckMode.RECORD");
var txnBatchListenerWithErrorHandler = this.isBatchListener && this.pulsarConsumerErrorHandler != null;
Assert.state(!(txnBatchListenerWithErrorHandler),
"Transactional batch listeners do not support custom error handlers");
}
@Nullable
@@ -679,7 +679,6 @@ public class DefaultPulsarMessageListenerContainer<T> extends AbstractPulsarMess
return Collections.emptyList();
}
catch (RuntimeException ex) {
// TODO enforce no error handler w/ batch listener w/ txn
if (this.pulsarConsumerErrorHandler != null) {
return invokeBatchListenerErrorHandler(inRetryMode, messagesPendingInBatch, messageList, ex, txn);
}

13
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.pulsar.listener;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
import static org.mockito.Mockito.mock;
import java.time.Duration;
import java.util.Arrays;
@@ -389,7 +390,17 @@ class DefaultPulsarMessageListenerContainerTxnTests {
@Test
void txnBatchListenerWithErrorHandlerNotSupported() {
// TODO
var containerProps = newContainerProps();
containerProps.setAckMode(AckMode.BATCH);
containerProps.setBatchListener(true);
containerProps.setMessageListener((PulsarBatchMessageListener<?>) (consumer, msg) -> {
throw new RuntimeException("should never get here");
});
var consumerFactory = new DefaultPulsarConsumerFactory<String>(client, List.of());
var container = new DefaultPulsarMessageListenerContainer<>(consumerFactory, containerProps);
container.setPulsarConsumerErrorHandler(mock(PulsarConsumerErrorHandler.class));
assertThatIllegalStateException().isThrownBy(() -> container.start())
.withMessage("Transactional batch listeners do not support custom error handlers");
}
private void startContainerAndSendInputsThenWaitForLatch(String topicIn, PulsarContainerProperties containerProps,