Release 3.2.0 (#279) (#280)

* DATAGO-67277: Feature to exclude header on consumer binding

Co-authored-by: mayur-solace <[email protected]>

pom.xml

@@ -10,7 +10,7 @@
 
     <groupId>com.solace.spring.cloud</groupId>
     <artifactId>solace-spring-cloud-build</artifactId>
-    <version>3.1.1-SNAPSHOT</version>
+    <version>3.2.0-SNAPSHOT</version>
     <packaging>pom</packaging>
 
     <name>Solace Spring Cloud Build</name>
@@ -32,7 +32,7 @@
         <!-- Remove this if the next version of solace-spring-boot works fine -->
         <spring.boot.version>3.1.5</spring.boot.version>
 
-        <solace.spring.cloud.stream-starter.version>4.1.1-SNAPSHOT</solace.spring.cloud.stream-starter.version>
+        <solace.spring.cloud.stream-starter.version>4.2.0-SNAPSHOT</solace.spring.cloud.stream-starter.version>
 
         <solace.integration.test.support.version>1.0.2</solace.integration.test.support.version>
         <solace.integration.test.support.fetch_checkout.skip>false</solace.integration.test.support.fetch_checkout.skip>

solace-spring-cloud-bom/README.md

@@ -26,6 +26,7 @@ Consult the table below to determine which version of the BOM you need to use:
 | 2021.0.6     | 2.5.0                      | 2.7.x       |
 | 2022.0.2     | 3.0.0                      | 3.0.x       |
 | 2022.0.4     | 3.1.0                      | 3.1.x       |
+| 2022.0.4     | 3.2.0                      | 3.1.x       |
 
 ## Including the BOM
 
@@ -38,7 +39,7 @@ In addition to showing how to include the BOM, the following snippets also shows
         <dependency>
             <groupId>com.solace.spring.cloud</groupId>
             <artifactId>solace-spring-cloud-bom</artifactId>
-            <version>3.1.0</version>
+            <version>3.2.0</version>
             <type>pom</type>
             <scope>import</scope>
         </dependency>
@@ -66,7 +67,7 @@ apply plugin: 'io.spring.dependency-management'
 
 dependencyManagement {
     imports {
-        mavenBom "com.solace.spring.cloud:solace-spring-cloud-bom:3.1.0"
+        mavenBom "com.solace.spring.cloud:solace-spring-cloud-bom:3.2.0"
     }
 }
 
@@ -78,7 +79,7 @@ dependencies {
 ### Using it with Gradle 5
 ```groovy
 dependencies {
-    implementation(platform("com.solace.spring.cloud:solace-spring-cloud-bom:3.1.0"))
+    implementation(platform("com.solace.spring.cloud:solace-spring-cloud-bom:3.2.0"))
     implementation("com.solace.spring.cloud:spring-cloud-starter-stream-solace")
 }
 ```

solace-spring-cloud-bom/pom.xml

@@ -5,13 +5,13 @@
     <parent>
         <groupId>com.solace.spring.cloud</groupId>
         <artifactId>solace-spring-cloud-build</artifactId>
-        <version>3.1.1-SNAPSHOT</version>
+        <version>3.2.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 
     <artifactId>solace-spring-cloud-bom</artifactId>
     <packaging>pom</packaging>
-    <version>3.1.1-SNAPSHOT</version>
+    <version>3.2.0-SNAPSHOT</version>
 
     <name>Solace Spring Cloud BOM</name>
     <description>BOM for Solace Spring Cloud</description>

solace-spring-cloud-parent/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>com.solace.spring.cloud</groupId>
         <artifactId>solace-spring-cloud-build</artifactId>
-        <version>3.1.1-SNAPSHOT</version>
+        <version>3.2.0-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

solace-spring-cloud-starters/solace-spring-cloud-stream-starter/README.adoc

@@ -1,5 +1,5 @@
 = Spring Cloud Stream Binder for Solace PubSub+
-:revnumber: 4.1.0
+:revnumber: 4.2.0
 :toc: preamble
 :toclevels: 3
 :icons: font
@@ -346,6 +346,11 @@ The number of milliseconds before republished messages are discarded or moved to
 +
 Default: `null`
 
+headerExclusions::
+The list of headers to exclude when converting consumed Solace message to Spring message.
++
+Default: Empty `List<String>`
+
 ==== Solace Producer Properties
 
 The following properties are available for Solace producers only and must be prefixed with `spring.cloud.stream.solace.bindings.<bindingName>.producer.` where `bindingName` looks something like `functionName-out-0` as defined in https://docs.spring.io/spring-cloud-stream/docs/{scst-version}/reference/html/spring-cloud-stream.html#_functional_binding_names[Functional Binding Names].

solace-spring-cloud-starters/solace-spring-cloud-stream-starter/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <groupId>com.solace.spring.cloud</groupId>
         <artifactId>solace-spring-cloud-parent</artifactId>
-        <version>3.1.1-SNAPSHOT</version>
+        <version>3.2.0-SNAPSHOT</version>
         <relativePath>../../solace-spring-cloud-parent/pom.xml</relativePath>
     </parent>
 
     <artifactId>spring-cloud-starter-stream-solace</artifactId>
-    <version>4.1.1-SNAPSHOT</version>
+    <version>4.2.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
 

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <groupId>com.solace.spring.cloud</groupId>
         <artifactId>solace-spring-cloud-parent</artifactId>
-        <version>3.1.1-SNAPSHOT</version>
+        <version>3.2.0-SNAPSHOT</version>
         <relativePath>../../solace-spring-cloud-parent/pom.xml</relativePath>
     </parent>
 
     <artifactId>spring-cloud-stream-binder-solace-core</artifactId>
-    <version>4.1.1-SNAPSHOT</version>
+    <version>4.2.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>Solace Spring Cloud Stream Binder Core</name>

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/inbound/InboundXMLMessageListener.java

@@ -241,13 +241,13 @@ private void processBatchIfAvailable() {
 
 	Message<?> createOneMessage(BytesXMLMessage bytesXMLMessage, AcknowledgmentCallback acknowledgmentCallback) {
 		setAttributesIfNecessary(bytesXMLMessage, acknowledgmentCallback);
-		return xmlMessageMapper.map(bytesXMLMessage, acknowledgmentCallback);
+		return xmlMessageMapper.map(bytesXMLMessage, acknowledgmentCallback, consumerProperties.getExtension());
 	}
 
 	Message<?> createBatchMessage(List<BytesXMLMessage> bytesXMLMessages,
 								  AcknowledgmentCallback acknowledgmentCallback) {
 		setAttributesIfNecessary(bytesXMLMessages, acknowledgmentCallback);
-		return xmlMessageMapper.mapBatchMessage(bytesXMLMessages, acknowledgmentCallback);
+		return xmlMessageMapper.mapBatchMessage(bytesXMLMessages, acknowledgmentCallback, consumerProperties.getExtension());
 	}
 
 	void sendOneToConsumer(final Message<?> message, final BytesXMLMessage bytesXMLMessage)

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/inbound/JCSMPMessageSource.java

@@ -167,7 +167,7 @@ protected Object doReceive() {
 	private Message<?> processMessage(MessageContainer messageContainer) {
 		AcknowledgmentCallback acknowledgmentCallback = ackCallbackFactory.createCallback(messageContainer);
 		try {
-			return xmlMessageMapper.map(messageContainer.getMessage(), acknowledgmentCallback, true);
+			return xmlMessageMapper.map(messageContainer.getMessage(), acknowledgmentCallback, true, consumerProperties.getExtension());
 		} catch (Exception e) {
 			//TODO If one day the errorChannel or attributesHolder can be retrieved, use those instead
 			logger.warn(e, String.format("XMLMessage %s cannot be consumed. It will be rejected",
@@ -188,7 +188,7 @@ private Message<List<?>> processBatchIfAvailable() {
 			return xmlMessageMapper.mapBatchMessage(batchedMessages.get()
 					.stream()
 					.map(MessageContainer::getMessage)
-					.collect(Collectors.toList()), acknowledgmentCallback, true);
+					.collect(Collectors.toList()), acknowledgmentCallback, true, consumerProperties.getExtension());
 		} catch (Exception e) {
 			logger.warn(e, "Message batch cannot be consumed. It will be rejected");
 			AckUtils.reject(acknowledgmentCallback);

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/properties/SolaceConsumerProperties.java

@@ -1,6 +1,8 @@
 package com.solace.spring.cloud.stream.binder.properties;
 
 import com.solacesystems.jcsmp.EndpointProperties;
+import java.util.ArrayList;
+import java.util.List;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.util.Assert;
 
@@ -133,6 +135,10 @@ public class SolaceConsumerProperties extends SolaceCommonProperties {
 	private Long errorMsgTtl = null;
 	// ------------------------
 
+	/**
+	 * The list of headers to exclude when converting consumed Solace message to Spring message.
+	 */
+	private List<String> headerExclusions = new ArrayList<>();
 
 	public int getBatchMaxSize() {
 		return batchMaxSize;
@@ -317,4 +323,12 @@ public Long getErrorMsgTtl() {
 	public void setErrorMsgTtl(Long errorMsgTtl) {
 		this.errorMsgTtl = errorMsgTtl;
 	}
+
+	public List<String> getHeaderExclusions() {
+		return headerExclusions;
+	}
+
+	public void setHeaderExclusions(List<String> headerExclusions) {
+		this.headerExclusions = headerExclusions;
+	}
 }

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/XMLMessageMapper.java

@@ -164,18 +164,18 @@ public XMLMessage map(Message<?> message, Collection<String> excludedHeaders, bo
 	}
 
 	public Message<List<?>> mapBatchMessage(List<? extends XMLMessage> xmlMessages,
-											AcknowledgmentCallback acknowledgmentCallback)
+											AcknowledgmentCallback acknowledgmentCallback, SolaceConsumerProperties solaceConsumerProperties)
 			throws SolaceMessageConversionException {
-		return mapBatchMessage(xmlMessages, acknowledgmentCallback, false);
+		return mapBatchMessage(xmlMessages, acknowledgmentCallback, false, solaceConsumerProperties);
 	}
 
 	public Message<List<?>> mapBatchMessage(List<? extends XMLMessage> xmlMessages,
 											AcknowledgmentCallback acknowledgmentCallback,
-											boolean setRawMessageHeader) throws SolaceMessageConversionException {
+											boolean setRawMessageHeader, SolaceConsumerProperties solaceConsumerProperties) throws SolaceMessageConversionException {
 		List<Map<String, Object>> batchedHeaders = new ArrayList<>();
 		List<Object> batchedPayloads = new ArrayList<>();
 		for (XMLMessage xmlMessage : xmlMessages) {
-			Message<?> message = mapInternal(xmlMessage).build();
+			Message<?> message = mapInternal(xmlMessage, solaceConsumerProperties).build();
 			batchedHeaders.add(message.getHeaders());
 			batchedPayloads.add(message.getPayload());
 		}
@@ -186,20 +186,21 @@ public Message<List<?>> mapBatchMessage(List<? extends XMLMessage> xmlMessages,
 				.build();
 	}
 
-	public Message<?> map(XMLMessage xmlMessage, AcknowledgmentCallback acknowledgmentCallback)
+	public Message<?> map(XMLMessage xmlMessage, AcknowledgmentCallback acknowledgmentCallback, SolaceConsumerProperties solaceConsumerProperties)
 			throws SolaceMessageConversionException {
-		return map(xmlMessage, acknowledgmentCallback, false);
+		return map(xmlMessage, acknowledgmentCallback, false, solaceConsumerProperties);
 	}
 
 	public Message<?> map(XMLMessage xmlMessage, AcknowledgmentCallback acknowledgmentCallback,
-						  boolean setRawMessageHeader) throws SolaceMessageConversionException {
-		return injectRootMessageHeaders(mapInternal(xmlMessage), acknowledgmentCallback, setRawMessageHeader ?
+						  boolean setRawMessageHeader, SolaceConsumerProperties solaceConsumerProperties) throws SolaceMessageConversionException {
+		return injectRootMessageHeaders(mapInternal(xmlMessage, solaceConsumerProperties), acknowledgmentCallback, setRawMessageHeader ?
 				xmlMessage : null).build();
 	}
 
-	private AbstractIntegrationMessageBuilder<?> mapInternal(XMLMessage xmlMessage)
+	private AbstractIntegrationMessageBuilder<?> mapInternal(XMLMessage xmlMessage, SolaceConsumerProperties solaceConsumerProperties)
 			throws SolaceMessageConversionException {
 		SDTMap metadata = xmlMessage.getProperties();
+		List<String> excludedHeaders = solaceConsumerProperties.getHeaderExclusions();
 
 		Object payload;
 		if (xmlMessage instanceof BytesMessage) {
@@ -247,7 +248,7 @@ private AbstractIntegrationMessageBuilder<?> mapInternal(XMLMessage xmlMessage)
 
 		AbstractIntegrationMessageBuilder<?> builder = MESSAGE_BUILDER_FACTORY
 				.withPayload(payload)
-				.copyHeaders(map(metadata))
+				.copyHeaders(map(metadata, excludedHeaders))
 				.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, xmlMessage.getHTTPContentType());
 
 		if (isNullPayload) {
@@ -261,6 +262,9 @@ private AbstractIntegrationMessageBuilder<?> mapInternal(XMLMessage xmlMessage)
 			if (!header.getValue().isReadable()) {
 				continue;
 			}
+			if (excludedHeaders != null && excludedHeaders.contains(header.getKey())) {
+				continue;
+			}
 			if (ignoredHeaderProperties.contains(header.getKey())) {
 				continue;
 			}
@@ -320,15 +324,19 @@ SDTMap map(MessageHeaders headers, Collection<String> excludedHeaders, boolean c
 		return metadata;
 	}
 
-	MessageHeaders map(SDTMap metadata) {
+	MessageHeaders map(SDTMap metadata, Collection<String> excludedHeaders) {
 		if (metadata == null) {
 			return new MessageHeaders(Collections.emptyMap());
 		}
 
+		final Collection<String> exclusionList =
+				excludedHeaders != null ? excludedHeaders : Collections.emptyList();
+
 		Map<String,Object> headers = new HashMap<>();
 
 		// Deserialize headers
-		if (metadata.containsKey(SolaceBinderHeaders.SERIALIZED_HEADERS)) {
+		if (!exclusionList.contains(SolaceBinderHeaders.SERIALIZED_HEADERS) &&
+				metadata.containsKey(SolaceBinderHeaders.SERIALIZED_HEADERS)) {
 			Encoder encoder = null;
 			if (metadata.containsKey(SolaceBinderHeaders.SERIALIZED_HEADERS_ENCODING)) {
 				String encoding = rethrowableCall(metadata::getString, SolaceBinderHeaders.SERIALIZED_HEADERS_ENCODING);
@@ -345,7 +353,9 @@ MessageHeaders map(SDTMap metadata) {
 					rethrowableCall(metadata::getString, SolaceBinderHeaders.SERIALIZED_HEADERS));
 
 			for (String headerName : serializedHeaders) {
-				if (metadata.containsKey(headerName)) {
+				if (exclusionList.contains(headerName)) {
+					continue;
+				} else if (metadata.containsKey(headerName)) {
 					byte[] serializedValue = encoder != null ?
 							encoder.decode(rethrowableCall(metadata::getString, headerName)) :
 							rethrowableCall(metadata::getBytes, headerName);
@@ -359,6 +369,7 @@ MessageHeaders map(SDTMap metadata) {
 		}
 
 		metadata.keySet().stream()
+				.filter(h -> !exclusionList.contains(h))
 				.filter(h -> !headers.containsKey(h))
 				.filter(h -> !SolaceBinderHeaderMeta.META.containsKey(h))
 				.filter(h -> !SolaceHeaderMeta.META.containsKey(h))
@@ -370,7 +381,8 @@ MessageHeaders map(SDTMap metadata) {
 					headers.put(h, value);
 				});
 
-		if (metadata.containsKey(SolaceBinderHeaders.MESSAGE_VERSION)) {
+		if (!exclusionList.contains(SolaceBinderHeaders.MESSAGE_VERSION) &&
+				metadata.containsKey(SolaceBinderHeaders.MESSAGE_VERSION)) {
 			int messageVersion = rethrowableCall(metadata::getInteger, SolaceBinderHeaders.MESSAGE_VERSION);
 			headers.put(SolaceBinderHeaders.MESSAGE_VERSION, messageVersion);
 		}

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/test/java/com/solace/spring/cloud/stream/binder/properties/SolaceConsumerPropertiesTest.java

@@ -1,10 +1,12 @@
 package com.solace.spring.cloud.stream.binder.properties;
 
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class SolaceConsumerPropertiesTest {
 	@ParameterizedTest
@@ -81,4 +83,9 @@ public void testFailSetFlowRebindBackOffMultiplier(double multiplier) {
 		assertThrows(IllegalArgumentException.class, () -> new SolaceConsumerProperties()
 				.setFlowRebindBackOffMultiplier(multiplier));
 	}
+
+	@Test
+	void testDefaultHeaderExclusionsListIsEmpty() {
+		assertTrue(new SolaceConsumerProperties().getHeaderExclusions().isEmpty());
+	}
 }

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/test/java/com/solace/spring/cloud/stream/binder/util/JmsCompatibilityIT.java

@@ -6,6 +6,7 @@
 import com.solace.spring.cloud.stream.binder.messaging.HeaderMeta;
 import com.solace.spring.cloud.stream.binder.messaging.SolaceBinderHeaderMeta;
 import com.solace.spring.cloud.stream.binder.messaging.SolaceBinderHeaders;
+import com.solace.spring.cloud.stream.binder.properties.SolaceConsumerProperties;
 import com.solace.spring.cloud.stream.binder.test.util.SerializableFoo;
 import com.solace.test.integration.junit.jupiter.extension.PubSubPlusExtension;
 import com.solacesystems.jcsmp.BytesMessage;
@@ -339,6 +340,7 @@ public void testPayloadFromJmsToSpring(JCSMPSession jcsmpSession, SoftAssertions
 			messages.add(mapMessage);
 		}
 
+		SolaceConsumerProperties consumerProperties = new SolaceConsumerProperties();
 		XMLMessageConsumer messageConsumer = null;
 		try {
 			Set<Class<? extends XMLMessage>> processedMessageTypes = new HashSet<>();
@@ -349,7 +351,7 @@ public void testPayloadFromJmsToSpring(JCSMPSession jcsmpSession, SoftAssertions
 				public void onReceive(BytesXMLMessage bytesXMLMessage) {
 					logger.info("Got message " + bytesXMLMessage);
 					try {
-						Message<?> msg = xmlMessageMapper.map(bytesXMLMessage, null);
+						Message<?> msg = xmlMessageMapper.map(bytesXMLMessage, null, consumerProperties);
 						if (msg.getPayload() instanceof byte[]) {
 							softly.assertThat(msg.getPayload()).isEqualTo("test".getBytes());
 							processedMessageTypes.add(BytesMessage.class);
@@ -428,6 +430,7 @@ public void testSerializedPayloadFromJmsToSpring(JCSMPSession jcsmpSession, Soft
 		ObjectMessage message = jmsSession.createObjectMessage(payload);
 		message.setBooleanProperty(SolaceBinderHeaders.SERIALIZED_PAYLOAD, true);
 
+		SolaceConsumerProperties consumerProperties = new SolaceConsumerProperties();
 		XMLMessageConsumer messageConsumer = null;
 		try {
 			AtomicReference<Exception> exceptionAtomicReference = new AtomicReference<>();
@@ -437,7 +440,7 @@ public void testSerializedPayloadFromJmsToSpring(JCSMPSession jcsmpSession, Soft
 				public void onReceive(BytesXMLMessage bytesXMLMessage) {
 					logger.info("Got message " + bytesXMLMessage);
 					try {
-						softly.assertThat(xmlMessageMapper.map(bytesXMLMessage, null).getPayload())
+						softly.assertThat(xmlMessageMapper.map(bytesXMLMessage, null, consumerProperties).getPayload())
 								.isEqualTo(payload);
 					} catch (Exception e) {
 						exceptionAtomicReference.set(e);