Update to Pulsar 3.3.2 (#875)

This commit updates the version of Pulsar to 3.3.2.

Also, in Pulsar 3.3.2 the schema validation of an outgoing message value
happens later than it did previously. This requires the PulsarTemplate
to widen the try/catch net so that when this happens the producer is
closed properly. The backing change in Pulsar 3.3.2 can be seen here
f3c177e224 (diff-095bc2359e03726e031d8c2f210560c6a3218f48a969b0f75a2482e25cd54744R69)

gradle/libs.versions.toml

@@ -9,7 +9,7 @@ micrometer = "1.14.0-M3"
 micrometer-docs-gen = "1.0.4"
 micrometer-tracing = "1.4.0-M3"
 protobuf = "3.25.5"
-pulsar = "3.3.1"
+pulsar = "3.3.2"
 pulsar-reactive = "0.5.7"
 reactor = "2024.0.0-M6"
 spring = "6.2.0-RC1"

spring-pulsar-sample-apps/sample-failover-custom-router/compose.yaml

@@ -1,6 +1,6 @@
 services:
   pulsar:
-    image: 'apachepulsar/pulsar:3.3.1'
+    image: 'apachepulsar/pulsar:3.3.2'
     ports:
       - '6650'
       - '8080'

spring-pulsar-sample-apps/sample-imperative-produce-consume/compose.yaml

@@ -1,6 +1,6 @@
 services:
   pulsar:
-    image: 'apachepulsar/pulsar:3.3.1'
+    image: 'apachepulsar/pulsar:3.3.2'
     ports:
       - '6650'
       - '8080'

spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml

@@ -1,6 +1,6 @@
 services:
   pulsar:
-    image: 'apachepulsar/pulsar:3.3.1'
+    image: 'apachepulsar/pulsar:3.3.2'
     ports:
       - '6650'
       - '8080'

spring-pulsar-sample-apps/sample-pulsar-functions/download-connectors.sh

@@ -2,6 +2,6 @@
 
 mkdir connectors
 cd connectors
-wget https://archive.apache.org/dist/pulsar/pulsar-3.3.1/connectors/pulsar-io-cassandra-3.3.1.nar
-wget https://archive.apache.org/dist/pulsar/pulsar-3.3.1/connectors/pulsar-io-rabbitmq-3.3.1.nar
+wget https://archive.apache.org/dist/pulsar/pulsar-3.3.2/connectors/pulsar-io-cassandra-3.3.2.nar
+wget https://archive.apache.org/dist/pulsar/pulsar-3.3.2/connectors/pulsar-io-rabbitmq-3.3.2.nar
 cd ..

spring-pulsar-sample-apps/sample-pulsar-reader/compose.yaml

@@ -1,6 +1,6 @@
 services:
   pulsar:
-    image: 'apachepulsar/pulsar:3.3.1'
+    image: 'apachepulsar/pulsar:3.3.2'
     ports:
       - '6650'
       - '8080'

spring-pulsar-sample-apps/sample-reactive/compose.yaml

@@ -1,6 +1,6 @@
 services:
   pulsar:
-    image: 'apachepulsar/pulsar:3.3.1'
+    image: 'apachepulsar/pulsar:3.3.2'
     ports:
       - '6650'
       - '8080'

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java

@@ -30,7 +30,6 @@ import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.api.TypedMessageBuilder;
 import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
 import org.apache.pulsar.client.api.transaction.Transaction;
 
@@ -285,25 +284,19 @@ public class PulsarTemplate<T>
 
 		PulsarMessageSenderContext senderContext = PulsarMessageSenderContext.newContext(topicName, this.beanName);
 		Observation observation = newObservation(senderContext);
+		Producer<T> producer = null;
 		try {
 			observation.start();
-			Producer<T> producer = prepareProducerForSend(topicName, message, schema, encryptionKeys,
-					producerCustomizer);
-			TypedMessageBuilder<T> messageBuilder;
-			try {
-				var txn = getTransaction();
-				messageBuilder = (txn != null) ? producer.newMessage(txn) : producer.newMessage();
-				messageBuilder = messageBuilder.value(message);
-				if (typedMessageBuilderCustomizer != null) {
-					typedMessageBuilderCustomizer.customize(messageBuilder);
-				}
-				// propagate props to message
-				senderContext.properties().forEach(messageBuilder::property);
-			}
-			catch (RuntimeException ex) {
-				ProducerUtils.closeProducerAsync(producer, this.logger);
-				throw ex;
+			producer = prepareProducerForSend(topicName, message, schema, encryptionKeys, producerCustomizer);
+			var txn = getTransaction();
+			var messageBuilder = (txn != null) ? producer.newMessage(txn) : producer.newMessage();
+			messageBuilder = messageBuilder.value(message);
+			if (typedMessageBuilderCustomizer != null) {
+				typedMessageBuilderCustomizer.customize(messageBuilder);
 			}
+			// propagate props to message
+			senderContext.properties().forEach(messageBuilder::property);
+			var finalProducer = producer;
 			return messageBuilder.sendAsync().whenComplete((msgId, ex) -> {
 				if (ex == null) {
 					this.logger.trace(() -> "Sent msg to '%s' topic".formatted(topicName));
@@ -314,10 +307,13 @@ public class PulsarTemplate<T>
 					observation.error(ex);
 					observation.stop();
 				}
-				ProducerUtils.closeProducerAsync(producer, this.logger);
+				ProducerUtils.closeProducerAsync(finalProducer, this.logger);
 			});
 		}
 		catch (RuntimeException ex) {
+			if (producer != null) {
+				ProducerUtils.closeProducerAsync(producer, this.logger);
+			}
 			observation.error(ex);
 			observation.stop();
 			throw ex;

tools/pulsar/docker/standalone/pulsar-start.sh

@@ -3,5 +3,5 @@
 docker run -it -p 6650:6650 -p 8080:8080 \
   --mount source=pulsardata,target=/pulsar/data \
   --mount source=pulsarconf,target=/pulsar/conf \
-  apachepulsar/pulsar:3.3.1 \
+  apachepulsar/pulsar:3.3.2 \
   bin/pulsar standalone