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 322763f6..dd02d297 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 @@ -5,21 +5,21 @@ However, for more complex types (such as JSON or AVRO), you need to specify the IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE w/ INLINE encoding. -=== Custom Schema Resolver -As an alternative to specifying the schema on the `{listener-class}` for complex types, a custom schema resolver can be configured with mappings for the types. +=== Custom Schema Mapping +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 custom resolver with mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: +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 SchemaResolver customSchemaResolver() { - Map, Schema> customMappings = new HashMap<>(); - customMappings.put(User.class, Schema.AVRO(User.class)); - customMappings.put(Address.class, Schema.JSON(Address.class)); - return new DefaultSchemaResolver(customMappings); +public DefaultSchemaResolverCustomizer 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-template.adoc b/spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc index f4ce85fb..3927b959 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 @@ -11,20 +11,21 @@ template.setSchema(Schema.JSON(Foo.class)); IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE w/ INLINE encoding. -=== Custom Schema Resolver -As an alternative to specifying the schema on the `{template-class}` for complex types, a custom schema resolver can be configured with mappings for the types. +=== Custom Schema Mapping +As an alternative to specifying the schema on the `{template-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 template as the framework consults the resolver using the outgoing message type. -The following example shows a custom resolver with mappings for the `User` and `Address` complex objects using `AVRO` and `JSON` schemas, respectively: +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 SchemaResolver customSchemaResolver() { - Map, Schema> customMappings = new HashMap<>(); - customMappings.put(User.class, Schema.AVRO(User.class)); - customMappings.put(Address.class, Schema.JSON(Address.class)); - return new DefaultSchemaResolver(customMappings); +public DefaultSchemaResolverCustomizer schemaResolverCustomizer() { + return (schemaResolver) -> { + schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class)); + } } ---- ==== diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java index 45072cbd..76fa38e1 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java @@ -22,7 +22,6 @@ import static org.junit.jupiter.params.provider.Arguments.arguments; import java.time.Duration; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -95,8 +94,8 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>( client, senderSpec, null); // Custom schema resolver allows not calling setSchema on template - DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver( - Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); org.springframework.pulsar.reactive.core.ReactivePulsarTemplate pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>( producerFactory, schemaResolver); diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index f5afc7ce..38ca4073 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java @@ -565,11 +565,11 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Bean SchemaResolver customSchemaResolver() { - Map, Schema> customMappings = new HashMap<>(); - customMappings.put(User.class, Schema.AVRO(User.class)); - customMappings.put(User2.class, Schema.JSON(User2.class)); - customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); - return new DefaultSchemaResolver(customMappings); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + resolver.addCustomSchemaMapping(User2.class, Schema.JSON(User2.class)); + resolver.addCustomSchemaMapping(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); + return resolver; } @Bean 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 751e9844..ee9eb81d 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 @@ -16,6 +16,8 @@ package org.springframework.pulsar.autoconfigure; +import java.util.Optional; + import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; @@ -34,6 +36,7 @@ import org.springframework.pulsar.core.CachingPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.DefaultSchemaResolverCustomizer; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; @@ -108,9 +111,11 @@ public class PulsarAutoConfiguration { } @Bean - @ConditionalOnMissingBean - public SchemaResolver schemaResolver() { - return new DefaultSchemaResolver(); + @ConditionalOnMissingBean(SchemaResolver.class) + public DefaultSchemaResolver schemaResolver(Optional schemaResolverCustomizer) { + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + schemaResolverCustomizer.ifPresent((customizer) -> customizer.customize(schemaResolver)); + return schemaResolver; } @Bean 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 3e14ce60..a7605a39 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 @@ -21,6 +21,7 @@ import static org.mockito.Mockito.mock; import java.util.concurrent.TimeUnit; +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; @@ -48,6 +49,7 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.CachingPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.DefaultSchemaResolverCustomizer; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; @@ -105,8 +107,7 @@ class PulsarAutoConfigurationTests { .hasSingleBean(ConcurrentPulsarListenerContainerFactory.class) .hasSingleBean(PulsarListenerAnnotationBeanPostProcessor.class) .hasSingleBean(PulsarListenerEndpointRegistry.class).hasSingleBean(PulsarAdministration.class) - .hasSingleBean(SchemaResolver.class).getBean(SchemaResolver.class) - .isInstanceOf(DefaultSchemaResolver.class)); + .hasSingleBean(DefaultSchemaResolver.class)); } @Test @@ -131,6 +132,25 @@ class PulsarAutoConfigurationTests { .isSameAs(clientFactoryBean)); } + @Test + void customSchemaResolverIsRespected() { + SchemaResolver customSchemaResolver = mock(SchemaResolver.class); + this.contextRunner.withBean("customSchemaResolver", SchemaResolver.class, () -> customSchemaResolver) + .run((context) -> assertThat(context).hasNotFailed().getBean(SchemaResolver.class) + .isSameAs(customSchemaResolver)); + } + + @Test + void defaultSchemaResolverCanBeCustomized() { + record Foo() { + } + DefaultSchemaResolverCustomizer customizer = (sr) -> sr.addCustomSchemaMapping(Foo.class, Schema.STRING); + this.contextRunner.withBean("schemaResolverCustomizer", DefaultSchemaResolverCustomizer.class, () -> customizer) + .run((context) -> assertThat(context).hasNotFailed().getBean(DefaultSchemaResolver.class) + .extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP) + .containsEntry(Foo.class, Schema.STRING)); + } + @Test void customPulsarProducerFactoryIsRespected() { PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java index e5315f59..962378bb 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.autoconfigure; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -143,7 +142,9 @@ class PulsarListenerTests implements PulsarTestContainerSupport { @Bean SchemaResolver customSchemaResolver() { - return new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + resolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); + return resolver; } @PulsarListener(subscriptionName = "plt-custom-sub2", topics = "plt-custom-topic2") diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java index 2990e4dd..660fb2ed 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java @@ -18,7 +18,6 @@ package org.springframework.pulsar.autoconfigure; import static org.assertj.core.api.Assertions.assertThat; -import java.util.Collections; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -158,7 +157,9 @@ class ReactivePulsarListenerTests implements PulsarTestContainerSupport { @Bean SchemaResolver customSchemaResolver() { - return new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + resolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); + return resolver; } @ReactivePulsarListener(subscriptionName = "rplt-custom-sub2", topics = "rplt-custom-topic2", 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 77bb4464..45885381 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 @@ -26,6 +26,7 @@ import java.time.LocalTime; import java.util.Collections; import java.util.Date; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.Map; import java.util.Objects; @@ -46,8 +47,8 @@ import com.google.protobuf.GeneratedMessageV3; * Default schema resolver capable of handling basic message types. * *

- * Additional message types can be configured with the {@link #DefaultSchemaResolver(Map)} - * constructor. + * Additional message types can be configured with + * {@link #addCustomSchemaMapping(Class, Schema)}. * * @author Soby Chacko * @author Alexander Preuß @@ -85,21 +86,37 @@ public class DefaultSchemaResolver implements SchemaResolver { BASE_SCHEMA_MAPPINGS.put(LocalTime.class, Schema.LOCAL_TIME); } - private final Map, Schema> customSchemaMappings; + private final Map, Schema> customSchemaMappings = new LinkedHashMap<>(); /** - * Constructs a resolver with no custom type mappings. + * Adds a custom mapping from message type to schema. + * @param messageType the message type + * @param schema the schema to use for messages of type {@code messageType} + * @return the previously mapped schema or {@code null} if there was no mapping for + * {@code messageType}. */ - public DefaultSchemaResolver() { - this(Collections.emptyMap()); + @Nullable + public Schema addCustomSchemaMapping(Class messageType, Schema schema) { + return this.customSchemaMappings.put(messageType, schema); } /** - * Constructs a resolver with custom type mappings. - * @param customTypeSchemaMappings additional type to schema mappings to use + * Removes the custom mapping from message type to schema. + * @param messageType the message type + * @return the previously mapped schema or {@code null} if there was no mapping for + * {@code messageType}. */ - public DefaultSchemaResolver(Map, Schema> customTypeSchemaMappings) { - this.customSchemaMappings = Objects.requireNonNull(customTypeSchemaMappings); + @Nullable + public Schema removeCustomMapping(Class messageType) { + return this.customSchemaMappings.remove(messageType); + } + + /** + * Gets the currently registered custom mappings from message type to schema. + * @return unmodifiable map of custom mappings + */ + public Map, Schema> getCustomSchemaMappings() { + return Collections.unmodifiableMap(this.customSchemaMappings); } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolverCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolverCustomizer.java new file mode 100644 index 00000000..9422e746 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolverCustomizer.java @@ -0,0 +1,33 @@ +/* + * 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.core; + +/** + * The interface to customize a {@link DefaultSchemaResolver}. + * + * @author Chris Bono + */ +@FunctionalInterface +public interface DefaultSchemaResolverCustomizer { + + /** + * Customizes a {@link DefaultSchemaResolver}. + * @param schemaResolver the target of the customization + */ + void customize(DefaultSchemaResolver schemaResolver); + +} 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 faf1497e..6fa554a5 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 @@ -28,10 +28,7 @@ import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; -import java.util.Collections; import java.util.Date; -import java.util.HashMap; -import java.util.Map; import java.util.stream.Stream; import org.apache.pulsar.client.api.Schema; @@ -63,6 +60,38 @@ class DefaultSchemaResolverTests { private DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + @Nested + class CustomSchemaMappingsAPI { + + @Test + void noMappingsByDefault() { + assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP).isEmpty(); + } + + @Test + void addMappings() { + Schema previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); + assertThat(previouslyMappedSchema).isNull(); + assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP) + .containsEntry(Foo.class, Schema.STRING); + previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.BOOL); + assertThat(previouslyMappedSchema).isEqualTo(Schema.STRING); + assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP) + .containsEntry(Foo.class, Schema.BOOL); + } + + @Test + void removeMappings() { + Schema previouslyMappedSchema = resolver.removeCustomMapping(Foo.class); + assertThat(previouslyMappedSchema).isNull(); + resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); + previouslyMappedSchema = resolver.removeCustomMapping(Foo.class); + assertThat(previouslyMappedSchema).isEqualTo(Schema.STRING); + assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP).isEmpty(); + } + + } + @Nested class SchemaByMessageInstance { @@ -106,10 +135,8 @@ class DefaultSchemaResolverTests { @Test void customTypeMessages() { Schema fooSchema = Schema.AVRO(Foo.class); - Map, Schema> customTypes = new HashMap<>(); - customTypes.put(Foo.class, fooSchema); - customTypes.put(Bar.class, Schema.STRING); - DefaultSchemaResolver resolver = new DefaultSchemaResolver(customTypes); + resolver.addCustomSchemaMapping(Foo.class, fooSchema); + resolver.addCustomSchemaMapping(Bar.class, Schema.STRING); assertThat(resolver.getSchema(new Foo("foo1"))).isSameAs(fooSchema); assertThat(resolver.getSchema(new Bar<>("bar1"))).isEqualTo(Schema.STRING); assertThat(resolver.getSchema(new Zaa("zaa1"))).isEqualTo(Schema.BYTES); // default @@ -163,7 +190,7 @@ class DefaultSchemaResolverTests { void customMessageTypes() { assertThat(resolver.getSchema(Foo.class, false)).isNull(); assertThat(resolver.getSchema(Foo.class, true)).isEqualTo(Schema.BYTES); - resolver = new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.STRING)); + resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); assertThat(resolver.getSchema(Foo.class, false)).isEqualTo(Schema.STRING); assertThat(resolver.getSchema(Bar.class, false)).isNull(); assertThat(resolver.getSchema(Bar.class, true)).isEqualTo(Schema.BYTES); @@ -259,7 +286,7 @@ class DefaultSchemaResolverTests { @Test void customMessageType() { assertThat(resolver.getSchema(SchemaType.NONE, ResolvableType.forType(Foo.class))).isNull(); - resolver = new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.STRING)); + resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); assertThat(resolver.getSchema(SchemaType.NONE, ResolvableType.forType(Foo.class))) .isEqualTo(Schema.STRING); } @@ -293,10 +320,8 @@ class DefaultSchemaResolverTests { void customKeyValueMessageTypeWithCustomTypeMappings() { Schema fooSchema = mock(Schema.class); Schema barSchema = mock(Schema.class); - Map, Schema> customTypes = new HashMap<>(); - customTypes.put(Foo.class, fooSchema); - customTypes.put(Bar.class, barSchema); - resolver = new DefaultSchemaResolver(customTypes); + resolver.addCustomSchemaMapping(Foo.class, fooSchema); + resolver.addCustomSchemaMapping(Bar.class, barSchema); ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class); assertThat(resolver.getSchema(SchemaType.NONE, kvType)) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)) diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index 977d7a4d..11b697ea 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -113,8 +113,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, Collections.singletonMap("topicName", topic)); // Custom schema resolver allows not calling setSchema on template - DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver( - Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), schemaResolver, null, null); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 84406947..7360fceb 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -691,11 +691,11 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { @Bean SchemaResolver customSchemaResolver() { - Map, Schema> customMappings = new HashMap<>(); - customMappings.put(User.class, Schema.AVRO(User.class)); - customMappings.put(User2.class, Schema.JSON(User2.class)); - customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); - return new DefaultSchemaResolver(customMappings); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + resolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class)); + resolver.addCustomSchemaMapping(User2.class, Schema.JSON(User2.class)); + resolver.addCustomSchemaMapping(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); + return resolver; } @Bean