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

Add schema resolver customizer

Browse Source 
See #269
This commit is contained in:
Chris Bono
2023-01-20 11:02:56 -06:00
committed by Soby Chacko
parent e52ef5dea3
commit e36ef5f19c
13 changed files with 165 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc
View File

@@ -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<Class<?>, 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));
}
}
----
====

17
spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc
View File

@@ -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<Class<?>, 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));
}
}
----
====

5
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java
View File

@@ -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<Foo> 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<Foo> pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>(
producerFactory, schemaResolver);

10
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java
View File

@@ -565,11 +565,11 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
@Bean
SchemaResolver customSchemaResolver() {
Map<Class<?>, 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

11
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java
View File

@@ -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<DefaultSchemaResolverCustomizer> schemaResolverCustomizer) {
DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
schemaResolverCustomizer.ifPresent((customizer) -> customizer.customize(schemaResolver));
return schemaResolver;
}
@Bean

24
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -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<String> producerFactory = mock(PulsarProducerFactory.class);

5
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarListenerTests.java
View File

@@ -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")

5
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/ReactivePulsarListenerTests.java
View File

@@ -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",

37
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java
View File

@@ -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.
*
* <p>
* 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<Class<?>, Schema<?>> customSchemaMappings;
private final Map<Class<?>, 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<Class<?>, 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<Class<?>, Schema<?>> getCustomSchemaMappings() {
return Collections.unmodifiableMap(this.customSchemaMappings);
}
@Override

33
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolverCustomizer.java Normal file
View File

@@ -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);
}

51
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java
View File

@@ -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<Class<?>, 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<Class<?>, 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))

4
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
View File

@@ -113,8 +113,8 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
PulsarProducerFactory<Foo> 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<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(),
schemaResolver, null, null);

10
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -691,11 +691,11 @@ public class PulsarListenerTests implements PulsarTestContainerSupport {
@Bean
SchemaResolver customSchemaResolver() {
Map<Class<?>, 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