From e309277d0d13b29f19cc36ec8125fc20798a2292 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sun, 21 Apr 2024 12:08:57 -0500 Subject: [PATCH] Add test for batch listener w/ error handler See #661 --- ...DefaultPulsarMessageListenerContainer.java | 25 +++++++++---------- ...ulsarMessageListenerContainerTxnTests.java | 13 +++++++++- 2 files changed, 24 insertions(+), 14 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java index 6ba4b59d..cb3dc2d8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainer.java @@ -256,6 +256,7 @@ public class DefaultPulsarMessageListenerContainer 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 extends AbstractPulsarMess this.listener = null; this.batchMessageListener = null; } - this.pulsarConsumerErrorHandler = getPulsarConsumerErrorHandler(); this.consumerBuilderCustomizer = getConsumerBuilderCustomizer(); try { Map propertiesToConsumer = extractDirectConsumerProperties(); @@ -319,19 +319,19 @@ public class DefaultPulsarMessageListenerContainer 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 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); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java index 2c30e440..6c2fff70 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTxnTests.java @@ -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(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,