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

Add AVRO and (inline) KeyValue support for PulsarListener

Browse Source
This commit is contained in:
Alexander Preuß
2022-09-02 15:05:03 +02:00
committed by Chris Bono
parent b188436ab5
commit c2732c6ee6
3 changed files with 171 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java
View File

@@ -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 <V> Message payload type
* @author Soby Chacko
* @author Alexander Preuß
*/
public class MethodPulsarListenerEndpoint<V> extends AbstractPulsarListenerEndpoint<V> {
@@ -143,7 +148,15 @@ public class MethodPulsarListenerEndpoint<V> 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<V> extends AbstractPulsarListenerEndpo
return messageListener;
}
private Schema<?> getRequiredSchema(MethodParameter methodParameter,
PulsarContainerProperties pulsarContainerProperties) {
private Schema<?> getRequiredSchema(MethodParameter methodParameter, Function<Class<?>, 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<? extends Class<?>> schema1 = SchemaUtils.getSchema(generic1.getName());
Schema<? extends Class<?>> schema2 = SchemaUtils.getSchema(generic2.getName());
return Schema.KeyValue(schema1, schema2, KeyValueEncodingType.INLINE);
}
private boolean isContainerType(Class<?> rawClass) {

7
spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java
View File

@@ -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 <T> Schema<T> getSchema(T message) {
final String clazzName = message.getClass().getName();
return getSchema(clazzName);
}
@SuppressWarnings("unchecked")
public static <T> Schema<T> getSchema(String clazzName) {
return switch (clazzName) {
case "java.lang.String" -> (Schema<T>) Schema.STRING;
case "[B" -> (Schema<T>) Schema.BYTES;

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

@@ -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<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap());
PulsarTemplate<User> 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<User> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap());
PulsarTemplate<User> 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<KeyValue<String, Integer>> pulsarProducerFactory = new DefaultPulsarProducerFactory<>(
pulsarClient, Collections.emptyMap());
PulsarTemplate<KeyValue<String, Integer>> template = new PulsarTemplate<>(pulsarProducerFactory);
Schema<KeyValue<String, Integer>> 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<String, Integer> 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 + '}';
}
}
}
}