From 4da2f5025c3eb6ff2f1e5739932d8a48af1a084a Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Tue, 7 Feb 2023 20:04:21 +0100 Subject: [PATCH] Use Schema.JSON instead of Schema.BYTES as default schema (#325) * Use Schema.JSON instead of Schema.BYTES as default schema * Default to BYTES schema if JSON cannot be constructed for the message type * Add doc --- .../schema-info/schema-info-listener.adoc | 2 +- .../schema-info/schema-info-template.adoc | 2 +- .../pulsar/core/DefaultSchemaResolver.java | 19 ++++- .../core/DefaultSchemaResolverTests.java | 16 +++-- .../pulsar/core/PulsarTemplateTests.java | 71 ++++++++++++++++++- 5 files changed, 97 insertions(+), 13 deletions(-) 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 ec678ec1..0e24a6b6 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 @@ -1,7 +1,7 @@ == Specifying Schema Information As indicated earlier, for Java primitives, the Spring Pulsar framework can infer the proper Schema to use on the `{listener-class}`. -However, for more complex types (such as JSON or AVRO), you need to specify the schema type on the annotation. +For non-primitive types, if the Schema is not explicitly specified on the annotation, the Spring Pulsar framework will try to build a `Schema.JSON` from the type. IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE w/ INLINE encoding. 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 273a6a94..10395369 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 @@ -1,6 +1,6 @@ == Specifying Schema Information If you use Java primitive types, the framework auto-detects the schema for you, and you need not specify any schema types for publishing the data. -However, if you use any complex types (such as `JSON`, `AVRO`, `PROTOBUF`, and others), you need to specify the proper schema when invoking send operations on the `{template-class}`. +For non-primitive types, if the Schema is not explicitly specified when invoking send operations on the `{template-class}`, the Spring Pulsar framework will try to build a `Schema.JSON` from the type. IMPORTANT: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE w/ INLINE encoding. 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 45885381..752030ba 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 @@ -39,6 +39,7 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.ResolvableType; +import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; import com.google.protobuf.GeneratedMessageV3; @@ -56,6 +57,8 @@ import com.google.protobuf.GeneratedMessageV3; */ public class DefaultSchemaResolver implements SchemaResolver { + private final LogAccessor logger = new LogAccessor(this.getClass()); + private static final Map, Schema> BASE_SCHEMA_MAPPINGS = new HashMap<>(); static { BASE_SCHEMA_MAPPINGS.put(byte[].class, Schema.BYTES); @@ -129,8 +132,20 @@ public class DefaultSchemaResolver implements SchemaResolver { } @Nullable - private Schema getCustomSchemaOrMaybeDefault(Class messageClass, boolean returnDefault) { - return this.customSchemaMappings.getOrDefault(messageClass, (returnDefault ? Schema.BYTES : null)); + protected Schema getCustomSchemaOrMaybeDefault(Class messageClass, boolean returnDefault) { + Schema schema = this.customSchemaMappings.get(messageClass); + if (schema == null && returnDefault) { + if (messageClass != null) { + try { + return Schema.JSON(messageClass); + } + catch (Exception e) { + this.logger.debug(e, "Failed to create JSON schema for " + messageClass.getName()); + } + } + return Schema.BYTES; + } + return schema; } @Override 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 6fa554a5..c616179b 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 @@ -139,7 +139,8 @@ class DefaultSchemaResolverTests { 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 + assertThat(resolver.getSchema(new Zaa("zaa1")).getSchemaInfo()) + .isEqualTo(Schema.JSON(Zaa.class).getSchemaInfo()); } } @@ -189,7 +190,8 @@ class DefaultSchemaResolverTests { @Test void customMessageTypes() { assertThat(resolver.getSchema(Foo.class, false)).isNull(); - assertThat(resolver.getSchema(Foo.class, true)).isEqualTo(Schema.BYTES); + assertThat(resolver.getSchema(Foo.class, true).getSchemaInfo()) + .isEqualTo(Schema.JSON(Foo.class).getSchemaInfo()); resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); assertThat(resolver.getSchema(Foo.class, false)).isEqualTo(Schema.STRING); assertThat(resolver.getSchema(Bar.class, false)).isNull(); @@ -305,13 +307,15 @@ class DefaultSchemaResolverTests { } @Test - void customKeyValueMessageTypeDefaultsToBytesSchema() { - ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class); + void customKeyValueMessageTypeDefaultsToJSONSchema() { + ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Zaa.class); assertThat(resolver.getSchema(SchemaType.NONE, kvType)) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)) .satisfies((keyValueSchema -> { - assertThat(keyValueSchema.getKeySchema()).isEqualTo(Schema.BYTES); - assertThat(keyValueSchema.getValueSchema()).isEqualTo(Schema.BYTES); + assertThat(keyValueSchema.getKeySchema().getSchemaInfo()) + .isEqualTo(Schema.JSON(Foo.class).getSchemaInfo()); + assertThat(keyValueSchema.getValueSchema().getSchemaInfo()) + .isEqualTo(Schema.JSON(Zaa.class).getSchemaInfo()); assertThat(keyValueSchema.getKeyValueEncodingType()).isEqualTo(KeyValueEncodingType.INLINE); })); } 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 1360fe13..d7215572 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 @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; @@ -201,13 +202,31 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { String topic = "ptt-specificSchema-topic"; try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + try (Consumer consumer = client.newConsumer(Schema.AVRO(Foo.class)).topic(topic) .subscriptionName("ptt-specificSchema-subs").subscribe()) { PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, Collections.singletonMap("topicName", topic)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); - pulsarTemplate.send(foo, Schema.JSON(Foo.class)); + pulsarTemplate.send(foo, Schema.AVRO(Foo.class)); + assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) + .isEqualTo(foo); + } + } + } + + @Test + void sendMessageWithoutSpecificSchema() throws Exception { + String topic = "ptt-nospecificSchema-topic"; + try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + .subscriptionName("ptt-nospecificSchema-subs").subscribe()) { + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + pulsarTemplate.send(foo); assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) .isEqualTo(foo); } @@ -336,7 +355,53 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { } - record Foo(String foo, String bar) { + public static class Foo { + + private String foo; + + private String bar; + + Foo() { + } + + Foo(String foo, String bar) { + this.foo = foo; + this.bar = bar; + } + + public String getFoo() { + return foo; + } + + public void setFoo(String foo) { + this.foo = foo; + } + + public String getBar() { + return bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Foo foo1 = (Foo) o; + return foo.equals(foo1.foo) && bar.equals(foo1.bar); + } + + @Override + public int hashCode() { + return Objects.hash(foo, bar); + } + } }