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

Consolidate to single PulsarTxnTestsBase

Browse Source 
See #661
This commit is contained in:
Chris Bono
2024-05-07 16:09:04 -05:00
parent 26305ef994
commit ae1081742f
3 changed files with 37 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTxnTests.java
View File

@@ -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.<String>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<String> expectedMessages) {
assertThat(PulsarConsumerTestUtil.<String>consumeMessages(pulsarClient)
.fromTopic(topicOut)
.withSchema(Schema.STRING)
.awaitAtMost(Duration.ofSeconds(5))
.get()).map(Message::getValue).containsExactlyInAnyOrderElementsOf(expectedMessages);
}
private PulsarTemplate<String> newNonTransactionalTemplate(boolean sendInBatch, int numMessages) {
List<ProducerBuilderCustomizer<String>> 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();

136
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarTxnTestsBase.java
View File

@@ -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<String> transactionalPulsarTemplate;
@Configuration(proxyBeanMethods = false)
@EnablePulsar
static class TopLevelConfig {
@Bean
PulsarProducerFactory<String> 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<String> transactionalPulsarTemplate(PulsarProducerFactory<String> pulsarProducerFactory) {
var template = new PulsarTemplate<>(pulsarProducerFactory);
template.transactions().setEnabled(true);
return template;
}
@Bean
public PulsarConsumerFactory<?> pulsarConsumerFactory(PulsarClient pulsarClient,
ObjectProvider<ConsumerBuilderCustomizer<String>> 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<Object> pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps,
ObjectProvider<PulsarContainerPropertiesCustomizer> 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);
}
}

19
spring-pulsar/src/test/java/org/springframework/pulsar/transaction/PulsarTxnTestsBase.java
View File

@@ -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<String> transactionalPulsarTemplate;
protected PulsarTemplate<String> newNonTransactionalTemplate(boolean sendInBatch, int numMessages) {
List<ProducerBuilderCustomizer<String>> 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<String> pulsarProducerFactory(PulsarClient pulsarClient) {
@@ -133,7 +144,7 @@ class PulsarTxnTestsBase {
}
@Bean
PulsarListenerContainerFactory pulsarListenerContainerFactory(
ConcurrentPulsarListenerContainerFactory<?> pulsarListenerContainerFactory(
PulsarConsumerFactory<Object> pulsarConsumerFactory, PulsarContainerProperties pulsarContainerProps) {
return new ConcurrentPulsarListenerContainerFactory<>(pulsarConsumerFactory, pulsarContainerProps);
}