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

Use Schema.JSON instead of Schema.BYTES as default schema (#325)

Browse Source 
* 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
This commit is contained in:
Christophe Bornet
2023-02-07 20:04:21 +01:00
committed by GitHub
parent c6680973a5
commit 4da2f5025c
5 changed files with 97 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

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

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

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

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

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

@@ -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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
try (Consumer<Foo> consumer = client.newConsumer(Schema.AVRO(Foo.class)).topic(topic)
.subscriptionName("ptt-specificSchema-subs").subscribe()) {
PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<Foo> 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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
.subscriptionName("ptt-nospecificSchema-subs").subscribe()) {
PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<Foo> 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);
}
}
}