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

Allow schema mapping via config props (#371)

Browse Source 
* 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
This commit is contained in:
Chris Bono
2023-03-14 16:16:53 -05:00
committed by GitHub
parent fd41a8f836
commit 6fedfa9c0c
21 changed files with 788 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc
View File

@@ -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.<binding-name>.producer|consumer.schema-type` for schema information and `spring.cloud.stream.pulsar.bindings.<binding-name>.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 <<pulsar.adoc#schema-info-template-imperative,custom schema mappings>> 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.

2
spring-pulsar-docs/src/main/asciidoc/pulsar.adoc
View File

@@ -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]

2
spring-pulsar-docs/src/main/asciidoc/reactive-pulsar.adoc
View File

@@ -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<String> 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]

38
spring-pulsar-docs/src/main/asciidoc/schema-info/custom-schema-mapping.adoc Normal file
View File

@@ -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<DefaultSchemaResolver> schemaResolverCustomizer() {
return (schemaResolver) -> {
schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class));
}
}
----
====

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

@@ -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<DefaultSchemaResolver> 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:

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

@@ -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<DefaultSchemaResolver> 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.

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

@@ -111,9 +111,17 @@ public class PulsarAutoConfiguration {
@Bean
@ConditionalOnMissingBean(SchemaResolver.class)
public DefaultSchemaResolver schemaResolver(
public DefaultSchemaResolver schemaResolver(PulsarProperties pulsarProperties,
Optional<SchemaResolverCustomizer<DefaultSchemaResolver>> 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;

39
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -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<TypeMapping> 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<String, Object> {

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

@@ -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<String> 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 {

67
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java
View File

@@ -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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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) {

61
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
View File

@@ -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<PulsarProducerProperties> producerProperties, MessageChannel errorChannel) {
final Schema<Object> 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<Object> 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);

18
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java
View File

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

18
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarProducerProperties.java
View File

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

355
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderIntegrationTests.java
View File

@@ -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<String> fooSupplier() {
return () -> "5150";
}
@Bean
public Consumer<Foo> 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<DefaultSchemaResolver> customMappings() {
return (resolver) -> resolver.addCustomSchemaMapping(Foo.class, JSONSchema.of(Foo.class));
public Supplier<User> userSupplier() {
return () -> new User("user21", 21);
}
@Bean
public Consumer<User> userLogger() {
return f -> this.logger.info("Hello binder: " + f);
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(AvroUserConfig.class)
static class AvroUserConfigCustomMappings {
@Bean
public SchemaResolverCustomizer<DefaultSchemaResolver> 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<KeyValue<String, User>> userSupplier() {
return () -> new KeyValue<>("21", new User("user21", 21));
}
@Bean
public Consumer<KeyValue<String, User>> userLogger() {
return f -> this.logger.info("Hello binder: " + f.getKey() + "->" + f.getValue());
}
}
@EnableAutoConfiguration
@SpringBootConfiguration
@Import(KeyValueAvroUserConfig.class)
static class KeyValueAvroUserConfigCustomMappings {
@Bean
public SchemaResolverCustomizer<DefaultSchemaResolver> 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<KeyValue<String, Foo>> 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 + '}';
}
}
}

151
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java
View File

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

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

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

14
spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java
View File

@@ -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<T> {
}
public static <T> Resolved<T> of(T value) {
return new Resolved<T>(value, null);
return new Resolved<>(value, null);
}
public static <T> Resolved<T> failed(String reason) {
return new Resolved<T>(null, new IllegalArgumentException(reason));
return new Resolved<>(null, new IllegalArgumentException(reason));
}
public static <T> Resolved<T> failed(RuntimeException e) {
return new Resolved<T>(null, e);
return new Resolved<>(null, e);
}
public Optional<T> get() {
@@ -69,4 +70,11 @@ public final class Resolved<T> {
return this.value;
}
public T orElseThrow(Supplier<String> wrappingErrorMessage) {
if (this.value == null && this.exception != null) {
throw new RuntimeException(wrappingErrorMessage.get(), this.exception);
}
return this.value;
}
}

42
spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java
View File

@@ -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 {
*/
<T> Resolved<Schema<T>> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType);
/**
* Get the schema to use given a schema type and schema type information.
* @param <T> 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 <T> Resolved<Schema<T>> 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.

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

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

61
spring-pulsar/src/test/java/org/springframework/pulsar/core/ResolvedTests.java Normal file
View File

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

134
spring-pulsar/src/test/java/org/springframework/pulsar/core/SchemaResolverTests.java Normal file
View File

@@ -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 <T> Resolved<Schema<T>> resolveSchema(@Nullable Class<?> messageType, boolean returnDefault) {
return delegate.resolveSchema(messageType, returnDefault);
}
@Override
public <T> Resolved<Schema<T>> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType) {
return delegate.resolveSchema(schemaType, messageType);
}
}
record Foo(String value) {
}
// @formatter:off
record Bar<T>(T value) {
}
// @formatter:on
}