From bd75eefdce3337c882dd070035c1d637f33226ff Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Sat, 21 Jan 2023 22:52:55 -0600 Subject: [PATCH] Add schema resolution to binder --- .../binder/PulsarMessageChannelBinder.java | 61 +++- .../properties/PulsarConsumerProperties.java | 57 +++- .../properties/PulsarProducerProperties.java | 39 ++- .../binder/PulsarBinderIntegrationTests.java | 285 +++++++++++++++--- ...essageChannelBinderResolveSchemaTests.java | 137 +++++++++ .../src/test/resources/logback-test.xml | 13 + src/checkstyle/checkstyle-suppressions.xml | 9 +- 7 files changed, 534 insertions(+), 67 deletions(-) create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml 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 9564f7f1..ea86db06 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 @@ -20,6 +20,7 @@ import java.util.Objects; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; @@ -30,8 +31,10 @@ 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.core.ResolvableType; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -77,10 +80,14 @@ public class PulsarMessageChannelBinder extends @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { - SchemaType schemaType = producerProperties.getExtension().getSchemaType(); - if (producerProperties.isUseNativeEncoding() && schemaType != null) { - Schema schema = Objects.requireNonNull(this.schemaResolver.getSchema(schemaType, null)); - this.pulsarTemplate.setSchema(schema); + + if (producerProperties.isUseNativeEncoding()) { + Schema schema = resolveSchema(producerProperties.getExtension().getSchemaType(), + producerProperties.getExtension().getMessageType(), + producerProperties.getExtension().getMessageKeyType(), + producerProperties.getExtension().getMessageValueType()); + this.pulsarTemplate.setSchema( + Objects.requireNonNull(schema, "Could not determine producer schema for " + destination.getName())); } return message -> { try { @@ -102,14 +109,18 @@ public class PulsarMessageChannelBinder extends Message message = MessageBuilder.withPayload(msg.getValue()).build(); pulsarMessageDrivenChannelAdapter.send(message); }); - SchemaType schemaType = properties.getExtension().getSchemaType(); - if (properties.isUseNativeDecoding() && schemaType != null) { - pulsarContainerProperties - .setSchema(Objects.requireNonNull(this.schemaResolver.getSchema(schemaType, null))); + + if (properties.isUseNativeDecoding()) { + Schema schema = resolveSchema(properties.getExtension().getSchemaType(), + properties.getExtension().getMessageType(), properties.getExtension().getMessageKeyType(), + properties.getExtension().getMessageValueType()); + pulsarContainerProperties.setSchema( + Objects.requireNonNull(schema, "Could not determine consumer schema for " + destination.getName())); } else { pulsarContainerProperties.setSchema(Schema.BYTES); } + String subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination); pulsarContainerProperties.setSubscriptionName(subscriptionName); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( @@ -118,6 +129,40 @@ public class PulsarMessageChannelBinder extends return pulsarMessageDrivenChannelAdapter; } + // VisibleForTesting + @Nullable + Schema resolveSchema(@Nullable SchemaType schemaType, @Nullable Class messageType, + @Nullable Class messageKeyType, @Nullable Class messageValueType) { + if (schemaType == null) { + schemaType = SchemaType.NONE; + } + ResolvableType resolvableType = null; + if (schemaType.isStruct()) { + resolvableType = ResolvableType.forClass(Objects.requireNonNull(messageType, + "'message-type' required for 'schema-type' " + schemaType.name())); + } + else if (schemaType == SchemaType.KEY_VALUE) { + resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class, + Objects.requireNonNull(messageKeyType, "'message-key-type' required for 'schema-type' KEY_VALUE"), + Objects.requireNonNull(messageValueType, + "'message-value-type' required for 'schema-type' KEY_VALUE")); + } + else if (schemaType == SchemaType.NONE) { + if (messageType != null) { + resolvableType = ResolvableType.forClass(messageType); + } + else if (messageKeyType != null && messageValueType != null) { + resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class, messageKeyType, messageValueType); + } + if (resolvableType == null) { + throw new IllegalArgumentException( + "'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE"); + } + } + // TODO if schema == null then default lookup bean Schema w/ name == binding + return this.schemaResolver.getSchema(schemaType, resolvableType); + } + @Override public PulsarConsumerProperties getExtendedConsumerProperties(String channelName) { return this.extendedBindingProperties.getExtendedConsumerProperties(channelName); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java index 5bf9938c..5324d4b2 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java @@ -25,17 +25,27 @@ import org.springframework.lang.Nullable; * Pulsar consumer properties used by the binder. * * @author Soby Chacko + * @author Chris Bono */ public class PulsarConsumerProperties { @Nullable private String subscriptionName; + @Nullable + private SubscriptionType subscriptionType; + @Nullable private SchemaType schemaType; @Nullable - private SubscriptionType subscriptionType; + private Class messageType; + + @Nullable + private Class messageKeyType; + + @Nullable + private Class messageValueType; @Nullable public String getSubscriptionName() { @@ -46,15 +56,6 @@ public class PulsarConsumerProperties { this.subscriptionName = subscriptionName; } - @Nullable - public SchemaType getSchemaType() { - return this.schemaType; - } - - public void setSchemaType(SchemaType schemaType) { - this.schemaType = schemaType; - } - @Nullable public SubscriptionType getSubscriptionType() { return this.subscriptionType; @@ -64,4 +65,40 @@ public class PulsarConsumerProperties { this.subscriptionType = subscriptionType; } + @Nullable + public SchemaType getSchemaType() { + return this.schemaType; + } + + public void setSchemaType(@Nullable SchemaType schemaType) { + this.schemaType = schemaType; + } + + @Nullable + public Class getMessageType() { + return this.messageType; + } + + public void setMessageType(@Nullable Class messageType) { + this.messageType = messageType; + } + + @Nullable + public Class getMessageKeyType() { + return this.messageKeyType; + } + + public void setMessageKeyType(@Nullable Class messageKeyType) { + this.messageKeyType = messageKeyType; + } + + @Nullable + public Class getMessageValueType() { + return this.messageValueType; + } + + public void setMessageValueType(@Nullable Class messageValueType) { + this.messageValueType = messageValueType; + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java index 0cdacc80..2edcb380 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java @@ -24,19 +24,56 @@ import org.springframework.lang.Nullable; * Pulsar producer properties used by the binder. * * @author Soby Chacko + * @author Chris Bono */ public class PulsarProducerProperties { @Nullable private SchemaType schemaType; + @Nullable + private Class messageType; + + @Nullable + private Class messageKeyType; + + @Nullable + private Class messageValueType; + @Nullable public SchemaType getSchemaType() { return this.schemaType; } - public void setSchemaType(SchemaType schemaType) { + public void setSchemaType(@Nullable SchemaType schemaType) { this.schemaType = schemaType; } + @Nullable + public Class getMessageType() { + return this.messageType; + } + + public void setMessageType(@Nullable Class messageType) { + this.messageType = messageType; + } + + @Nullable + public Class getMessageKeyType() { + return this.messageKeyType; + } + + public void setMessageKeyType(@Nullable Class messageKeyType) { + this.messageKeyType = messageKeyType; + } + + @Nullable + public Class getMessageValueType() { + return this.messageValueType; + } + + public void setMessageValueType(@Nullable Class messageValueType) { + this.messageValueType = messageValueType; + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java index 91aa3bbe..61566a2d 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java @@ -20,7 +20,10 @@ import java.time.Duration; import java.util.function.Consumer; import java.util.function.Supplier; +import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.common.schema.KeyValue; import org.awaitility.Awaitility; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.slf4j.Logger; @@ -34,66 +37,187 @@ import org.springframework.boot.test.system.CapturedOutput; import org.springframework.boot.test.system.OutputCaptureExtension; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.SchemaResolver.SchemaResolverCustomizer; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; /** + * Integration tests for {@link PulsarBinderIntegrationTests}. + * * @author Soby Chacko + * @author Chris Bono */ @ExtendWith(OutputCaptureExtension.class) class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { - @Test - void basicProducerConsumerBindingEndToEnd(CapturedOutput output) { - SpringApplication app = new SpringApplication(BasicScenarioConfig.class); - app.setWebApplicationType(WebApplicationType.NONE); - try (ConfigurableApplicationContext ignored = app.run( - "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), - "--spring.cloud.function.definition=textSupplier;textLogger", - "--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0", - "--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription-name=basic-scenario-sub-1")) { - Awaitility.await().atMost(Duration.ofSeconds(10)) - .until(() -> output.toString().contains("Hello binder: test-basic-scenario")); + @Nested + class DefaultEncoding { + + @Test + void primitiveTypeString(CapturedOutput output) { + SpringApplication app = new SpringApplication(PrimitiveTextConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=textSupplier;textLogger", + "--spring.cloud.stream.bindings.textLogger-in-0.destination=textSupplier-out-0", + "--spring.cloud.stream.pulsar.bindings.textLogger-in-0.consumer.subscription-name=pbit-text-sub1")) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: test-basic-scenario")); + } } + + @Test + void primitiveTypeFloat(CapturedOutput output) { + SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=piSupplier;piLogger", + "--spring.cloud.stream.bindings.piSupplier-out-0.destination=pi-stream", + "--spring.cloud.stream.bindings.piLogger-in-0.destination=pi-stream", + "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub1")) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: 3.14")); + } + } + } - @Test - void useNativeEncodingDecodingWorkAsExpected(CapturedOutput output) { - SpringApplication app = new SpringApplication(PiStreamConfig.class); - app.setWebApplicationType(WebApplicationType.NONE); - try (ConfigurableApplicationContext ignored = app.run( - "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), - "--spring.cloud.function.definition=piSupplier;piLogger", - "--spring.cloud.stream.bindings.piLogger-in-0.destination=piSupplier-out-0", - "--spring.cloud.stream.bindings.piSupplier-out-0.producer.use-native-encoding=true", - "--spring.cloud.stream.bindings.piLogger-in-0.consumer.use-native-decoding=true", - "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.schema-type=FLOAT", - "--spring.cloud.stream.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT", - "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=native-encoding-decoding-sub-1")) { - Awaitility.await().atMost(Duration.ofSeconds(10)) - .until(() -> output.toString().contains("Hello binder: 3.14")); - } - } + @Nested + class NativeEncoding { - @Test - void basicProducerConsumerBindingEndToEndWithNonTextPayloadType(CapturedOutput output) { - SpringApplication app = new SpringApplication(PiStreamConfig.class); - app.setWebApplicationType(WebApplicationType.NONE); - try (ConfigurableApplicationContext ignored = app.run( - "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), - "--spring.cloud.function.definition=piSupplier;piLogger", - "--spring.cloud.stream.bindings.piSupplier-out-0.destination=pi-stream", - "--spring.cloud.stream.bindings.piLogger-in-0.destination=pi-stream", - "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=native-encoding-decoding-sub-2")) { - Awaitility.await().atMost(Duration.ofSeconds(10)) - .until(() -> output.toString().contains("Hello binder: 3.14")); + @Test + void primitiveTypeFloat(CapturedOutput output) { + SpringApplication app = new SpringApplication(PrimitiveFloatConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=piSupplier;piLogger", + "--spring.cloud.stream.bindings.piLogger-in-0.destination=piSupplier-out-0", + "--spring.cloud.stream.bindings.piSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.piLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.schema-type=FLOAT", + "--spring.cloud.stream.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT", + "--spring.cloud.stream.pulsar.bindings.piLogger-in-0.consumer.subscription-name=pbit-float-sub2")) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: 3.14")); + } } + + @Test + void jsonTypeFoo(CapturedOutput output) { + SpringApplication app = new SpringApplication(JsonFooConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=fooSupplier-out-0", + "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.schema-type=JSON", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub1", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.schema-type=JSON", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type=" + + Foo.class.getName())) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: Foo[value=5150]")); + } + } + + @Test + void jsonTypeFooWithNoSchemaTypeAndCustomFooMapping(CapturedOutput output) { + SpringApplication app = new SpringApplication(JsonFooWithCustomMappingConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-2", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-2", + "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub2", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type=" + + Foo.class.getName())) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: Foo[value=5150]")); + } + } + + @Test + void jsonTypeFooWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) { + SpringApplication app = new SpringApplication(JsonFooConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-3", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-3", + "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub3", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-type=" + + Foo.class.getName())) { + Awaitility.await().atMost(Duration.ofSeconds(10)).until( + () -> output.toString().contains("Could not determine producer schema for foo-stream-3")); + } + } + + @Test + void jsonTypeFooConsumerWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) { + SpringApplication app = new SpringApplication(JsonFooConsumerConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-4", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-4", + "--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-type=" + Foo.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub4")) { + Awaitility.await().atMost(Duration.ofSeconds(10)).until( + () -> output.toString().contains("Could not determine consumer schema for foo-stream-4")); + } + } + + @Test + void keyValueTypeWithCustomFooMapping(CapturedOutput output) { + SpringApplication app = new SpringApplication(KeyValueFooConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=kv-stream-1", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=kv-stream-1", + "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.bindings.fooLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-value-type=" + + Foo.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-kv-sub1", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-value-type=" + + Foo.class.getName())) { + Awaitility.await().atMost(Duration.ofSeconds(10)) + .until(() -> output.toString().contains("Hello binder: 5150->Foo[value=5150]")); + } + } + } @EnableAutoConfiguration @SpringBootConfiguration - static class BasicScenarioConfig { + static class PrimitiveTextConfig { - private final Logger logger = LoggerFactory.getLogger(BasicScenarioConfig.class); + private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean public Supplier textSupplier() { @@ -109,9 +233,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { @EnableAutoConfiguration @SpringBootConfiguration - static class PiStreamConfig { + static class PrimitiveFloatConfig { - private final Logger logger = LoggerFactory.getLogger(BasicScenarioConfig.class); + private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean public Supplier piSupplier() { @@ -125,4 +249,77 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { } + @EnableAutoConfiguration + @SpringBootConfiguration + static class JsonFooConsumerConfig { + + private final Logger logger = LoggerFactory.getLogger(getClass()); + + public Supplier fooSupplier() { + return () -> "5150"; + } + + @Bean + public Consumer fooLogger() { + return f -> this.logger.info("Hello binder: " + f); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + static class JsonFooConfig { + + private final Logger logger = LoggerFactory.getLogger(getClass()); + + @Bean + public Supplier fooSupplier() { + return () -> new Foo("5150"); + } + + @Bean + public Consumer fooLogger() { + return f -> this.logger.info("Hello binder: " + f); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(JsonFooConfig.class) + static class JsonFooWithCustomMappingConfig { + + @Bean + public SchemaResolverCustomizer customMappings() { + return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class)); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + static class KeyValueFooConfig { + + private final Logger logger = LoggerFactory.getLogger(getClass()); + + @Bean + public SchemaResolverCustomizer customMappings() { + return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class)); + } + + @Bean + public Supplier> fooSupplier() { + return () -> new KeyValue<>("5150", new Foo("5150")); + } + + @Bean + public Consumer> fooLogger() { + return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue()); + } + + } + + record Foo(String value) { + } + } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java new file mode 100644 index 00000000..e0eda44f --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java @@ -0,0 +1,137 @@ +/* + * 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.AssertionsForClassTypes.assertThatExceptionOfType; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import org.apache.pulsar.common.schema.KeyValue; +import org.apache.pulsar.common.schema.SchemaType; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.EnumSource.Mode; + +import org.springframework.core.ResolvableType; +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; + +/** + * Unit tests for {@link PulsarMessageChannelBinder#resolveSchema}. + * + * @author Chris Bono + */ +public class PulsarMessageChannelBinderResolveSchemaTests { + + private SchemaResolver resolver = mock(SchemaResolver.class); + + @SuppressWarnings("unchecked") + private PulsarMessageChannelBinder binder = new PulsarMessageChannelBinder(mock(PulsarTopicProvisioner.class), + mock(PulsarTemplate.class), mock(PulsarConsumerFactory.class), resolver); + + @ParameterizedTest + @EnumSource(mode = Mode.MATCH_NONE, names = "^(AUTO.*|AVRO|JSON|KEY_VALUE|NONE|PROTOBUF.*)$") + void primitiveSchemaTypes(SchemaType schemaType) { + binder.resolveSchema(schemaType, null, null, null); + verify(resolver).getSchema(schemaType, null); + } + + @ParameterizedTest + @EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$") + void structSchemaTypes(SchemaType schemaType) { + binder.resolveSchema(schemaType, Foo.class, null, null); + verify(resolver).getSchema(schemaType, ResolvableType.forClass(Foo.class)); + } + + @Test + void keyValueSchemaType() { + binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, Bar.class); + verify(resolver).getSchema(SchemaType.KEY_VALUE, + ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class)); + } + + @ParameterizedTest + @EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$") + void structSchemaTypesRequireMessageType(SchemaType schemaType) { + assertThatExceptionOfType(NullPointerException.class) + .isThrownBy(() -> binder.resolveSchema(schemaType, null, null, null)) + .withMessage("'message-type' required for 'schema-type' " + schemaType.name()); + } + + @Test + void keyValueSchemaTypeRequiresKeyAndValueTypes() { + assertThatExceptionOfType(NullPointerException.class) + .isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, null, null)) + .withMessage("'message-key-type' required for 'schema-type' KEY_VALUE"); + assertThatExceptionOfType(NullPointerException.class) + .isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, null, Bar.class)) + .withMessage("'message-key-type' required for 'schema-type' KEY_VALUE"); + assertThatExceptionOfType(NullPointerException.class) + .isThrownBy(() -> binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, null)) + .withMessage("'message-value-type' required for 'schema-type' KEY_VALUE"); + } + + @Nested + class SchemaTypeNone { + + @Test + void withMesssageType() { + binder.resolveSchema(SchemaType.NONE, Foo.class, null, null); + verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); + } + + @Test + void withKeyAndValueTypes() { + binder.resolveSchema(SchemaType.NONE, null, Foo.class, Bar.class); + verify(resolver).getSchema(SchemaType.NONE, + ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class)); + } + + @Test + void withMessageTypeAndKeyAndValueTypes() { + binder.resolveSchema(SchemaType.NONE, Foo.class, String.class, Bar.class); + verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); + } + + @Test + void withOnlyKeyType() { + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> binder.resolveSchema(SchemaType.NONE, null, Foo.class, null)).withMessage( + "'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE"); + } + + @Test + void withOnlyValueType() { + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> binder.resolveSchema(SchemaType.NONE, null, null, Foo.class)).withMessage( + "'message-type' OR ('message-key-type' AND 'message-value-type') required for 'schema-type' NONE"); + + } + + } + + record Foo(String value) { + } + + record Bar(String value) { + } + +} diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml b/spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml new file mode 100644 index 00000000..e142997d --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/resources/logback-test.xml @@ -0,0 +1,13 @@ + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n + + + + + + + + + diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index c3a5d455..9c7b024b 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -3,10 +3,11 @@ "-//Checkstyle//DTD SuppressionFilter Configuration 1.2//EN" "https://checkstyle.org/dtds/suppressions_1_2.dtd"> - - - - + + + + +