From 4eac6b50f2990b2ce446c76a052a589dc68d261c Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 28 Sep 2022 13:06:41 -0400 Subject: [PATCH] Support Pulsar Headers (#151) Convert Pulsar Message metadata as PulsarHeaders and make them available on PulsarListener through Spring's @Header annotation. See #150 --- .../config/MethodPulsarListenerEndpoint.java | 13 +- ...rBatchMessagingMessageListenerAdapter.java | 111 +++++---- ...PulsarMessagingMessageListenerAdapter.java | 158 +++++------- ...RecordMessagingMessageListenerAdapter.java | 15 +- .../DefaultPulsarMessageHeaderMapper.java | 57 +++++ .../pulsar/support/PulsarHeaders.java | 116 +++++++++ .../support/PulsarMessageHeaderMapper.java | 38 +++ .../PulsarMessagingMessageConverter.java | 29 +-- .../pulsar/listener/PulsarListenerTests.java | 233 ++++++++++++++++++ 9 files changed, 610 insertions(+), 160 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java 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 00ea537b..dc096997 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 @@ -41,6 +41,7 @@ import org.springframework.core.log.LogAccessor; import org.springframework.expression.BeanResolver; import org.springframework.lang.Nullable; import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.pulsar.core.SchemaUtils; @@ -131,11 +132,13 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo MethodParameter messageParameter = null; final Optional parameter = Arrays.stream(methodParameters) .filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class) - || !methodParameter1.getParameterType().equals(Acknowledgement.class)) + || !methodParameter1.getParameterType().equals(Acknowledgement.class) + || !methodParameter1.hasParameterAnnotation(Header.class)) .findFirst(); final long count = Arrays.stream(methodParameters) .filter(methodParameter1 -> !methodParameter1.getParameterType().equals(Consumer.class) - && !methodParameter1.getParameterType().equals(Acknowledgement.class)) + && !methodParameter1.getParameterType().equals(Acknowledgement.class) + && !methodParameter1.hasParameterAnnotation(Header.class)) .count(); Assert.isTrue(count == 1, "More than 1 expected payload types found"); if (parameter.isPresent()) { @@ -222,7 +225,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo if (rawClass != null && isContainerType(rawClass)) { resolvableType = resolvableType.getGeneric(0); } - if (Message.class.isAssignableFrom(resolvableType.getRawClass())) { + if (Message.class.isAssignableFrom(resolvableType.getRawClass()) + || org.springframework.messaging.Message.class.isAssignableFrom(resolvableType.getRawClass())) { resolvableType = resolvableType.getGeneric(0); } return resolvableType; @@ -230,7 +234,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo private boolean isContainerType(Class rawClass) { return rawClass.isAssignableFrom(List.class) || rawClass.isAssignableFrom(Message.class) - || rawClass.isAssignableFrom(Messages.class); + || rawClass.isAssignableFrom(Messages.class) + || rawClass.isAssignableFrom(org.springframework.messaging.Message.class); } protected HandlerAdapter configureListenerAdapter(PulsarMessagingMessageListenerAdapter messageListener) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java index d7322197..87b1eb35 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarBatchMessagingMessageListenerAdapter.java @@ -18,8 +18,10 @@ package org.springframework.pulsar.listener.adapter; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Messages; @@ -68,58 +70,83 @@ public class PulsarBatchMessagingMessageListenerAdapter extends PulsarMessagi @Override public void received(Consumer consumer, List> msg, @Nullable Acknowledgement acknowledgement) { - Message message; - if (!isConsumerRecordList()) { - if (isMessageList()) { - List> messages = new ArrayList<>(msg.size()); - for (org.apache.pulsar.client.api.Message record : msg) { - messages.add(toMessagingMessage(record, consumer)); + Message message = null; + Object theRecord = null; + + if (isPulsarMessageList() && !isHeaderFound()) { // List + theRecord = msg; // the incoming list as is. + } + else if (isPulsarMessageList() && isHeaderFound()) { // List, + // @Header + List> messages = toSpringMessages(consumer, msg); + final Map> aggregatedHeaders = withAggregatedHeaders(messages); + List list1 = new ArrayList<>(msg); + message = MessageBuilder.withPayload(list1).copyHeaders(aggregatedHeaders).build(); + } + else if (isMessageList() && !isHeaderFound()) { // List + List> messages = toSpringMessages(consumer, msg); + message = MessageBuilder.withPayload(messages).build(); + } + else if (isMessageList() && isHeaderFound()) { // List, @Header + List> messages = toSpringMessages(consumer, msg); + final Map> aggregatedHeaders = withAggregatedHeaders(messages); + message = MessageBuilder.withPayload(messages).copyHeaders(aggregatedHeaders).build(); + } + else if (this.isSimpleExtraction()) { // List + List list = new ArrayList<>(msg.size()); + msg.stream().iterator().forEachRemaining(vMessage -> list.add(vMessage.getValue())); + theRecord = list; + } + else if (isHeaderFound()) { // List, @Header + List> messages = toSpringMessages(consumer, msg); + final Map> aggregatedHeaders = withAggregatedHeaders(messages); + List list = new ArrayList<>(msg.size()); + msg.stream().iterator().forEachRemaining(vMessage -> list.add(vMessage.getValue())); + message = MessageBuilder.withPayload(list).copyHeaders(aggregatedHeaders).build(); + } + + if (isConsumerRecords()) { // Messages + theRecord = new Messages() { + + @Override + public Iterator> iterator() { + return msg.iterator(); } - message = MessageBuilder.withPayload(messages).build(); - } - else { - message = toMessagingMessage(msg, consumer); - } + + @Override + public int size() { + return msg.size(); + } + }; } - else { - message = MessageBuilder.withPayload(msg).build(); - } - logger.debug(() -> "Processing [" + message + "]"); - - // In order to avoid clash with target List payload type. - final Messages messages = new Messages<>() { - - @Override - public Iterator> iterator() { - return msg.iterator(); - } - - @Override - public int size() { - return msg.size(); - } - }; - invoke(messages, consumer, message, acknowledgement); + invoke(theRecord, consumer, message, acknowledgement); } - protected void invoke(Object records, Consumer consumer, final Message messageArg, - Acknowledgement acknowledgement) { + private Map> withAggregatedHeaders(List> messages) { + final Map> aggregatedHeaders = new HashMap<>(); + for (Message message : messages) { + message.getHeaders().forEach((s, o) -> { + List objects = aggregatedHeaders.computeIfAbsent(s, k -> new ArrayList<>()); + objects.add(o); + }); + } + return aggregatedHeaders; + } - Message message = messageArg; + private List> toSpringMessages(Consumer consumer, List> msg) { + List> messages = new ArrayList<>(msg.size()); + msg.stream().iterator().forEachRemaining(record -> messages.add(toMessagingMessage(record, consumer))); + return messages; + } + + protected void invoke(Object records, Consumer consumer, final Message message, + Acknowledgement acknowledgement) { try { - Object result = invokeHandler(records, message, consumer, acknowledgement); - // if (result != null) { - // handleResult(result, records, message); - // } + invokeHandler(records, message, consumer, acknowledgement); } catch (Exception e) { throw e; } } - protected Message toMessagingMessage(List> msg, Consumer consumer) { - - return getBatchMessageConverter().toMessage(msg, consumer, getType()); - } - } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java index 70bc241a..1c505a9d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarMessagingMessageListenerAdapter.java @@ -20,7 +20,6 @@ import java.lang.reflect.Method; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.lang.reflect.WildcardType; -import java.util.Collection; import java.util.List; import org.apache.commons.logging.LogFactory; @@ -37,8 +36,10 @@ import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.expression.spel.support.StandardTypeConverter; import org.springframework.messaging.converter.MessageConversionException; import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.pulsar.listener.Acknowledgement; +import org.springframework.pulsar.support.DefaultPulsarMessageHeaderMapper; import org.springframework.pulsar.support.converter.PulsarMessagingMessageConverter; import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; import org.springframework.util.Assert; @@ -64,19 +65,22 @@ public abstract class PulsarMessagingMessageListenerAdapter { private HandlerAdapter handlerMethod; - private boolean conversionNeeded = true; + private boolean headerFound = false; - private boolean messageReturnType; + private boolean simpleExtraction = false; - private boolean isConsumerRecordList; + private boolean isPulsarMessageList; - private boolean isMessageList; + private boolean isSpringMessageList; + + private boolean isSpringMessage; private boolean isConsumerRecords; private boolean converterSet; - private PulsarRecordMessageConverter messageConverter = new PulsarMessagingMessageConverter(); + private PulsarRecordMessageConverter messageConverter = new PulsarMessagingMessageConverter( + new DefaultPulsarMessageHeaderMapper()); private Type fallbackType = Object.class; @@ -112,16 +116,8 @@ public abstract class PulsarMessagingMessageListenerAdapter { this.handlerMethod = handlerMethod; } - protected boolean isConsumerRecordList() { - return this.isConsumerRecordList; - } - - public boolean isConsumerRecords() { - return this.isConsumerRecords; - } - - public boolean isConversionNeeded() { - return this.conversionNeeded; + protected boolean isPulsarMessageList() { + return this.isPulsarMessageList; } public void setBeanResolver(BeanResolver beanResolver) { @@ -131,7 +127,7 @@ public abstract class PulsarMessagingMessageListenerAdapter { } protected boolean isMessageList() { - return this.isMessageList; + return this.isSpringMessageList; } protected org.springframework.messaging.Message toMessagingMessage(Message record, Consumer consumer) { @@ -143,12 +139,6 @@ public abstract class PulsarMessagingMessageListenerAdapter { try { return this.handlerMethod.invoke(message, data, consumer, acknowledgement); - // if (data instanceof List && !this.isConsumerRecordList) { - // return this.handlerMethod.invoke(message, consumer); - // } - // else { - // return this.handlerMethod.invoke(message, data, consumer); - // } } catch (Exception ex) { throw new MessageConversionException("Cannot handle message", ex); @@ -159,10 +149,32 @@ public abstract class PulsarMessagingMessageListenerAdapter { if (method == null) { return null; } - Type genericParameterType = null; - int allowedBatchParameters = 1; - int notConvertibleParameters = 0; + + boolean pulsarMessageFound = false; + boolean collectionFound = false; + + for (int i = 0; i < method.getParameterCount(); i++) { + MethodParameter methodParameter = new MethodParameter(method, i); + Type parameterType = methodParameter.getGenericParameterType(); + + if (methodParameter.hasParameterAnnotation(Header.class)) { + this.headerFound = true; + } + else if (parameterIsType(parameterType, org.springframework.messaging.Message.class)) { + this.isSpringMessage = true; + } + else if (parameterIsType(parameterType, Message.class)) { + pulsarMessageFound = true; + } + else if (parameterIsType(parameterType, List.class) || parameterIsType(parameterType, Messages.class)) { + collectionFound = true; + } + } + + if (!this.headerFound && !this.isSpringMessage && !pulsarMessageFound && !collectionFound) { + this.simpleExtraction = true; + } for (int i = 0; i < method.getParameterCount(); i++) { MethodParameter methodParameter = new MethodParameter(method, i); @@ -173,10 +185,7 @@ public abstract class PulsarMessagingMessageListenerAdapter { */ Type parameterType = methodParameter.getGenericParameterType(); boolean isNotConvertible = parameterIsType(parameterType, Message.class); - boolean isConsumer = parameterIsType(parameterType, Consumer.class); - if (isNotConvertible) { - notConvertibleParameters++; - } + if (!isNotConvertible && !isMessageWithNoTypeInfo(parameterType) && (methodParameter.getParameterAnnotations().length == 0 || methodParameter.hasParameterAnnotation(Payload.class))) { @@ -189,37 +198,13 @@ public abstract class PulsarMessagingMessageListenerAdapter { break; } } - else { - if (isConsumer) { - allowedBatchParameters++; - } - else { - if (parameterType instanceof ParameterizedType - && ((ParameterizedType) parameterType).getRawType().equals(Consumer.class)) { - allowedBatchParameters++; - } - } - } } - - if (notConvertibleParameters == method.getParameterCount() && method.getReturnType().equals(void.class)) { - this.conversionNeeded = false; - } - boolean validParametersForBatch = method.getGenericParameterTypes().length <= allowedBatchParameters; - - if (!validParametersForBatch) { - String stateMessage = "A parameter of type '%s' must be the only parameter " - + "(except for an optional 'Acknowledgment' and/or 'Consumer' " - + "and/or '@Header(KafkaHeaders.GROUP_ID) String groupId'"; - } - this.messageReturnType = returnTypeMessageOrCollectionOf(method); return genericParameterType; } private Type extractGenericParameterTypFromMethodParameter(MethodParameter methodParameter) { Type genericParameterType = methodParameter.getGenericParameterType(); - if (genericParameterType instanceof ParameterizedType) { - ParameterizedType parameterizedType = (ParameterizedType) genericParameterType; + if (genericParameterType instanceof ParameterizedType parameterizedType) { if (parameterizedType.getRawType().equals(org.springframework.messaging.Message.class)) { genericParameterType = ((ParameterizedType) genericParameterType).getActualTypeArguments()[0]; } @@ -227,14 +212,19 @@ public abstract class PulsarMessagingMessageListenerAdapter { && parameterizedType.getActualTypeArguments().length == 1) { Type paramType = parameterizedType.getActualTypeArguments()[0]; - this.isConsumerRecordList = paramType instanceof ParameterizedType + this.isPulsarMessageList = paramType instanceof ParameterizedType && ((ParameterizedType) paramType).getRawType().equals(Message.class); boolean messageHasGeneric = paramType instanceof ParameterizedType && ((ParameterizedType) paramType) .getRawType().equals(org.springframework.messaging.Message.class); - this.isMessageList = paramType.equals(org.springframework.messaging.Message.class) || messageHasGeneric; + this.isSpringMessageList = paramType.equals(org.springframework.messaging.Message.class) + || messageHasGeneric; if (messageHasGeneric) { genericParameterType = ((ParameterizedType) paramType).getActualTypeArguments()[0]; } + + if (!this.isSpringMessageList && !this.isPulsarMessageList && !isHeaderFound()) { + this.simpleExtraction = true; + } } else { this.isConsumerRecords = parameterizedType.getRawType().equals(Messages.class); @@ -243,33 +233,8 @@ public abstract class PulsarMessagingMessageListenerAdapter { return genericParameterType; } - public static boolean returnTypeMessageOrCollectionOf(Method method) { - Type returnType = method.getGenericReturnType(); - if (returnType.equals(org.springframework.messaging.Message.class)) { - return true; - } - if (returnType instanceof ParameterizedType) { - ParameterizedType prt = (ParameterizedType) returnType; - Type rawType = prt.getRawType(); - if (rawType.equals(org.springframework.messaging.Message.class)) { - return true; - } - if (rawType.equals(Collection.class)) { - Type collectionType = prt.getActualTypeArguments()[0]; - if (collectionType.equals(org.springframework.messaging.Message.class)) { - return true; - } - return collectionType instanceof ParameterizedType && ((ParameterizedType) collectionType).getRawType() - .equals(org.springframework.messaging.Message.class); - } - } - return false; - - } - private boolean parameterIsType(Type parameterType, Type type) { - if (parameterType instanceof ParameterizedType) { - ParameterizedType parameterizedType = (ParameterizedType) parameterType; + if (parameterType instanceof ParameterizedType parameterizedType) { Type rawType = parameterizedType.getRawType(); if (rawType.equals(type)) { return true; @@ -279,20 +244,29 @@ public abstract class PulsarMessagingMessageListenerAdapter { } private boolean isMessageWithNoTypeInfo(Type parameterType) { - if (parameterType instanceof ParameterizedType) { - ParameterizedType parameterizedType = (ParameterizedType) parameterType; + if (parameterType instanceof ParameterizedType parameterizedType) { Type rawType = parameterizedType.getRawType(); if (rawType.equals(org.springframework.messaging.Message.class)) { return parameterizedType.getActualTypeArguments()[0] instanceof WildcardType; } } - return parameterType.equals(org.springframework.messaging.Message.class); // could - // be - // Message - // without - // a - // generic - // type + return parameterType.equals(org.springframework.messaging.Message.class); + } + + public boolean isSimpleExtraction() { + return this.simpleExtraction; + } + + public boolean isConsumerRecords() { + return this.isConsumerRecords; + } + + public boolean isHeaderFound() { + return this.headerFound; + } + + public boolean isSpringMessage() { + return this.isSpringMessage; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java index 40c4ea49..25c45c31 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/adapter/PulsarRecordMessagingMessageListenerAdapter.java @@ -45,22 +45,21 @@ public class PulsarRecordMessagingMessageListenerAdapter extends PulsarMessag @Override public void received(Consumer consumer, Message record, @Nullable Acknowledgement acknowledgement) { org.springframework.messaging.Message message = null; - if (isConversionNeeded()) { + Object theRecord = record; + if (isHeaderFound() || isSpringMessage()) { message = toMessagingMessage(record, consumer); } - else { - // message = NULL_MESSAGE; + else if (isSimpleExtraction()) { + theRecord = record.getValue(); } + if (logger.isDebugEnabled()) { this.logger.debug("Processing [" + message + "]"); } try { - Object result = invokeHandler(record, message, consumer, acknowledgement); - if (result != null) { - // handleResult(result, record, message); - } + invokeHandler(theRecord, message, consumer, acknowledgement); } - catch (Exception e) { // NOSONAR ex flow control + catch (Exception e) { throw e; } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java new file mode 100644 index 00000000..9dbd53ba --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/DefaultPulsarMessageHeaderMapper.java @@ -0,0 +1,57 @@ +/* + * Copyright 2022 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.support; + +import java.util.Map; + +import org.apache.pulsar.client.api.Message; + +/** + * Implementation of {@link PulsarMessageHeaderMapper}. + * + * @author Soby Chacko + */ +public class DefaultPulsarMessageHeaderMapper implements PulsarMessageHeaderMapper { + + @Override + public void toHeaders(Message source, Map target) { + target.putAll(source.getProperties()); + if (source.hasKey()) { + target.put(PulsarHeaders.KEY, source.getKey()); + target.put(PulsarHeaders.KEY_BYTES, source.getKeyBytes()); + } + if (source.hasOrderingKey()) { + target.put(PulsarHeaders.ORDERING_KEY, source.getOrderingKey()); + } + if (source.hasIndex()) { + target.put(PulsarHeaders.INDEX, source.getIndex()); + } + target.put(PulsarHeaders.MESSAGE_ID, source.getMessageId()); + target.put(PulsarHeaders.BROKER_PUBLISH_TIME, source.getBrokerPublishTime()); + target.put(PulsarHeaders.EVENT_TIME, source.getEventTime()); + target.put(PulsarHeaders.MESSAGE_SIZE, source.size()); + target.put(PulsarHeaders.PRODUCER_NAME, source.getProducerName()); + target.put(PulsarHeaders.RAW_DATA, source.getData()); + target.put(PulsarHeaders.PUBLISH_TIME, source.getPublishTime()); + target.put(PulsarHeaders.REDELIVERY_COUNT, source.getRedeliveryCount()); + target.put(PulsarHeaders.REPLICATED_FROM, source.getReplicatedFrom()); + target.put(PulsarHeaders.SCHEMA_VERSION, source.getSchemaVersion()); + target.put(PulsarHeaders.SEQUENCE_ID, source.getSequenceId()); + target.put(PulsarHeaders.TOPIC_NAME, source.getTopicName()); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java new file mode 100644 index 00000000..5b1de0c3 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarHeaders.java @@ -0,0 +1,116 @@ +/* + * Copyright 2022 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.support; + +/** + * Pulsar specific message headers. + * + * @author Soby Chacko + */ +public abstract class PulsarHeaders { + + /** + * The prefix for Pulsar headers. + */ + public static final String PREFIX = "pulsar_"; + + /** + * The prefix for the message. + */ + public static final String PULSAR_MESSAGE = PREFIX + "message_"; + + /** + * Prefix for the unique message id. + */ + public static final String MESSAGE_ID = PULSAR_MESSAGE + "id"; + + /** + * Prefix for the raw message data. + */ + public static final String RAW_DATA = PULSAR_MESSAGE + "raw_data"; + + /** + * Prefix for message size. + */ + public static final String MESSAGE_SIZE = PULSAR_MESSAGE + "size"; + + /** + * Prefix for message publish time. + */ + public static final String PUBLISH_TIME = PULSAR_MESSAGE + "publish_time"; + + /** + * Prefix for event time. + */ + public static final String EVENT_TIME = PULSAR_MESSAGE + "event_time"; + + /** + * Prefix for message sequence id. + */ + public static final String SEQUENCE_ID = PULSAR_MESSAGE + "sequence_id"; + + /** + * prefix for the producer name. + */ + public static final String PRODUCER_NAME = PULSAR_MESSAGE + "producer_name"; + + /** + * Prefix for the message key. + */ + public static final String KEY = PULSAR_MESSAGE + "key"; + + /** + * Prefix for the message key as bytes. + */ + public static final String KEY_BYTES = PULSAR_MESSAGE + "key_bytes"; + + /** + * Prefix for the order key. + */ + public static final String ORDERING_KEY = PULSAR_MESSAGE + "ordering_key"; + + /** + * Prefix for the topic name. + */ + public static final String TOPIC_NAME = PULSAR_MESSAGE + "topic_name"; + + /** + * Prefix for redelivery count. + */ + public static final String REDELIVERY_COUNT = PULSAR_MESSAGE + "redelivery_count"; + + /** + * Prefix for schema version. + */ + public static final String SCHEMA_VERSION = PULSAR_MESSAGE + "schema_version"; + + /** + * Prefix for the cluster replicated from. + */ + public static final String REPLICATED_FROM = PULSAR_MESSAGE + "replicated_from"; + + /** + * Prefix for broker publish time. + */ + public static final String BROKER_PUBLISH_TIME = PULSAR_MESSAGE + "broker_publish_time"; + + /** + * Prefix for index. + */ + public static final String INDEX = PULSAR_MESSAGE + "index"; + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java new file mode 100644 index 00000000..146bcbe7 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/PulsarMessageHeaderMapper.java @@ -0,0 +1,38 @@ +/* + * Copyright 2022 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.support; + +import java.util.Map; + +import org.apache.pulsar.client.api.Message; + +/** + * API for Pulsar message header mapper. + * + * @author Soby Chacko + */ +public interface PulsarMessageHeaderMapper { + + /** + * Map from the given message metadata to a map of headers for the eventual + * {@link org.springframework.messaging.MessageHeaders}. + * @param source Pulsar message. + * @param target the target headers. + */ + void toHeaders(Message source, Map target); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java index 629674a6..4e2386e5 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/support/converter/PulsarMessagingMessageConverter.java @@ -16,17 +16,17 @@ package org.springframework.pulsar.support.converter; -import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; -import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import org.apache.pulsar.client.api.Consumer; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.SmartMessageConverter; -import org.springframework.messaging.support.GenericMessage; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.pulsar.support.PulsarMessageHeaderMapper; /** * @@ -38,21 +38,22 @@ import org.springframework.messaging.support.MessageBuilder; */ public class PulsarMessagingMessageConverter implements PulsarRecordMessageConverter { + private final PulsarMessageHeaderMapper pulsarMessageHeaderMapper; + private SmartMessageConverter messagingConverter; + public PulsarMessagingMessageConverter(PulsarMessageHeaderMapper pulsarMessageHeaderMapper) { + this.pulsarMessageHeaderMapper = pulsarMessageHeaderMapper; + } + @Override public Message toMessage(org.apache.pulsar.client.api.Message record, Consumer consumer, Type type) { - Message message = MessageBuilder.createMessage(extractAndConvertValue(record, type), - new MessageHeaders(Collections.emptyMap())); - if (this.messagingConverter != null) { - Class clazz = type instanceof Class ? (Class) type : type instanceof ParameterizedType - ? (Class) ((ParameterizedType) type).getRawType() : Object.class; - Object payload = this.messagingConverter.fromMessage(message, clazz, type); - if (payload != null) { - message = new GenericMessage<>(payload, message.getHeaders()); - } - } + final Map messageHeaders = new HashMap<>(); + this.pulsarMessageHeaderMapper.toHeaders(record, messageHeaders); + Message message = MessageBuilder.createMessage(extractAndConvertValue(record), + new MessageHeaders(messageHeaders)); + return message; } @@ -77,7 +78,7 @@ public class PulsarMessagingMessageConverter implements PulsarRecordMessageCo this.messagingConverter = messagingConverter; } - protected Object extractAndConvertValue(org.apache.pulsar.client.api.Message record, Type type) { + protected Object extractAndConvertValue(org.apache.pulsar.client.api.Message record) { return record.getValue(); } 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 59745910..821bfd17 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,6 +19,8 @@ package org.springframework.pulsar.listener; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -27,9 +29,12 @@ import java.util.Objects; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.DeadLetterPolicy; +import org.apache.pulsar.client.api.Message; +import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Messages; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.RedeliveryBackoff; @@ -49,6 +54,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.handler.annotation.Header; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarListener; import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; @@ -64,6 +70,7 @@ import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTestContainerSupport; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -610,4 +617,230 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } + @Nested + @ContextConfiguration(classes = PulsarListenerTests.PulsarHeadersTest.PulsarListerWithHeadersConfig.class) + class PulsarHeadersTest { + + static CountDownLatch simpleListenerLatch = new CountDownLatch(1); + static CountDownLatch pulsarMessageListenerLatch = new CountDownLatch(1); + static CountDownLatch springMessagingMessageListenerLatch = new CountDownLatch(1); + + static volatile String capturedData; + static volatile MessageId messageId; + static volatile String topicName; + static volatile String fooValue; + static volatile byte[] rawData; + + static CountDownLatch simpleBatchListenerLatch = new CountDownLatch(1); + static CountDownLatch pulsarMessageBatchListenerLatch = new CountDownLatch(1); + static CountDownLatch springMessagingMessageBatchListenerLatch = new CountDownLatch(1); + static CountDownLatch pulsarMessagesBatchListenerLatch = new CountDownLatch(1); + + static volatile List capturedBatchData; + static volatile List batchMessageIds; + static volatile List batchTopicNames; + static volatile List batchFooValues; + + @Test + void simpleListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-simple-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "simpleListenerWithHeaders")) + .withTopic("simpleListenerWithHeaders").send(); + assertThat(simpleListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData).isEqualTo("hello-simple-listener"); + assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId); + assertThat(topicName).isEqualTo("persistent://public/default/simpleListenerWithHeaders"); + assertThat(fooValue).isEqualTo("simpleListenerWithHeaders"); + assertThat(rawData).isEqualTo("hello-simple-listener".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void pulsarMessageListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-message-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "pulsarMessageListenerWithHeaders")) + .withTopic("pulsarMessageListenerWithHeaders").send(); + assertThat(pulsarMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData).isEqualTo("hello-pulsar-message-listener"); + assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId); + assertThat(topicName).isEqualTo("persistent://public/default/pulsarMessageListenerWithHeaders"); + assertThat(fooValue).isEqualTo("pulsarMessageListenerWithHeaders"); + assertThat(rawData).isEqualTo("hello-pulsar-message-listener".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void springMessagingMessageListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-spring-messaging-message-listener") + .withMessageCustomizer(messageBuilder -> messageBuilder.property("foo", + "springMessagingMessageListenerWithHeaders")) + .withTopic("springMessagingMessageListenerWithHeaders").send(); + assertThat(springMessagingMessageListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedData).isEqualTo("hello-spring-messaging-message-listener"); + assertThat(PulsarHeadersTest.messageId).isEqualTo(messageId); + assertThat(topicName).isEqualTo("persistent://public/default/springMessagingMessageListenerWithHeaders"); + assertThat(fooValue).isEqualTo("springMessagingMessageListenerWithHeaders"); + assertThat(rawData).isEqualTo("hello-spring-messaging-message-listener".getBytes(StandardCharsets.UTF_8)); + } + + @Test + void simpleBatchListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-simple-batch-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "simpleBatchListenerWithHeaders")) + .withTopic("simpleBatchListenerWithHeaders").send(); + assertThat(simpleBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedBatchData).containsExactly("hello-simple-batch-listener"); + assertThat(batchMessageIds).containsExactly(messageId); + assertThat(batchTopicNames).containsExactly("persistent://public/default/simpleBatchListenerWithHeaders"); + assertThat(batchFooValues).containsExactly("simpleBatchListenerWithHeaders"); + } + + @Test + void pulsarMessageBatchListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-message-batch-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "pulsarMessageBatchListenerWithHeaders")) + .withTopic("pulsarMessageBatchListenerWithHeaders").send(); + assertThat(pulsarMessageBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedBatchData).containsExactly("hello-pulsar-message-batch-listener"); + assertThat(batchTopicNames) + .containsExactly("persistent://public/default/pulsarMessageBatchListenerWithHeaders"); + assertThat(batchFooValues).containsExactly("pulsarMessageBatchListenerWithHeaders"); + assertThat(batchMessageIds).containsExactly(messageId); + } + + @Test + void springMessagingMessageBatchListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-spring-messaging-message-batch-listener") + .withMessageCustomizer(messageBuilder -> messageBuilder.property("foo", + "springMessagingMessageBatchListenerWithHeaders")) + .withTopic("springMessagingMessageBatchListenerWithHeaders").send(); + assertThat(springMessagingMessageBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedBatchData).containsExactly("hello-spring-messaging-message-batch-listener"); + assertThat(batchTopicNames) + .containsExactly("persistent://public/default/springMessagingMessageBatchListenerWithHeaders"); + assertThat(batchFooValues).containsExactly("springMessagingMessageBatchListenerWithHeaders"); + assertThat(batchMessageIds).containsExactly(messageId); + } + + @Test + void pulsarMessagesBatchListenerWithHeaders() throws Exception { + final MessageId messageId = pulsarTemplate.newMessage("hello-pulsar-messages-batch-listener") + .withMessageCustomizer( + messageBuilder -> messageBuilder.property("foo", "pulsarMessagesBatchListenerWithHeaders")) + .withTopic("pulsarMessagesBatchListenerWithHeaders").send(); + assertThat(pulsarMessagesBatchListenerLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(capturedBatchData).containsExactly("hello-pulsar-messages-batch-listener"); + assertThat(batchTopicNames) + .containsExactly("persistent://public/default/pulsarMessagesBatchListenerWithHeaders"); + assertThat(batchFooValues).containsExactly("pulsarMessagesBatchListenerWithHeaders"); + assertThat(batchMessageIds).containsExactly(messageId); + } + + @EnablePulsar + @Configuration + static class PulsarListerWithHeadersConfig { + + @PulsarListener(subscriptionName = "simple-listener-with-headers-sub", topics = "simpleListenerWithHeaders") + void simpleListenerWithHeaders(String data, @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData, + @Header("foo") String foo) { + capturedData = data; + PulsarHeadersTest.messageId = messageId; + PulsarHeadersTest.topicName = topicName; + fooValue = foo; + PulsarHeadersTest.rawData = rawData; + simpleListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub", + topics = "pulsarMessageListenerWithHeaders") + void pulsarMessageListenerWithHeaders(Message data, + @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.TOPIC_NAME) String topicName, @Header(PulsarHeaders.RAW_DATA) byte[] rawData, + @Header("foo") String foo) { + capturedData = data.getValue(); + PulsarHeadersTest.messageId = messageId; + PulsarHeadersTest.topicName = topicName; + fooValue = foo; + PulsarHeadersTest.rawData = rawData; + pulsarMessageListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "pulsar-message-listener-with-headers-sub", + topics = "springMessagingMessageListenerWithHeaders") + void springMessagingMessageListenerWithHeaders(org.springframework.messaging.Message data, + @Header(PulsarHeaders.MESSAGE_ID) MessageId messageId, + @Header(PulsarHeaders.RAW_DATA) byte[] rawData, @Header(PulsarHeaders.TOPIC_NAME) String topicName, + @Header("foo") String foo) { + capturedData = data.getPayload(); + PulsarHeadersTest.messageId = messageId; + PulsarHeadersTest.topicName = topicName; + fooValue = foo; + PulsarHeadersTest.rawData = rawData; + springMessagingMessageListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "simple-batch-listener-with-headers-sub", + topics = "simpleBatchListenerWithHeaders", batch = true) + void simpleBatchListenerWithHeaders(List data, + @Header(PulsarHeaders.MESSAGE_ID) List messageIds, + @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { + capturedBatchData = data; + batchMessageIds = messageIds; + batchTopicNames = topicNames; + batchFooValues = fooValues; + simpleBatchListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "pulsarMessage-batch-listener-with-headers-sub", + topics = "pulsarMessageBatchListenerWithHeaders", batch = true) + void pulsarMessageBatchListenerWithHeaders(List> data, + @Header(PulsarHeaders.MESSAGE_ID) List messageIds, + @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { + + capturedBatchData = data.stream().map(Message::getValue).collect(Collectors.toList()); + + batchMessageIds = messageIds; + batchTopicNames = topicNames; + batchFooValues = fooValues; + pulsarMessageBatchListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "spring-messaging-message-batch-listener-with-headers-sub", + topics = "springMessagingMessageBatchListenerWithHeaders", batch = true) + void springMessagingMessageBatchListenerWithHeaders( + List> data, + @Header(PulsarHeaders.MESSAGE_ID) List messageIds, + @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { + + capturedBatchData = data.stream().map(org.springframework.messaging.Message::getPayload) + .collect(Collectors.toList()); + + batchMessageIds = messageIds; + batchTopicNames = topicNames; + batchFooValues = fooValues; + springMessagingMessageBatchListenerLatch.countDown(); + } + + @PulsarListener(subscriptionName = "pulsarMessages-batch-listener-with-headers-sub", + topics = "pulsarMessagesBatchListenerWithHeaders", batch = true) + void pulsarMessagesBatchListenerWithHeaders(Messages data, + @Header(PulsarHeaders.MESSAGE_ID) List messageIds, + @Header(PulsarHeaders.TOPIC_NAME) List topicNames, @Header("foo") List fooValues) { + List list = new ArrayList<>(); + data.iterator().forEachRemaining(m -> list.add(m.getValue())); + capturedBatchData = list; + + batchMessageIds = messageIds; + batchTopicNames = topicNames; + batchFooValues = fooValues; + pulsarMessagesBatchListenerLatch.countDown(); + } + + } + + } + }