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
apache/pulsar@f3c177e#diff-095bc2359e03726e031d8c2f210560c6a3218f48a969b0f75a2482e25cd54744R69

Author: onobc

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.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 @@ private CompletableFuture<MessageId> doSendAsync(@Nullable String topic, @Nullab
 
 		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 @@ private CompletableFuture<MessageId> doSendAsync(@Nullable String topic, @Nullab
 					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