From 6fedfa9c0c05e8e88c24b078092a897469de88de Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Tue, 14 Mar 2023 16:16:53 -0500 Subject: [PATCH] Allow schema mapping via config props (#371) * Allow schema mapping via config props * Adds a `schema-info` property to the `spring.pulsar.defaults.type-mappings` which allows users to configure a default schema by message type via config props * Add docs for schema mapping via config props * Fix unit test --- .../src/main/asciidoc/pulsar-binder.adoc | 6 +- .../src/main/asciidoc/pulsar.adoc | 2 + .../src/main/asciidoc/reactive-pulsar.adoc | 2 + .../schema-info/custom-schema-mapping.adoc | 38 ++ .../schema-info/schema-info-listener.adoc | 15 +- .../schema-info/schema-info-template.adoc | 14 +- .../PulsarAutoConfiguration.java | 16 +- .../autoconfigure/PulsarProperties.java | 39 +- .../PulsarAutoConfigurationTests.java | 90 ++++- .../autoconfigure/PulsarPropertiesTests.java | 67 +++- .../binder/PulsarMessageChannelBinder.java | 61 +-- .../properties/PulsarConsumerProperties.java | 18 +- .../properties/PulsarProducerProperties.java | 18 +- .../binder/PulsarBinderIntegrationTests.java | 355 ++++++++++++++---- ...essageChannelBinderResolveSchemaTests.java | 151 -------- .../pulsar/core/DefaultSchemaResolver.java | 2 +- .../springframework/pulsar/core/Resolved.java | 14 +- .../pulsar/core/SchemaResolver.java | 42 +++ .../core/DefaultSchemaResolverTests.java | 10 +- .../pulsar/core/ResolvedTests.java | 61 +++ .../pulsar/core/SchemaResolverTests.java | 134 +++++++ 21 files changed, 788 insertions(+), 367 deletions(-) create mode 100644 spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc delete 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/src/test/java/org/springframework/pulsar/core/ResolvedTests.java create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/core/SchemaResolverTests.java diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc index 171b630f..420dd6bd 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc @@ -174,11 +174,11 @@ spring: message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time timeProcessor-out-0: producer: - schema-type: JSON + schema-type: AVRO message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime timeLogger-in-0: consumer: - schema-type: JSON + schema-type: AVRO message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime ---- @@ -191,6 +191,8 @@ This information is provided as extended binding properties. As you can see above in the configuration, the properties are - `spring.cloud.stream.pulsar.bindings..producer|consumer.schema-type` for schema information and `spring.cloud.stream.pulsar.bindings..producer|consumer.message-type` for the actual target type. If you have both keys and values on the message, you can use `message-key-type` and `message-value-type` to specify their target types. +TIP: Any configured <> will be consulted when the `schema-type` property is omitted. + === Message Header Conversion Each message typically has header information that needs to be carried along as the message traverses between Pulsar and Spring Messaging via Spring Cloud Stream input and output bindings. To support this traversal, the framework handles the necessary message header conversion. diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 0b4038e4..f066663b 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -91,6 +91,7 @@ template.newMessage(msg) ---- ==== +[[schema-info-template-imperative]] :template-class: PulsarTemplate include::schema-info/schema-info-template.adoc[leveloffset=+1] @@ -318,6 +319,7 @@ void listen(String message) { TIP: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.consumer` application configuration properties +[[schema-info-listener-imperative]] :listener-class: PulsarListener include::schema-info/schema-info-listener.adoc[leveloffset=+1] diff --git a/spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc index be6545a5..592ef957 100644 --- a/spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc @@ -105,6 +105,7 @@ template.newMessage(msg) TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.reactive.sender.message-routing-mode` is `custom`. +[[schema-info-template-reactive]] :template-class: ReactivePulsarTemplate include::schema-info/schema-info-template.adoc[leveloffset=+1] @@ -305,6 +306,7 @@ ReactiveMessageConsumerBuilderCustomizer directConsumerPropsCustomizer() CAUTION: The properties used are direct Pulsar consumer properties, not the `spring.pulsar.reactive.consumer` Spring Boot configuration properties +[[schema-info-listener-reactive]] :listener-class: ReactivePulsarListener include::schema-info/schema-info-listener.adoc[leveloffset=+1] diff --git a/spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc b/spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc new file mode 100644 index 00000000..85233ff1 --- /dev/null +++ b/spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc @@ -0,0 +1,38 @@ +Schema mappings can be configured with the `spring.pulsar.defaults.type-mappings` property. +The following example uses `application.yml` to add mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: + +[source,yaml,indent=0,subs="verbatim"] +---- +spring: + pulsar: + defaults: + type-mappings: + - message-type: com.acme.User + schema-info: + schema-type: AVRO + message-type: com.acme.User + - message-type: com.acme.Address + schema-info: + schema-type: JSON + message-type: com.acme.Address +---- + +NOTE: The `message-type` is the fully-qualified name of the message class. + +The preferred method of adding mappings is via the property mentioned above. +However, if more control is needed you can provide a schema resolver customizer to add the mapping(s). + +The following example uses a schema resolver customizer to add mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: + +==== +[source, java] +---- +@Bean +public SchemaResolverCustomizer schemaResolverCustomizer() { + return (schemaResolver) -> { + schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class)); + } +} +---- +==== diff --git a/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc b/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc index 0e24a6b6..88a0e280 100644 --- a/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc @@ -9,20 +9,7 @@ IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PRO As an alternative to specifying the schema on the `{listener-class}` for complex types, the schema resolver can be configured with mappings for the types. This removes the need to set the schema on the listener as the framework consults the resolver using the incoming message type. -The following example shows a schema resolver customizer that adds mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: - -==== -[source, java] ----- -@Bean -public SchemaResolverCustomizer schemaResolverCustomizer() { - return (schemaResolver) -> { - schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); - schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class)); - } -} ----- -==== +include::custom-schema-mapping.adoc[] With this configuration in place, there is no need to set the schema on the listener, for example: diff --git a/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc b/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc index 10395369..292788dd 100644 --- a/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc @@ -8,18 +8,6 @@ IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PRO As an alternative to specifying the schema when invoking send operations on the `{template-class}` for complex types, the schema resolver can be configured with mappings for the types. This removes the need to specify the schema as the framework consults the resolver using the outgoing message type. -The following example shows a schema resolver customizer that adds mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: +include::custom-schema-mapping.adoc[] -==== -[source, java] ----- -@Bean -public SchemaResolverCustomizer schemaResolverCustomizer() { - return (schemaResolver) -> { - schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); - schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class)); - } -} ----- -==== With this configuration in place, there is no need to set specify the schema on send operations. diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 64cdd32f..adb87fdc 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -111,9 +111,17 @@ public class PulsarAutoConfiguration { @Bean @ConditionalOnMissingBean(SchemaResolver.class) - public DefaultSchemaResolver schemaResolver( + public DefaultSchemaResolver schemaResolver(PulsarProperties pulsarProperties, Optional> schemaResolverCustomizer) { - DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + var schemaResolver = new DefaultSchemaResolver(); + if (pulsarProperties.getDefaults().getTypeMappings() != null) { + pulsarProperties.getDefaults().getTypeMappings().stream().filter((tm) -> tm.schemaInfo() != null) + .forEach((tm) -> { + var schema = schemaResolver.resolveSchema(tm.schemaInfo().schemaType(), + tm.schemaInfo().messageType(), tm.schemaInfo().messageKeyType()).orElseThrow(); + schemaResolver.addCustomSchemaMapping(tm.messageType(), schema); + }); + } schemaResolverCustomizer.ifPresent((customizer) -> customizer.customize(schemaResolver)); return schemaResolver; } @@ -121,9 +129,9 @@ public class PulsarAutoConfiguration { @Bean @ConditionalOnMissingBean(TopicResolver.class) public DefaultTopicResolver topicResolver(PulsarProperties pulsarProperties) { - DefaultTopicResolver topicResolver = new DefaultTopicResolver(); + var topicResolver = new DefaultTopicResolver(); if (pulsarProperties.getDefaults().getTypeMappings() != null) { - pulsarProperties.getDefaults().getTypeMappings() + pulsarProperties.getDefaults().getTypeMappings().stream().filter((tm) -> tm.topicName() != null) .forEach((tm) -> topicResolver.addCustomTopicMapping(tm.messageType(), tm.topicName())); } return topicResolver; diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index 2d1f12e2..018aab35 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import org.apache.pulsar.client.api.ProxyProtocol; @@ -29,6 +30,7 @@ import org.apache.pulsar.common.schema.SchemaType; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.NestedConfigurationProperty; import org.springframework.boot.context.properties.PropertyMapper; +import org.springframework.lang.Nullable; import org.springframework.pulsar.listener.AckMode; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -1327,9 +1329,9 @@ public class PulsarProperties { public static class Defaults { /** - * List of mappings from message type to topic name to use as a default topic when - * a topic is not explicitly specified when producing or consuming messages of the - * mapped type. + * List of mappings from message type to topic name and schema info to use as a + * defaults when a topic name and/or schema is not explicitly specified when + * producing or consuming messages of the mapped type. */ private List typeMappings = new ArrayList<>(); @@ -1344,11 +1346,36 @@ public class PulsarProperties { } /** - * A mapping from message type to topic and schema - used as defaults for the type. + * A mapping from message type to topic and/or schema info to use (at least one of + * {@code topicName} or {@code schemaInfo} must be specified. * @param messageType the message type - * @param topicName the default topic name to use for the type + * @param topicName the topic name + * @param schemaInfo the schema info */ - record TypeMapping(Class messageType, String topicName) { + public record TypeMapping(Class messageType, @Nullable String topicName, @Nullable SchemaInfo schemaInfo) { + public TypeMapping { + Objects.requireNonNull(messageType, "messageType must not be null"); + if (topicName == null && schemaInfo == null) { + throw new IllegalArgumentException("At least one of topicName or schemaInfo must not be null"); + } + } + } + + /** + * Represents a schema - holds enough information to construct an actual schema + * instance. + * @param schemaType schema type + * @param messageType message type (not required for primitive schema types or key + * value type) + * @param messageKeyType message key type (required for key value type) + */ + public record SchemaInfo(SchemaType schemaType, @Nullable Class messageType, @Nullable Class messageKeyType) { + public SchemaInfo { + Objects.requireNonNull(schemaType, "schemaType must not be null"); + if (schemaType == SchemaType.NONE) { + throw new IllegalArgumentException("schemaType NONE not supported"); + } + } } static class Properties extends HashMap { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 0c1fa5fb..74529576 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -28,6 +28,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionInitialPosition; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; +import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; import org.assertj.core.api.AbstractObjectAssert; import org.assertj.core.api.InstanceOfAssertFactories; @@ -154,22 +155,6 @@ class PulsarAutoConfigurationTests { .isSameAs(customTopicResolver)); } - @Test - void defaultTypeMappingsAreAppliedToTopicResolver() { - record Foo() { - } - contextRunner - .withPropertyValues( - "spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()), - "spring.pulsar.defaults.type-mappings[0].topic-name=foo-topic", - "spring.pulsar.defaults.type-mappings[1].message-type=%s".formatted(String.class.getName()), - "spring.pulsar.defaults.type-mappings[1].topic-name=string-topic") - .run((context -> assertThat(context).hasNotFailed().getBean(TopicResolver.class) - .asInstanceOf(InstanceOfAssertFactories.type(DefaultTopicResolver.class)) - .extracting(DefaultTopicResolver::getCustomTopicMappings, InstanceOfAssertFactories.MAP) - .containsOnly(entry(Foo.class, "foo-topic"), entry(String.class, "string-topic")))); - } - @Test void customPulsarProducerFactoryIsRespected() { PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); @@ -302,6 +287,79 @@ class PulsarAutoConfigurationTests { })); } + @Nested + class DefaultsTypeMappingsTests { + + @Test + void topicMappingsAreAddedToTopicResolver() { + contextRunner + .withPropertyValues( + "spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()), + "spring.pulsar.defaults.type-mappings[0].topic-name=foo-topic", + "spring.pulsar.defaults.type-mappings[1].message-type=%s".formatted(String.class.getName()), + "spring.pulsar.defaults.type-mappings[1].topic-name=string-topic") + .run((context -> assertThat(context).hasNotFailed().getBean(TopicResolver.class) + .asInstanceOf(InstanceOfAssertFactories.type(DefaultTopicResolver.class)) + .extracting(DefaultTopicResolver::getCustomTopicMappings, InstanceOfAssertFactories.MAP) + .containsOnly(entry(Foo.class, "foo-topic"), entry(String.class, "string-topic")))); + } + + @Test + void schemaMappingForPrimitiveIsAddedToSchemaResolver() { + contextRunner + .withPropertyValues( + "spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()), + "spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=STRING") + .run((context -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) + .asInstanceOf(InstanceOfAssertFactories.type(DefaultSchemaResolver.class)) + .extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP) + .containsOnly(entry(Foo.class, Schema.STRING)))); + } + + @Test + void schemaMappingForStructIsAddedToSchemaResolver() { + contextRunner + .withPropertyValues( + "spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()), + "spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=JSON", + "spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s" + .formatted(Foo.class.getName())) + .run((context -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) + .asInstanceOf(InstanceOfAssertFactories.type(DefaultSchemaResolver.class)) + .extracting(DefaultSchemaResolver::getCustomSchemaMappings, + InstanceOfAssertFactories.map(Class.class, Schema.class)) + .hasEntrySatisfying(Foo.class, + (schema) -> assertSchemaEquals(schema, Schema.JSON(Foo.class))))); + } + + @Test + void schemaMappingForKeyValueIsAddedToSchemaResolver() { + contextRunner + .withPropertyValues( + "spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(Foo.class.getName()), + "spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=%s" + .formatted(SchemaType.KEY_VALUE.name()), + "spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s" + .formatted(Foo.class.getName()), + "spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type=%s" + .formatted(String.class.getName())) + .run((context -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) + .asInstanceOf(InstanceOfAssertFactories.type(DefaultSchemaResolver.class)) + .extracting(DefaultSchemaResolver::getCustomSchemaMappings, + InstanceOfAssertFactories.map(Class.class, Schema.class)) + .hasEntrySatisfying(Foo.class, (schema) -> assertSchemaEquals(schema, Schema + .KeyValue(Schema.STRING, Schema.JSON(Foo.class), KeyValueEncodingType.INLINE))))); + } + + private void assertSchemaEquals(Schema left, Schema right) { + assertThat(left.getSchemaInfo()).isEqualTo(right.getSchemaInfo()); + } + + record Foo() { + } + + } + @Nested class ClientAutoConfigurationTests { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java index 76b81bec..9ed6e0ff 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.autoconfigure; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatRuntimeException; @@ -42,14 +43,17 @@ import org.apache.pulsar.client.impl.conf.ConfigurationDataUtils; import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.apache.pulsar.client.impl.conf.ReaderConfigurationData; +import org.apache.pulsar.common.schema.SchemaType; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.springframework.boot.context.properties.bind.BindException; import org.springframework.boot.context.properties.bind.Bindable; import org.springframework.boot.context.properties.bind.Binder; import org.springframework.boot.context.properties.source.ConfigurationPropertySource; import org.springframework.boot.context.properties.source.MapConfigurationPropertySource; +import org.springframework.pulsar.autoconfigure.PulsarProperties.SchemaInfo; import org.springframework.pulsar.autoconfigure.PulsarProperties.TypeMapping; /** @@ -301,23 +305,76 @@ public class PulsarPropertiesTests { } @Nested - class DefaultsPropertiesTests { + class DefaultsTypeMappingsPropertiesTests { @Test - void defaultsTypeMappingsEmptyByDefault() { + void emptyByDefault() { assertThat(properties.getDefaults().getTypeMappings()).isEmpty(); } @Test - void defaultsTypeMappings() { + void withTopicsOnly() { Map props = new HashMap<>(); props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); props.put("spring.pulsar.defaults.type-mappings[0].topic-name", "foo-topic"); props.put("spring.pulsar.defaults.type-mappings[1].message-type", String.class.getName()); props.put("spring.pulsar.defaults.type-mappings[1].topic-name", "string-topic"); bind(props); - assertThat(properties.getDefaults().getTypeMappings()).hasSize(2).containsExactly( - new TypeMapping(Foo.class, "foo-topic"), new TypeMapping(String.class, "string-topic")); + assertThat(properties.getDefaults().getTypeMappings()).containsExactly( + new TypeMapping(Foo.class, "foo-topic", null), new TypeMapping(String.class, "string-topic", null)); + } + + @Test + void withSchemaOnly() { + Map props = new HashMap<>(); + props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON"); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName()); + bind(props); + assertThat(properties.getDefaults().getTypeMappings()).containsExactly( + new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.JSON, Foo.class, null))); + } + + @Test + void withTopicAndSchema() { + Map props = new HashMap<>(); + props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].topic-name", "foo-topic"); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "JSON"); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName()); + bind(props); + assertThat(properties.getDefaults().getTypeMappings()).containsExactly( + new TypeMapping(Foo.class, "foo-topic", new SchemaInfo(SchemaType.JSON, Foo.class, null))); + } + + @Test + void withKeyValueSchema() { + Map props = new HashMap<>(); + props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "KEY_VALUE"); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-key-type", String.class.getName()); + bind(props); + assertThat(properties.getDefaults().getTypeMappings()).containsExactly( + new TypeMapping(Foo.class, null, new SchemaInfo(SchemaType.KEY_VALUE, Foo.class, String.class))); + } + + @Test + void schemaTypeRequired() { + Map props = new HashMap<>(); + props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.message-type", Foo.class.getName()); + assertThatExceptionOfType(BindException.class).isThrownBy(() -> bind(props)).havingRootCause() + .withMessageContaining("schemaType must not be null"); + } + + @Test + void schemaTypeNoneNotAllowed() { + Map props = new HashMap<>(); + props.put("spring.pulsar.defaults.type-mappings[0].message-type", Foo.class.getName()); + props.put("spring.pulsar.defaults.type-mappings[0].schema-info.schema-type", "NONE"); + assertThatExceptionOfType(BindException.class).isThrownBy(() -> bind(props)).havingRootCause() + .withMessageContaining("schemaType NONE not supported"); } record Foo(String value) { 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 ae9f63ff..495195d8 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 @@ -16,11 +16,10 @@ package org.springframework.pulsar.spring.cloud.stream.binder; -import java.util.Objects; +import java.util.Optional; 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; @@ -32,7 +31,6 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; 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.integration.handler.AbstractMessageProducingHandler; @@ -101,16 +99,16 @@ public class PulsarMessageChannelBinder extends ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { final Schema schema; if (producerProperties.isUseNativeEncoding()) { - schema = resolveSchema(producerProperties.getExtension().getSchemaType(), - producerProperties.getExtension().getMessageType(), - producerProperties.getExtension().getMessageKeyType(), - producerProperties.getExtension().getMessageValueType()); - Objects.requireNonNull(schema, "Could not determine producer schema for " + destination.getName()); + var schemaType = Optional.ofNullable(producerProperties.getExtension().getSchemaType()) + .orElse(SchemaType.NONE); + schema = this.schemaResolver + .resolveSchema(schemaType, producerProperties.getExtension().getMessageType(), + producerProperties.getExtension().getMessageKeyType()) + .orElseThrow(() -> "Could not determine producer schema for " + destination.getName()); } else { schema = null; } - var baseProducerProps = new ProducerConfigProperties().buildProperties(); var binderProducerProps = this.binderConfigProps.getProducer().buildProperties(); var bindingProducerProps = producerProperties.getExtension().buildProperties(); @@ -152,11 +150,12 @@ public class PulsarMessageChannelBinder extends }); if (properties.isUseNativeDecoding()) { - var schema = resolveSchema(properties.getExtension().getSchemaType(), - properties.getExtension().getMessageType(), properties.getExtension().getMessageKeyType(), - properties.getExtension().getMessageValueType()); - containerProperties.setSchema( - Objects.requireNonNull(schema, "Could not determine consumer schema for " + destination.getName())); + var schemaType = Optional.ofNullable(properties.getExtension().getSchemaType()).orElse(SchemaType.NONE); + var schema = this.schemaResolver + .resolveSchema(schemaType, properties.getExtension().getMessageType(), + properties.getExtension().getMessageKeyType()) + .orElseThrow(() -> "Could not determine consumer schema for " + destination.getName()); + containerProperties.setSchema(schema); } else { containerProperties.setSchema(Schema.BYTES); @@ -186,40 +185,6 @@ public class PulsarMessageChannelBinder extends return new PulsarBinderHeaderMapper(this.headerMapper); } - // 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.resolveSchema(schemaType, resolvableType).get().orElse(null); - } - @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 5451b2b8..d360c31c 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 @@ -42,17 +42,12 @@ public class PulsarConsumerProperties extends ConsumerConfigProperties { private Class messageType; /** - * Type for the Pulsar message key. + * Pulsar message key type for this binding (only used when schema type is + * {@code }KEY_VALUE}). */ @Nullable private Class messageKeyType; - /** - * Type for the Pulsar message value. - */ - @Nullable - private Class messageValueType; - /** * Number of topic partitions. */ @@ -86,15 +81,6 @@ public class PulsarConsumerProperties extends ConsumerConfigProperties { this.messageKeyType = messageKeyType; } - @Nullable - public Class getMessageValueType() { - return this.messageValueType; - } - - public void setMessageValueType(@Nullable Class messageValueType) { - this.messageValueType = messageValueType; - } - @Nullable public Integer getPartitionCount() { return this.partitionCount; 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 9e49ca05..d5537d7a 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 @@ -42,17 +42,12 @@ public class PulsarProducerProperties extends ProducerConfigProperties { private Class messageType; /** - * Type for the Pulsar message key. + * Pulsar message key type for this binding (only used when schema type is + * {@code }KEY_VALUE}). */ @Nullable private Class messageKeyType; - /** - * Type for the Pulsar message value. - */ - @Nullable - private Class messageValueType; - /** * Number of topic partitions. */ @@ -86,15 +81,6 @@ public class PulsarProducerProperties extends ProducerConfigProperties { this.messageKeyType = messageKeyType; } - @Nullable - public Class getMessageValueType() { - return this.messageValueType; - } - - public void setMessageValueType(@Nullable Class messageValueType) { - this.messageValueType = messageValueType; - } - @Nullable public Integer getPartitionCount() { return this.partitionCount; 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 01fbcb2d..7bfbf025 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 @@ -23,6 +23,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.function.Consumer; import java.util.function.Supplier; @@ -162,9 +163,9 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { "--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.pulsar.bindings.piSupplier-out-0.producer.schema-type=FLOAT", "--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(AWAIT_DURATION)) .until(() -> output.toString().contains("Hello binder: 3.14")); @@ -172,30 +173,31 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { } @Test - void jsonTypeFoo(CapturedOutput output) { + void jsonTypeFooWithSchemaType(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.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), "--spring.cloud.function.definition=fooSupplier;fooLogger", - "--spring.cloud.stream.bindings.fooLogger-in-0.destination=fooSupplier-out-0", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=foo-stream-1", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=foo-stream-1", "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--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(), "--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())) { + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub1")) { Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) .until(() -> output.toString().contains("Hello binder: Foo[value=5150]")); } } @Test - void jsonTypeFooWithNoSchemaTypeAndCustomFooMapping(CapturedOutput output) { - SpringApplication app = new SpringApplication(JsonFooWithCustomMappingConfig.class); + void jsonTypeFooWithoutSchemaTypeDefaultsToJsonSchema(CapturedOutput output) { + SpringApplication app = new SpringApplication(JsonFooConfig.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), @@ -204,78 +206,198 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { "--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.pulsar.bindings.fooSupplier-out-0.producer.message-type=" + + Foo.class.getName(), "--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())) { + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-foo-sub2")) { Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) .until(() -> output.toString().contains("Hello binder: Foo[value=5150]")); } } @Test - void jsonTypeFooWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) { - SpringApplication app = new SpringApplication(JsonFooConfig.class); + void avroTypeUserWithSchemaType(CapturedOutput output) { + SpringApplication app = new SpringApplication(AvroUserConfig.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), - "--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(AWAIT_DURATION)).until( - () -> output.toString().contains("Could not determine producer schema for foo-stream-3")); + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-1", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-1", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=AVRO", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=AVRO", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub1")) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}")); } } @Test - void jsonTypeFooConsumerWithNoSchemaTypeAndNoCustomFooMapping(CapturedOutput output) { - SpringApplication app = new SpringApplication(JsonFooConsumerConfig.class); + void avroTypeUserWithoutSchemaTypeWithCustomMappingsViaProps(CapturedOutput output) { + SpringApplication app = new SpringApplication(AvroUserConfig.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), - "--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(AWAIT_DURATION)).until( - () -> output.toString().contains("Could not determine consumer schema for foo-stream-4")); + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-2", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-2", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub2", + "--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()), + "--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO", + "--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s" + .formatted(User.class.getName()))) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}")); } } @Test - void keyValueTypeWithCustomFooMapping(CapturedOutput output) { - SpringApplication app = new SpringApplication(KeyValueFooConfig.class); + void avroTypeUserWithoutSchemaTypeWithCustomMappingsViaCustomizer(CapturedOutput output) { + SpringApplication app = new SpringApplication(AvroUserConfigCustomMappings.class); app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run( "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), - "--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=" + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=user-stream-3", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=user-stream-3", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-user-sub3")) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: User{name='user21', age=21}")); + } + } + + @Test + void keyValueAvroTypeWithSchemaTypeAndCustomTypeMappingsViaProps(CapturedOutput output) { + SpringApplication app = new SpringApplication(KeyValueAvroUserConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-1", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-1", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type=" + String.class.getName(), - "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.message-value-type=" + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub1", + "--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()), + "--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO", + "--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s" + .formatted(User.class.getName()))) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}")); + } + } + + @Test + void keyValueAvroTypeWithoutSchemaTypeAndCustomTypeMappingsViaProps(CapturedOutput output) { + SpringApplication app = new SpringApplication(KeyValueAvroUserConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-2", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-2", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub2", + "--spring.pulsar.defaults.type-mappings[0].message-type=%s".formatted(User.class.getName()), + "--spring.pulsar.defaults.type-mappings[0].schema-info.schema-type=AVRO", + "--spring.pulsar.defaults.type-mappings[0].schema-info.message-type=%s" + .formatted(User.class.getName()))) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}")); + } + } + + @Test + void keyValueAvroTypeWithSchemaTypeAndCustomTypeMappingsViaCustomizer(CapturedOutput output) { + SpringApplication app = new SpringApplication(KeyValueAvroUserConfigCustomMappings.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), + "--spring.cloud.function.definition=userSupplier;userLogger", + "--spring.cloud.stream.bindings.userSupplier-out-0.destination=kv-stream-3", + "--spring.cloud.stream.bindings.userLogger-in-0.destination=kv-stream-3", + "--spring.cloud.stream.bindings.userSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userSupplier-out-0.producer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.bindings.userLogger-in-0.consumer.use-native-decoding=true", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.schema-type=KEY_VALUE", + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-type=" + + User.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.userLogger-in-0.consumer.subscription-name=pbit-kv-sub3")) { + Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) + .until(() -> output.toString().contains("Hello binder: 21->User{name='user21', age=21}")); + } + } + + @Test + void keyValueJsonTypeWithoutSchemaTypeAndWithoutCustomTypeMappings(CapturedOutput output) { + SpringApplication app = new SpringApplication(KeyValueJsonFooConfig.class); + app.setWebApplicationType(WebApplicationType.NONE); + try (ConfigurableApplicationContext ignored = app.run( + "--spring.pulsar.client.service-url=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), + "--spring.cloud.function.definition=fooSupplier;fooLogger", + "--spring.cloud.stream.bindings.fooSupplier-out-0.destination=kv-stream-4", + "--spring.cloud.stream.bindings.fooLogger-in-0.destination=kv-stream-4", + "--spring.cloud.stream.bindings.fooSupplier-out-0.producer.use-native-encoding=true", + "--spring.cloud.stream.pulsar.bindings.fooSupplier-out-0.producer.message-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())) { + "--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.message-key-type=" + + String.class.getName(), + "--spring.cloud.stream.pulsar.bindings.fooLogger-in-0.consumer.subscription-name=pbit-kv-sub4")) { Awaitility.await().atMost(Duration.ofSeconds(AWAIT_DURATION)) .until(() -> output.toString().contains("Hello binder: 5150->Foo[value=5150]")); } @@ -598,23 +720,6 @@ 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 { @@ -647,15 +752,70 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { @EnableAutoConfiguration @SpringBootConfiguration - static class KeyValueFooConfig { + static class AvroUserConfig { private final Logger logger = LoggerFactory.getLogger(getClass()); @Bean - public SchemaResolverCustomizer customMappings() { - return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class)); + public Supplier userSupplier() { + return () -> new User("user21", 21); } + @Bean + public Consumer userLogger() { + return f -> this.logger.info("Hello binder: " + f); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(AvroUserConfig.class) + static class AvroUserConfigCustomMappings { + + @Bean + public SchemaResolverCustomizer customMappings() { + return (resolver) -> resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + static class KeyValueAvroUserConfig { + + private final Logger logger = LoggerFactory.getLogger(getClass()); + + @Bean + public Supplier> userSupplier() { + return () -> new KeyValue<>("21", new User("user21", 21)); + } + + @Bean + public Consumer> userLogger() { + return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue()); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + @Import(KeyValueAvroUserConfig.class) + static class KeyValueAvroUserConfigCustomMappings { + + @Bean + public SchemaResolverCustomizer customMappings() { + return (resolver) -> resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + } + + } + + @EnableAutoConfiguration + @SpringBootConfiguration + static class KeyValueJsonFooConfig { + + private final Logger logger = LoggerFactory.getLogger(getClass()); + @Bean public Supplier> fooSupplier() { return () -> new KeyValue<>("5150", new Foo("5150")); @@ -671,4 +831,61 @@ class PulsarBinderIntegrationTests implements PulsarTestContainerSupport { record Foo(String value) { } + /** + * Do not convert this to a Record as Avro does not seem to work well w/ records. + */ + static class User { + + private String name; + + private int age; + + User() { + } + + User(String name, int age) { + this.name = name; + this.age = age; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public int getAge() { + return age; + } + + public void setAge(int age) { + this.age = age; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + User user = (User) o; + return age == user.age && Objects.equals(name, user.name); + } + + @Override + public int hashCode() { + return Objects.hash(name, age); + } + + @Override + public String toString() { + return "User{" + "name='" + name + '\'' + ", age=" + age + '}'; + } + + } + } 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 deleted file mode 100644 index 8b5db905..00000000 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java +++ /dev/null @@ -1,151 +0,0 @@ -/* - * 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.ArgumentMatchers.any; -import static org.mockito.ArgumentMatchers.eq; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; - -import org.apache.pulsar.client.api.Schema; -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.Resolved; -import org.springframework.pulsar.core.SchemaResolver; -import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarBinderConfigurationProperties; -import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner; -import org.springframework.pulsar.support.header.JsonPulsarHeaderMapper; - -/** - * 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), - mock(PulsarBinderConfigurationProperties.class), resolver, JsonPulsarHeaderMapper.builder().build()); - - @ParameterizedTest - @EnumSource(mode = Mode.MATCH_NONE, names = "^(AUTO.*|AVRO|JSON|KEY_VALUE|NONE|PROTOBUF.*)$") - void primitiveSchemaTypes(SchemaType schemaType) { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(schemaType, null); - binder.resolveSchema(schemaType, null, null, null); - verify(resolver).resolveSchema(schemaType, null); - } - - @ParameterizedTest - @EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$") - void structSchemaTypes(SchemaType schemaType) { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(schemaType), any()); - binder.resolveSchema(schemaType, Foo.class, null, null); - verify(resolver).resolveSchema(schemaType, ResolvableType.forClass(Foo.class)); - } - - @Test - void keyValueSchemaType() { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.KEY_VALUE), any()); - binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, Bar.class); - verify(resolver).resolveSchema(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() { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); - binder.resolveSchema(SchemaType.NONE, Foo.class, null, null); - verify(resolver).resolveSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); - } - - @Test - void withKeyAndValueTypes() { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); - binder.resolveSchema(SchemaType.NONE, null, Foo.class, Bar.class); - verify(resolver).resolveSchema(SchemaType.NONE, - ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class)); - } - - @Test - void withMessageTypeAndKeyAndValueTypes() { - doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); - binder.resolveSchema(SchemaType.NONE, Foo.class, String.class, Bar.class); - verify(resolver).resolveSchema(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/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java index f4ec6b4b..8ad74c1a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java @@ -192,7 +192,7 @@ public class DefaultSchemaResolver implements SchemaResolver { if (KeyValue.class.isAssignableFrom(messageType.getRawClass())) { yield getMessageKeyValueSchema(messageType); } - yield resolveSchema(messageType.getRawClass(), false).orElseThrow(); + yield resolveSchema(messageType.getRawClass(), true).orElseThrow(); } default -> throw new IllegalArgumentException("Unsupported schema type: " + schemaType.name()); }; diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java index 28c611e4..38d15cb3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java @@ -18,6 +18,7 @@ package org.springframework.pulsar.core; import java.util.Optional; import java.util.function.Consumer; +import java.util.function.Supplier; import org.springframework.lang.Nullable; @@ -41,15 +42,15 @@ public final class Resolved { } public static Resolved of(T value) { - return new Resolved(value, null); + return new Resolved<>(value, null); } public static Resolved failed(String reason) { - return new Resolved(null, new IllegalArgumentException(reason)); + return new Resolved<>(null, new IllegalArgumentException(reason)); } public static Resolved failed(RuntimeException e) { - return new Resolved(null, e); + return new Resolved<>(null, e); } public Optional get() { @@ -69,4 +70,11 @@ public final class Resolved { return this.value; } + public T orElseThrow(Supplier wrappingErrorMessage) { + if (this.value == null && this.exception != null) { + throw new RuntimeException(wrappingErrorMessage.get(), this.exception); + } + return this.value; + } + } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java index 26596ed8..d7121969 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java @@ -16,7 +16,10 @@ package org.springframework.pulsar.core; +import java.util.Objects; + import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.ResolvableType; @@ -69,6 +72,45 @@ public interface SchemaResolver { */ Resolved> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType); + /** + * Get the schema to use given a schema type and schema type information. + * @param the schema type + * @param schemaType schema type + * @param messageType message type (not required for primitive schema types) + * @param messageKeyType message key type (must be specified when schema type is + * {@code KEY_VALUE}) + * @return the schema to use + */ + default Resolved> resolveSchema(SchemaType schemaType, @Nullable Class messageType, + @Nullable Class messageKeyType) { + Objects.requireNonNull(schemaType, "schemaType must not be null"); + ResolvableType resolvableType = null; + if (schemaType.isStruct()) { + if (messageType == null) { + return Resolved.failed("messageType must be specified for %s schema type".formatted(schemaType.name())); + } + resolvableType = ResolvableType.forClass(messageType); + } + else if (schemaType == SchemaType.KEY_VALUE) { + if (messageType == null) { + return Resolved.failed("messageType must be specified for KEY_VALUE schema type"); + } + if (messageKeyType == null) { + return Resolved.failed("messageKeyType must be specified for KEY_VALUE schema type"); + } + resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class, messageKeyType, messageType); + } + else if (schemaType == SchemaType.NONE) { + if (messageType != null && messageKeyType != null) { + resolvableType = ResolvableType.forClassWithGenerics(KeyValue.class, messageKeyType, messageType); + } + else if (messageType != null) { + resolvableType = ResolvableType.forClass(messageType); + } + } + return resolveSchema(schemaType, resolvableType); + } + /** * Callback interface that can be implemented by beans wishing to customize the schema * resolver before it is fully initialized, in particular to tune its configuration. diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java index f05a3da8..8aa3f340 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java @@ -289,9 +289,13 @@ class DefaultSchemaResolverTests { } @Test - void customMessageType() { - assertThatExceptionOfType(IllegalArgumentException.class).isThrownBy( - () -> resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(Foo.class)).orElseThrow()); + void customMessageTypeDefaultsToJson() { + assertThat(resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(Foo.class)).orElseThrow()) + .extracting(Schema::getSchemaInfo).isEqualTo(Schema.JSON(Foo.class).getSchemaInfo()); + } + + @Test + void customMessageTypeRespectsCustomMappings() { resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); assertThat(resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(Foo.class)).orElseThrow()) .isEqualTo(Schema.STRING); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/ResolvedTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ResolvedTests.java new file mode 100644 index 00000000..dd296f84 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/ResolvedTests.java @@ -0,0 +1,61 @@ +/* + * Copyright 2023-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.core; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.assertj.core.api.Assertions.assertThatIllegalStateException; +import static org.assertj.core.api.Assertions.assertThatRuntimeException; + +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link Resolved}. + * + * @author Chris Bono + */ +class ResolvedTests { + + @Test + void success() { + assertThat(Resolved.of("good").get()).hasValue("good"); + assertThat(Resolved.of("good").orElseThrow()).isEqualTo("good"); + } + + @Test + void failedWithSimpleReason() { + var resolved = Resolved.failed("oops"); + assertThatIllegalArgumentException().isThrownBy(resolved::orElseThrow).withMessage("oops"); + assertThat(resolved.get()).isEmpty(); + } + + @Test + void failedWithReason() { + var resolved = Resolved.failed(new IllegalStateException("5150")); + assertThatIllegalStateException().isThrownBy(resolved::orElseThrow).withMessage("5150"); + assertThat(resolved.get()).isEmpty(); + } + + @Test + void failedWithAdditionalMessage() { + var resolved = Resolved.failed(new IllegalStateException("5150")); + assertThatRuntimeException().isThrownBy(() -> resolved.orElseThrow(() -> "extra message")) + .withMessage("extra message").withCause(new IllegalStateException("5150")); + assertThat(resolved.get()).isEmpty(); + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/SchemaResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SchemaResolverTests.java new file mode 100644 index 00000000..df89628e --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/SchemaResolverTests.java @@ -0,0 +1,134 @@ +/* + * Copyright 2023-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.core; + +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +import org.apache.pulsar.client.api.Schema; +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.springframework.core.ResolvableType; +import org.springframework.lang.Nullable; + +/** + * Unit tests for {@link SchemaResolver} default methods. + * + * @author Chris Bono + */ +class SchemaResolverTests { + + private TestSchemaResolver resolver = new TestSchemaResolver(); + + @Nested + class ResolveBySchemaTypeAndSchemaInfo { + + @Test + void primitiveSchemaDoesNotRequireMessageType() { + resolver.resolveSchema(SchemaType.STRING, null, null); + verify(resolver.getMock()).resolveSchema(SchemaType.STRING, null); + } + + @Test + void structSchemaWithMessageType() { + resolver.resolveSchema(SchemaType.JSON, Foo.class, null); + verify(resolver.getMock()).resolveSchema(eq(SchemaType.JSON), eq(ResolvableType.forClass(Foo.class))); + } + + @Test + void structSchemasRequireMessageType() { + assertThatIllegalArgumentException() + .isThrownBy(() -> resolver.resolveSchema(SchemaType.JSON, null, null).orElseThrow()) + .withMessage("messageType must be specified for JSON schema type"); + } + + @Test + void keyValueSchemaType() { + resolver.resolveSchema(SchemaType.KEY_VALUE, Foo.class, String.class); + verify(resolver.getMock()).resolveSchema(SchemaType.KEY_VALUE, + ResolvableType.forClassWithGenerics(KeyValue.class, String.class, Foo.class)); + } + + @Test + void keyValueSchemaRequiresMessageKeyType() { + assertThatIllegalArgumentException() + .isThrownBy(() -> resolver.resolveSchema(SchemaType.KEY_VALUE, Foo.class, null).orElseThrow()) + .withMessage("messageKeyType must be specified for KEY_VALUE schema type"); + } + + @Test + void keyValueSchemaRequiresMessageType() { + assertThatIllegalArgumentException() + .isThrownBy(() -> resolver.resolveSchema(SchemaType.KEY_VALUE, null, String.class).orElseThrow()) + .withMessage("messageType must be specified for KEY_VALUE schema type"); + } + + @Test + void schemaTypeNoneWithNullMessageType() { + resolver.resolveSchema(SchemaType.NONE, null, null); + verify(resolver.getMock()).resolveSchema(SchemaType.NONE, null); + } + + @Test + void schemaTypeNoneWithMessageType() { + resolver.resolveSchema(SchemaType.NONE, Foo.class, null); + verify(resolver.getMock()).resolveSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); + } + + @Test + void schemaTypeNoneWithMessageTypeAndKeyType() { + resolver.resolveSchema(SchemaType.NONE, Foo.class, String.class); + verify(resolver.getMock()).resolveSchema(SchemaType.NONE, + ResolvableType.forClassWithGenerics(KeyValue.class, String.class, Foo.class)); + } + + } + + private static class TestSchemaResolver implements SchemaResolver { + + private final SchemaResolver delegate = mock(SchemaResolver.class); + + SchemaResolver getMock() { + return this.delegate; + } + + @Override + public Resolved> resolveSchema(@Nullable Class messageType, boolean returnDefault) { + return delegate.resolveSchema(messageType, returnDefault); + } + + @Override + public Resolved> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType) { + return delegate.resolveSchema(schemaType, messageType); + } + + } + + record Foo(String value) { + } + + // @formatter:off + record Bar(T value) { + } + // @formatter:on + +}