diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java index 961ec4d7..264cf2e3 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java @@ -20,15 +20,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatException; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import java.time.Duration; -import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; -import org.apache.pulsar.client.api.Message; -import org.apache.pulsar.client.api.Schema; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -37,9 +33,8 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarListener; +import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactoryCustomizer; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithCommit.BatchListenerWithCommitConfig; import org.springframework.pulsar.listener.PulsarListenerTxnTests.BatchListenerWithRollback.BatchListenerWithRollbackConfig; @@ -47,7 +42,7 @@ import org.springframework.pulsar.listener.PulsarListenerTxnTests.ListenerWithEx import org.springframework.pulsar.listener.PulsarListenerTxnTests.ListenerWithExternalTransactionRollback.ListenerWithExternalTransactionRollbackConfig; import org.springframework.pulsar.listener.PulsarListenerTxnTests.RecordListenerWithCommit.RecordListenerWithCommitConfig; import org.springframework.pulsar.listener.PulsarListenerTxnTests.RecordListenerWithRollback.RecordListenerWithRollbackConfig; -import org.springframework.pulsar.test.support.PulsarConsumerTestUtil; +import org.springframework.pulsar.transaction.PulsarTxnTestsBase; import org.springframework.test.context.ContextConfiguration; import org.springframework.transaction.annotation.Transactional; @@ -58,36 +53,6 @@ import org.springframework.transaction.annotation.Transactional; */ class PulsarListenerTxnTests extends PulsarTxnTestsBase { - private void assertNoMessagesAvailableInOutputTopic(String topicOut) { - assertThat(PulsarConsumerTestUtil.consumeMessages(pulsarClient) - .fromTopic(topicOut) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(7)) - .get()).isEmpty(); - } - - private void assertMessagesAvailableInOutputTopic(String topicOut, String... expectedMessages) { - this.assertMessagesAvailableInOutputTopic(topicOut, Arrays.stream(expectedMessages).toList()); - } - - private void assertMessagesAvailableInOutputTopic(String topicOut, List expectedMessages) { - assertThat(PulsarConsumerTestUtil.consumeMessages(pulsarClient) - .fromTopic(topicOut) - .withSchema(Schema.STRING) - .awaitAtMost(Duration.ofSeconds(5)) - .get()).map(Message::getValue).containsExactlyInAnyOrderElementsOf(expectedMessages); - } - - private PulsarTemplate newNonTransactionalTemplate(boolean sendInBatch, int numMessages) { - List> customizers = List.of(); - if (sendInBatch) { - customizers = List.of((pb) -> pb.enableBatching(true) - .batchingMaxPublishDelay(2, TimeUnit.SECONDS) - .batchingMaxMessages(numMessages)); - } - return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient, null, customizers)); - } - @Nested @ContextConfiguration(classes = ListenerWithExternalTransactionConfig.class) class ListenerWithExternalTransaction { @@ -101,7 +66,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); + assertThatMessagesAreInTopic(topicOut, "msg1-out"); } @EnablePulsar @@ -135,7 +100,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertNoMessagesAvailableInOutputTopic(topicOut); + assertThatMessagesAreNotInTopic(topicOut, "msg1-out"); } @EnablePulsar @@ -170,7 +135,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertMessagesAvailableInOutputTopic(topicOut, "msg1-out"); + assertThatMessagesAreInTopic(topicOut, "msg1-out"); } @EnablePulsar @@ -203,7 +168,7 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(false, 1); nonTransactionalTemplate.send(topicIn, "msg1"); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertNoMessagesAvailableInOutputTopic(topicOut); + assertThatMessagesAreNotInTopic(topicOut, "msg1-out"); } @EnablePulsar @@ -238,8 +203,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size()); inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg)); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toList(); - assertMessagesAvailableInOutputTopic(topicOut, outputMsgs); + var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toArray(String[]::new); + assertThatMessagesAreInTopic(topicOut, outputMsgs); } @EnablePulsar @@ -275,7 +240,8 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { var nonTransactionalTemplate = newNonTransactionalTemplate(true, inputMsgs.size()); inputMsgs.forEach((msg) -> nonTransactionalTemplate.sendAsync(topicIn, msg)); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertNoMessagesAvailableInOutputTopic(topicOut); + var outputMsgs = inputMsgs.stream().map((m) -> m.concat("-out")).toArray(String[]::new); + assertThatMessagesAreNotInTopic(topicOut, outputMsgs); } @EnablePulsar @@ -306,8 +272,9 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { assertThatIllegalStateException().isThrownBy(() -> { var context = new AnnotationConfigApplicationContext(); context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class); - context.registerBean("containerPropsRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, - () -> (c) -> c.transactions().setRequired(true)); + context.registerBean("containerPropsRequiredCustomizer", + ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> (cf) -> cf.getContainerProperties().transactions().setRequired(true)); context.refresh(); }).withMessage("Listener w/ id [%s] requested no transactions but txn are required".formatted(LISTENER_ID)); } @@ -316,8 +283,9 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { void disablesTransactionsWhenTransactionsAreNotRequired() { try (var context = new AnnotationConfigApplicationContext()) { context.register(TopLevelConfig.class, TransactionsDisabledOnListenerConfig.class); - context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, - () -> (c) -> c.transactions().setRequired(false)); + context.registerBean("containerPropsNotRequiredCustomizer", + ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> (cf) -> cf.getContainerProperties().transactions().setRequired(false)); context.refresh(); var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID); assertThat(container).isNotNull(); @@ -348,8 +316,9 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { assertThatException().isThrownBy(() -> { var context = new AnnotationConfigApplicationContext(); context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class); - context.registerBean("removeTxnManagerCustomizer", PulsarContainerPropertiesCustomizer.class, - () -> (c) -> c.transactions().setTransactionManager(null)); + context.registerBean("removeTxnManagerCustomizer", + ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> (cf) -> cf.getContainerProperties().transactions().setTransactionManager(null)); context.refresh(); }) .withCauseInstanceOf(IllegalStateException.class) @@ -361,8 +330,9 @@ class PulsarListenerTxnTests extends PulsarTxnTestsBase { void enablesTransactionsWhenTxnManagerAvailable() { try (var context = new AnnotationConfigApplicationContext()) { context.register(TopLevelConfig.class, TransactionsEnabledOnListenerConfig.class); - context.registerBean("containerPropsNotRequiredCustomizer", PulsarContainerPropertiesCustomizer.class, - () -> (c) -> c.transactions().setEnabled(false)); + context.registerBean("containerPropsNotRequiredCustomizer", + ConcurrentPulsarListenerContainerFactoryCustomizer.class, + () -> (cf) -> cf.getContainerProperties().transactions().setEnabled(false)); context.refresh(); var container = context.getBean(PulsarListenerEndpointRegistry.class).getListenerContainer(LISTENER_ID); assertThat(container).isNotNull(); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java deleted file mode 100644 index 31ccd9c3..00000000 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java +++ /dev/null @@ -1,136 +0,0 @@ -/* - * Copyright 2023-2024 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.pulsar.listener; - -import org.apache.pulsar.client.api.PulsarClient; -import org.junit.jupiter.api.BeforeAll; -import org.testcontainers.containers.PulsarContainer; -import org.testcontainers.junit.jupiter.Testcontainers; - -import org.springframework.beans.factory.ObjectProvider; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.pulsar.annotation.EnablePulsar; -import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; -import org.springframework.pulsar.config.PulsarListenerContainerFactory; -import org.springframework.pulsar.core.ConsumerBuilderCustomizer; -import org.springframework.pulsar.core.DefaultPulsarClientFactory; -import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; -import org.springframework.pulsar.core.DefaultPulsarProducerFactory; -import org.springframework.pulsar.core.PulsarAdministration; -import org.springframework.pulsar.core.PulsarConsumerFactory; -import org.springframework.pulsar.core.PulsarProducerFactory; -import org.springframework.pulsar.core.PulsarTemplate; -import org.springframework.pulsar.test.support.PulsarTestContainerSupport; -import org.springframework.pulsar.transaction.PulsarAwareTransactionManager; -import org.springframework.pulsar.transaction.PulsarTransactionManager; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; - -/** - * Provides base support for tests that use Pulsar transactions. - * - * @author Chris Bono - */ -@SpringJUnitConfig -@DirtiesContext -@Testcontainers(disabledWithoutDocker = true) -class PulsarTxnTestsBase { - - static PulsarContainer PULSAR_CONTAINER = new PulsarContainer(PulsarTestContainerSupport.getPulsarImage()) - .withTransactions(); - - @BeforeAll - static void startContainer() { - PULSAR_CONTAINER.start(); - } - - @Autowired - protected PulsarClient pulsarClient; - - @Autowired - protected PulsarTemplate transactionalPulsarTemplate; - - @Configuration(proxyBeanMethods = false) - @EnablePulsar - static class TopLevelConfig { - - @Bean - PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { - return new DefaultPulsarProducerFactory<>(pulsarClient, "foo-1"); - } - - @Bean - PulsarClient pulsarClient() { - return new DefaultPulsarClientFactory((clientBuilder) -> { - clientBuilder.serviceUrl(PULSAR_CONTAINER.getPulsarBrokerUrl()); - clientBuilder.enableTransaction(true); - }).createClient(); - } - - @Bean - PulsarTemplate transactionalPulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - var template = new PulsarTemplate<>(pulsarProducerFactory); - template.transactions().setEnabled(true); - return template; - } - - @Bean - public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient, - ObjectProvider> defaultConsumerCustomizersProvider) { - return new DefaultPulsarConsumerFactory<>(pulsarClient, - defaultConsumerCustomizersProvider.orderedStream().toList()); - } - - @Bean - PulsarContainerProperties pulsarContainerProperties(PulsarAwareTransactionManager pulsarTransactionManager) { - var containerProps = new PulsarContainerProperties(); - containerProps.transactions().setEnabled(true); - containerProps.transactions().setRequired(false); - containerProps.transactions().setTransactionManager(pulsarTransactionManager); - return containerProps; - } - - @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( - PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps, - ObjectProvider containerPropsCustomizer) { - containerPropsCustomizer.ifAvailable((c) -> c.customize(pulsarContainerProps)); - return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, pulsarContainerProps); - } - - @Bean - PulsarAdministration pulsarAdministration() { - return new PulsarAdministration(PULSAR_CONTAINER.getHttpServiceUrl()); - } - - @Bean - PulsarAwareTransactionManager pulsarTransactionManager(PulsarClient pulsarClient) { - return new PulsarTransactionManager(pulsarClient); - } - - } - - @FunctionalInterface - interface PulsarContainerPropertiesCustomizer { - - void customize(PulsarContainerProperties containerProperties); - - } - -} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/transaction/PulsarTxnTestsBase.java b/spring-pulsar/src/test/java/org/springframework/pulsar/transaction/PulsarTxnTestsBase.java index 61a3c06d..702b2785 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/transaction/PulsarTxnTestsBase.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/transaction/PulsarTxnTestsBase.java @@ -20,6 +20,7 @@ import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; import java.util.List; +import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.PulsarClient; @@ -36,11 +37,11 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; -import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.core.ConsumerBuilderCustomizer; import org.springframework.pulsar.core.DefaultPulsarClientFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.ProducerBuilderCustomizer; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; @@ -59,7 +60,7 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @SpringJUnitConfig @DirtiesContext @Testcontainers(disabledWithoutDocker = true) -class PulsarTxnTestsBase { +public class PulsarTxnTestsBase { static PulsarContainer PULSAR_CONTAINER = new PulsarContainer(PulsarTestContainerSupport.getPulsarImage()) .withTransactions(); @@ -75,6 +76,16 @@ class PulsarTxnTestsBase { @Autowired protected PulsarTemplate transactionalPulsarTemplate; + protected PulsarTemplate newNonTransactionalTemplate(boolean sendInBatch, int numMessages) { + List> customizers = List.of(); + if (sendInBatch) { + customizers = List.of((pb) -> pb.enableBatching(true) + .batchingMaxPublishDelay(2, TimeUnit.SECONDS) + .batchingMaxMessages(numMessages)); + } + return new PulsarTemplate<>(new DefaultPulsarProducerFactory<>(pulsarClient, null, customizers)); + } + protected void assertThatMessagesAreInTopic(String topicOut, String... expectedMessages) { assertMessagesInTopic(topicOut).contains(expectedMessages); } @@ -94,7 +105,7 @@ class PulsarTxnTestsBase { @Configuration @EnablePulsar - static class TopLevelConfig { + public static class TopLevelConfig { @Bean PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { @@ -133,7 +144,7 @@ class PulsarTxnTestsBase { } @Bean - PulsarListenerContainerFactory pulsarListenerContainerFactory( + ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory( PulsarConsumerFactory pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps) { return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, pulsarContainerProps); }