From bc38cecce5cdd803688c1074d4444d9a0e4a8cf6 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 8 Feb 2023 18:26:42 -0500 Subject: [PATCH] Basic binder compatibility tests * Ensuring that AbstractBinderTests and Partition capable tests work for Pulsar binder. Some tests needed to be overridden in order for them to work in a Pulsar context. Resolves https://github.com/spring-projects-experimental/spring-pulsar/issues/334 --- spring-pulsar-dependencies/build.gradle | 1 + .../build.gradle | 4 + .../binder/PulsarMessageChannelBinder.java | 72 +++- .../binder/AbstractPulsarTestBinder.java | 48 +++ .../stream/binder/PulsarBinderTests.java | 312 ++++++++++++++++++ .../cloud/stream/binder/PulsarTestBinder.java | 60 ++++ 6 files changed, 488 insertions(+), 9 deletions(-) create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/AbstractPulsarTestBinder.java create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java diff --git a/spring-pulsar-dependencies/build.gradle b/spring-pulsar-dependencies/build.gradle index f6b19afa..ed00b6f6 100644 --- a/spring-pulsar-dependencies/build.gradle +++ b/spring-pulsar-dependencies/build.gradle @@ -26,5 +26,6 @@ dependencies { api "org.apache.pulsar:pulsar-client-reactive-adapter:$pulsarClientReactiveVersion" api "org.apache.pulsar:pulsar-client-reactive-producer-cache-caffeine:$pulsarClientReactiveVersion" api "org.springframework.cloud:spring-cloud-stream:$springCloudStreamVersion" + api "org.springframework.cloud:spring-cloud-stream-test-support:$springCloudStreamVersion" } } diff --git a/spring-pulsar-spring-cloud-stream-binder/build.gradle b/spring-pulsar-spring-cloud-stream-binder/build.gradle index 92ecc470..f3f75e9f 100644 --- a/spring-pulsar-spring-cloud-stream-binder/build.gradle +++ b/spring-pulsar-spring-cloud-stream-binder/build.gradle @@ -12,6 +12,10 @@ dependencies { } testImplementation project(':spring-pulsar-test') testImplementation 'org.springframework.boot:spring-boot-starter-test' + testImplementation('org.springframework.cloud:spring-cloud-stream-test-support') { + exclude group: 'javax.activation', module: 'javax.activation-api' + exclude group: 'javax.annotation', module: 'javax.annotation-api' + } testImplementation 'org.awaitility:awaitility' testImplementation 'org.testcontainers:junit-jupiter' testImplementation 'org.testcontainers:pulsar' diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index e92c7b85..56774d87 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -31,9 +31,12 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.context.support.AbstractApplicationContext; import org.springframework.core.ResolvableType; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.support.management.ManageableLifecycle; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -92,15 +95,14 @@ public class PulsarMessageChannelBinder extends else { schema = null; } - return message -> { - try { - PulsarMessageChannelBinder.this.pulsarTemplate.sendAsync(destination.getName(), message.getPayload(), - schema); - } - catch (PulsarClientException ex) { - this.logger.trace("Failed to send message to destination: " + destination.getName(), ex); - } - }; + PulsarProducerConfigurationMessageHandler handler = new PulsarProducerConfigurationMessageHandler( + this.pulsarTemplate, schema, destination.getName()); + + AbstractApplicationContext applicationContext = getApplicationContext(); + handler.setApplicationContext(applicationContext); + handler.setBeanFactory(getBeanFactory()); + + return handler; } @Override @@ -217,4 +219,56 @@ public class PulsarMessageChannelBinder extends } + static class PulsarProducerConfigurationMessageHandler extends AbstractMessageProducingHandler + implements ManageableLifecycle { + + private boolean running = true; + + final PulsarTemplate pulsarTemplate; + + final Schema schema; + + final String destination; + + PulsarProducerConfigurationMessageHandler(PulsarTemplate pulsarTemplate, Schema schema, + String destination) { + this.pulsarTemplate = pulsarTemplate; + this.schema = schema; + this.destination = destination; + } + + @Override + public void start() { + try { + super.onInit(); + } + catch (Exception ex) { + this.logger.error(ex, "Initialization errors: "); + throw new RuntimeException(ex); + } + } + + @Override + public void stop() { + // TODO - should we close the underlyiung producer? + this.running = false; + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + protected void handleMessageInternal(Message message) { + try { + this.pulsarTemplate.sendAsync(this.destination, message.getPayload(), this.schema); + } + catch (PulsarClientException ex) { + logger.trace(ex, "Failed to send message to destination: " + this.destination); + } + } + + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/AbstractPulsarTestBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/AbstractPulsarTestBinder.java new file mode 100644 index 00000000..47f3ba19 --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/AbstractPulsarTestBinder.java @@ -0,0 +1,48 @@ +/* + * Copyright 2023 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.spring.cloud.stream.binder; + +import org.springframework.cloud.stream.binder.AbstractPollableConsumerTestBinder; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarProducerProperties; + +/** + * Base class for {@link PulsarTestBinder}. + * + * @author Soby Chacko + */ +public abstract class AbstractPulsarTestBinder extends + AbstractPollableConsumerTestBinder, ExtendedProducerProperties> { + + private ApplicationContext applicationContext; + + @Override + public void cleanup() { + } + + protected final void setApplicationContext(ApplicationContext context) { + this.applicationContext = context; + } + + public ApplicationContext getApplicationContext() { + return this.applicationContext; + } + +} diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java new file mode 100644 index 00000000..a9dcac7f --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderTests.java @@ -0,0 +1,312 @@ +/* + * Copyright 2023 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.spring.cloud.stream.binder; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; + +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.binder.PartitionCapableBinderTests; +import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; +import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.core.PulsarProducerFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBinderConfigurationProperties; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarProducerProperties; +import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner; +import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; + +/** + * Tests for {@link PulsarMessageChannelBinder}. + * + * @author Soby Chacko + */ +public class PulsarBinderTests extends + PartitionCapableBinderTests, ExtendedProducerProperties> + implements PulsarTestContainerSupport { + + private PulsarTestBinder binder; + + @Nullable + protected PulsarClient pulsarClient; + + @BeforeEach + void createPulsarClient() throws PulsarClientException { + pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); + } + + @AfterEach + void closePulsarClient() throws PulsarClientException { + if (pulsarClient != null && !pulsarClient.isClosed()) { + pulsarClient.close(); + } + } + + private final String CLASS_UNDER_TEST_NAME = PulsarMessageChannelBinder.class.getSimpleName(); + + @Override + protected boolean usesExplicitRouting() { + return false; + } + + @Override + protected String getClassUnderTestName() { + return CLASS_UNDER_TEST_NAME; + } + + @Override + protected PulsarTestBinder getBinder() { + PulsarAdministration pulsarAdministration = new PulsarAdministration( + Map.of("serviceUrl", PulsarTestContainerSupport.getHttpServiceUrl())); + PulsarBinderConfigurationProperties configurationProperties = new PulsarBinderConfigurationProperties(); + PulsarTopicProvisioner pulsarTopicProvisioner = new PulsarTopicProvisioner(pulsarAdministration, + configurationProperties); + + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + + Map config = Map.of("subscriptionInitialPosition", SubscriptionInitialPosition.Earliest); + DefaultPulsarConsumerFactory consumerFactory = new DefaultPulsarConsumerFactory<>(pulsarClient, config); + + if (this.binder == null) { + this.binder = new PulsarTestBinder(pulsarTopicProvisioner, pulsarTemplate, consumerFactory, + new DefaultSchemaResolver()); + } + return this.binder; + } + + @Override + protected ExtendedConsumerProperties createConsumerProperties() { + final ExtendedConsumerProperties pulsarConsumerProperties = new ExtendedConsumerProperties<>( + new PulsarConsumerProperties()); + return pulsarConsumerProperties; + } + + @Override + public Spy spyOn(String name) { + return null; + } + + private ExtendedProducerProperties createProducerProperties() { + return this.createProducerProperties(null); + } + + @Override + protected ExtendedProducerProperties createProducerProperties(TestInfo testInto) { + return new ExtendedProducerProperties<>(new PulsarProducerProperties()); + } + + @Override + protected void binderBindUnbindLatency() throws InterruptedException { + Thread.sleep(500); + } + + @Test + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public void testSendAndReceive(TestInfo testInfo) throws Exception { + Binder binder = getBinder(); + BindingProperties outputBindingProperties = createProducerBindingProperties(createProducerProperties()); + + DirectChannel moduleOutputChannel = createBindableChannel("output", outputBindingProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + Binding producerBinding = binder.bindProducer("foo.bar", moduleOutputChannel, + outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("foo.bar", null, moduleInputChannel, + consumerProperties); + + Message message = org.springframework.integration.support.MessageBuilder + .withPayload("foo".getBytes(StandardCharsets.UTF_8)).build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference<>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(new String(inboundMessageRef.get().getPayload(), StandardCharsets.UTF_8)).isEqualTo("foo"); + + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public void testAnonymousGroup(TestInfo testInfo) throws Exception { + Binder binder = getBinder(); + BindingProperties producerBindingProperties = createProducerBindingProperties( + createProducerProperties(testInfo)); + DirectChannel output = createBindableChannel("output", producerBindingProperties); + Binding producerBinding = binder.bindProducer( + String.format("defaultGroup%s0", getDestinationNameDelimiter()), output, + producerBindingProperties.getProducer()); + + QueueChannel input1 = new QueueChannel(); + Binding binding1 = binder.bindConsumer( + String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input1, + createConsumerProperties()); + + QueueChannel input2 = new QueueChannel(); + Binding binding2 = binder.bindConsumer( + String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input2, + createConsumerProperties()); + + String testPayload1 = "foo-" + UUID.randomUUID().toString(); + output.send(MessageBuilder.withPayload(testPayload1) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build()); + + Message receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload1); + + Message receivedMessage2 = (Message) receive(input2); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1); + + binding2.unbind(); + + String testPayload2 = "foo-" + UUID.randomUUID().toString(); + output.send(MessageBuilder.withPayload(testPayload2) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build()); + + binding2 = binder.bindConsumer(String.format("defaultGroup%s0", getDestinationNameDelimiter()), null, input2, + createConsumerProperties()); + String testPayload3 = "foo-" + UUID.randomUUID().toString(); + output.send(MessageBuilder.withPayload(testPayload3) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build()); + + receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload2); + receivedMessage1 = (Message) receive(input1); + assertThat(receivedMessage1).isNotNull(); + assertThat(new String(receivedMessage1.getPayload())).isEqualTo(testPayload3); + + receivedMessage2 = (Message) receive(input2); + assertThat(receivedMessage2).isNotNull(); + assertThat(new String(receivedMessage2.getPayload())).isEqualTo(testPayload1); + + producerBinding.unbind(); + binding1.unbind(); + binding2.unbind(); + } + + @Test + @Override + @SuppressWarnings({ "unchecked", "rawtypes" }) + public void testSendAndReceiveNoOriginalContentType(TestInfo testInfo) throws Exception { + Binder binder = getBinder(); + + BindingProperties producerBindingProperties = createProducerBindingProperties( + createProducerProperties(testInfo)); + DirectChannel moduleOutputChannel = createBindableChannel("output", producerBindingProperties); + BindingProperties inputBindingProperties = createConsumerBindingProperties(createConsumerProperties()); + DirectChannel moduleInputChannel = createBindableChannel("input", inputBindingProperties); + Binding producerBinding = binder.bindProducer( + String.format("bar%s0", getDestinationNameDelimiter()), moduleOutputChannel, + producerBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer( + String.format("bar%s0", getDestinationNameDelimiter()), "testSendAndReceiveNoOriginalContentType", + moduleInputChannel, createConsumerProperties()); + binderBindUnbindLatency(); + + Message message = MessageBuilder.withPayload("foo") + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference> inboundMessageRef = new AtomicReference>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.set((Message) message1); + } + finally { + latch.countDown(); + } + }); + + moduleOutputChannel.send(message); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + assertThat(inboundMessageRef.get()).isNotNull(); + assertThat(inboundMessageRef.get().getPayload()).isEqualTo("foo".getBytes()); + // TODO: The below content-type should be TEXT_PLAIN, but default to + // application/json + // This is because we don't currently preserve any message headers on send. We + // should look into this soon. + // Also, the content-type is "application/json" (with double quotes). We will have + // to fix that as well. + assertThat(inboundMessageRef.get().getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()) + .contains(MimeTypeUtils.APPLICATION_JSON.toString()); + producerBinding.unbind(); + consumerBinding.unbind(); + } + + @Test + @Override + @Disabled + public void testPartitionedModuleSpEL(TestInfo testInfo) throws Exception { + // This use-case needs to be further evaluated for Pulsar binder. + } + +} diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java new file mode 100644 index 00000000..99e9c7c7 --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarTestBinder.java @@ -0,0 +1,60 @@ +/* + * Copyright 2023 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.spring.cloud.stream.binder; + +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.SchemaResolver; +import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner; + +/** + * Test binder to exercise producer/consumer bindings in + * {@link PulsarMessageChannelBinder}. + * + * @author Soby Chacko + */ +public class PulsarTestBinder extends AbstractPulsarTestBinder { + + @SuppressWarnings({ "unchecked" }) + PulsarTestBinder(PulsarTopicProvisioner pulsarTopicProvisioner, PulsarTemplate pulsarTemplate, + PulsarConsumerFactory pulsarConsumerFactory, SchemaResolver schemaResolver) { + + try { + PulsarMessageChannelBinder binder = new PulsarMessageChannelBinder(pulsarTopicProvisioner, + (PulsarTemplate) pulsarTemplate, pulsarConsumerFactory, schemaResolver); + + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Config.class); + setApplicationContext(context); + binder.setApplicationContext(context); + binder.afterPropertiesSet(); + this.setPollableConsumerBinder(binder); + } + catch (Exception e) { + throw new RuntimeException(e); + } + } + + @Configuration + @EnableIntegration + static class Config { + + } + +}