From c2732c6ee602792605d4aafafc1ddef708ea1946 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Preu=C3=9F?= Date: Fri, 2 Sep 2022 15:05:03 +0200 Subject: [PATCH] Add AVRO and (inline) KeyValue support for PulsarListener --- .../config/MethodPulsarListenerEndpoint.java | 35 ++++- .../pulsar/core/SchemaUtils.java | 7 +- .../pulsar/listener/PulsarListenerTests.java | 134 ++++++++++++++++++ 3 files changed, 171 insertions(+), 5 deletions(-) diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index b18726d3..9578ba98 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -20,13 +20,16 @@ import java.lang.reflect.Method; import java.util.Arrays; import java.util.List; import java.util.Optional; +import java.util.function.Function; import org.apache.commons.logging.LogFactory; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Messages; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.MethodParameter; @@ -37,6 +40,7 @@ import org.springframework.lang.Nullable; import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; +import org.springframework.pulsar.core.SchemaUtils; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarContainerProperties; @@ -56,6 +60,7 @@ import org.springframework.util.Assert; * * @param Message payload type * @author Soby Chacko + * @author Alexander Preuß */ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpoint { @@ -143,7 +148,15 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo case LOCAL_DATE_TIME -> pulsarContainerProperties.setSchema(Schema.LOCAL_DATE_TIME); case LOCAL_TIME -> pulsarContainerProperties.setSchema(Schema.LOCAL_TIME); case JSON -> { - final Schema requiredSchema = getRequiredSchema(methodParameters[0], pulsarContainerProperties); + final Schema requiredSchema = getRequiredSchema(methodParameters[0], JSONSchema::of); + pulsarContainerProperties.setSchema(requiredSchema); + } + case AVRO -> { + final Schema requiredSchema = getRequiredSchema(methodParameters[0], AvroSchema::of); + pulsarContainerProperties.setSchema(requiredSchema); + } + case KEY_VALUE -> { + final Schema requiredSchema = getRequiredKeyValueSchema(methodParameters[0]); pulsarContainerProperties.setSchema(requiredSchema); } } @@ -188,15 +201,29 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo return messageListener; } - private Schema getRequiredSchema(MethodParameter methodParameter, - PulsarContainerProperties pulsarContainerProperties) { + private Schema getRequiredSchema(MethodParameter methodParameter, Function, Schema> schemaFactory) { ResolvableType resolvableType = ResolvableType.forMethodParameter(methodParameter); final Class rawClass = resolvableType.getRawClass(); if (rawClass != null && isContainerType(rawClass)) { resolvableType = resolvableType.getGeneric(0); } final Class rawClazz = resolvableType.getRawClass(); - return JSONSchema.of(rawClazz); + + return schemaFactory.apply(rawClazz); + } + + private Schema getRequiredKeyValueSchema(MethodParameter methodParameter) { + ResolvableType resolvableType = ResolvableType.forMethodParameter(methodParameter); + final Class rawClass = resolvableType.getRawClass(); + if (rawClass != null && isContainerType(rawClass)) { + resolvableType = resolvableType.getGeneric(0); + } + Class generic1 = resolvableType.resolveGeneric(0); + Class generic2 = resolvableType.resolveGeneric(1); + Schema> schema1 = SchemaUtils.getSchema(generic1.getName()); + Schema> schema2 = SchemaUtils.getSchema(generic2.getName()); + + return Schema.KeyValue(schema1, schema2, KeyValueEncodingType.INLINE); } private boolean isContainerType(Class rawClass) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java index e3540b82..1df37870 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java @@ -22,6 +22,7 @@ import org.apache.pulsar.client.api.Schema; * Utility class for Pulsar schema inference. * * @author Soby Chacko + * @author Alexander Preuß */ public final class SchemaUtils { @@ -29,9 +30,13 @@ public final class SchemaUtils { } - @SuppressWarnings("unchecked") public static Schema getSchema(T message) { final String clazzName = message.getClass().getName(); + return getSchema(clazzName); + } + + @SuppressWarnings("unchecked") + public static Schema getSchema(String clazzName) { return switch (clazzName) { case "java.lang.String" -> (Schema) Schema.STRING; case "[B" -> (Schema) Schema.BYTES; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 704a75be..5e8534ed 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -19,14 +19,22 @@ package org.springframework.pulsar.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.Objects; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.impl.schema.AvroSchema; +import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.common.schema.KeyValue; +import org.apache.pulsar.common.schema.KeyValueEncodingType; +import org.apache.pulsar.common.schema.SchemaType; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -241,4 +249,130 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { } + @Nested + @ContextConfiguration(classes = SchemaTestCases.SchemaTestConfig.class) + class SchemaTestCases { + + static CountDownLatch jsonLatch = new CountDownLatch(1); + + static CountDownLatch avroLatch = new CountDownLatch(1); + + static CountDownLatch keyvalueLatch = new CountDownLatch(1); + + + + @Test + void jsonSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(JSONSchema.of(User.class)); + template.send("json-topic", new User("Jason", 1)); + assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void avroSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(AvroSchema.of(User.class)); + template.send("avro-topic", new User("Avi", 2)); + assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void keyvalueSchema() throws Exception { + PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, Collections.emptyMap()); + PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); + + Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, + KeyValueEncodingType.INLINE); + + template.setSchema(kvSchema); + template.send("keyvalue-topic", new KeyValue<>("Kevin", 3)); + assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class SchemaTestConfig { + + @PulsarListener(id = "jsonListener", topics = "json-topic", subscriptionName = "subscription-4", + schemaType = SchemaType.JSON, properties = { "subscriptionInitialPosition=Earliest" }) + void listenJson(User message) { + jsonLatch.countDown(); + } + + @PulsarListener(id = "avroListener", topics = "avro-topic", subscriptionName = "subscription-5", + schemaType = SchemaType.AVRO, properties = { "subscriptionInitialPosition=Earliest" }) + void listenAvro(User message) { + avroLatch.countDown(); + } + + @PulsarListener(id = "keyvalueListener", topics = "keyvalue-topic", subscriptionName = "subscription-6", + schemaType = SchemaType.KEY_VALUE, properties = { "subscriptionInitialPosition=Earliest" }) + void listenKeyvalue(KeyValue message) { + keyvalueLatch.countDown(); + } + } + + 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 + '}'; + } + + } + + } + }