diff --git a/sdk/spring/spring-messaging-azure-eventhubs/src/main/java/com/azure/spring/messaging/eventhubs/core/EventHubsTemplate.java b/sdk/spring/spring-messaging-azure-eventhubs/src/main/java/com/azure/spring/messaging/eventhubs/core/EventHubsTemplate.java
index 93c63b4f4b13..0b9343efd0e9 100644
--- a/sdk/spring/spring-messaging-azure-eventhubs/src/main/java/com/azure/spring/messaging/eventhubs/core/EventHubsTemplate.java
+++ b/sdk/spring/spring-messaging-azure-eventhubs/src/main/java/com/azure/spring/messaging/eventhubs/core/EventHubsTemplate.java
@@ -98,48 +98,48 @@ public <T> Mono<Void> sendAsync(String destination, Message<T> message) {
     }
 
     private Mono<Void> doSend(String destination, List<EventData> events, PartitionSupplier partitionSupplier) {
-        try (EventHubProducerAsyncClient producer = producerFactory.createProducer(destination)) {
-            CreateBatchOptions options = buildCreateBatchOptions(partitionSupplier);
-            AtomicReference<EventDataBatch> currentBatch = new AtomicReference<>(
-                producer.createBatch(options).block());
-            Flux.fromIterable(events).flatMap(event -> {
-                final EventDataBatch batch = currentBatch.get();
-                try {
-                    if (batch.tryAdd(event)) {
-                        return Mono.empty();
-                    } else {
-                        LOGGER.warn("EventDataBatch is full in the collect process or the first event is "
-                            + "too large to fit in an empty batch! Max size: {}", batch.getMaxSizeInBytes());
-                    }
-                } catch (AmqpException e) {
-                    LOGGER.error("Event is larger than maximum allowed size.", e);
+        EventHubProducerAsyncClient producer = producerFactory.createProducer(destination);
+        CreateBatchOptions options = buildCreateBatchOptions(partitionSupplier);
+        AtomicReference<EventDataBatch> currentBatch = new AtomicReference<>(
+            producer.createBatch(options).block());
+        Flux.fromIterable(events).flatMap(event -> {
+            final EventDataBatch batch = currentBatch.get();
+            try {
+                if (batch.tryAdd(event)) {
                     return Mono.empty();
+                } else {
+                    LOGGER.warn("EventDataBatch is full in the collect process or the first event is "
+                        + "too large to fit in an empty batch! Max size: {}", batch.getMaxSizeInBytes());
                 }
-
-                return Mono.when(
-                    producer.send(batch),
-                    producer.createBatch(options).map(newBatch -> {
-                        currentBatch.set(newBatch);
-                        // Add the event that did not fit in the previous batch.
-                        try {
-                            if (!newBatch.tryAdd(event)) {
-                                LOGGER.error("Event was too large to fit in an empty batch. Max size:{} ",
-                                    newBatch.getMaxSizeInBytes());
-                            }
-                        } catch (AmqpException e) {
-                            LOGGER.error("Event was too large to fit in an empty batch. Max size:{}",
-                                newBatch.getMaxSizeInBytes(), e);
+            } catch (AmqpException e) {
+                LOGGER.error("Event is larger than maximum allowed size.", e);
+                return Mono.empty();
+            }
+
+            return Mono.when(
+                producer.send(batch),
+                producer.createBatch(options).map(newBatch -> {
+                    currentBatch.set(newBatch);
+                    // Add the event that did not fit in the previous batch.
+                    try {
+                        if (!newBatch.tryAdd(event)) {
+                            LOGGER.error("Event was too large to fit in an empty batch. Max size:{} ",
+                                newBatch.getMaxSizeInBytes());
                         }
+                    } catch (AmqpException e) {
+                        LOGGER.error("Event was too large to fit in an empty batch. Max size:{}",
+                            newBatch.getMaxSizeInBytes(), e);
+                    }
 
-                        return newBatch;
-                    }));
-            })
-            .then()
-            .block();
+                    return newBatch;
+                }));
+        })
+        .then()
+        .block();
 
-            final EventDataBatch batch = currentBatch.getAndSet(null);
-            return producer.send(batch);
-        }
+        final EventDataBatch batch = currentBatch.getAndSet(null);
+        return producer.send(batch)
+                .doFinally(s -> producer.close());
     }
 
     private CreateBatchOptions buildCreateBatchOptions(PartitionSupplier partitionSupplier) {
diff --git a/sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusTemplate.java b/sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusTemplate.java
index 4678c33efd0e..d04ee07ba1b5 100644
--- a/sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusTemplate.java
+++ b/sdk/spring/spring-messaging-azure-servicebus/src/main/java/com/azure/spring/messaging/servicebus/core/ServiceBusTemplate.java
@@ -41,11 +41,11 @@ public ServiceBusTemplate(@NonNull ServiceBusProducerFactory producerFactory) {
     @Override
     public <U> Mono<Void> sendAsync(String destination, Message<U> message) {
         Assert.hasText(destination, "destination can't be null or empty");
-        try (ServiceBusSenderAsyncClient senderAsyncClient =
-                     this.producerFactory.createProducer(destination, defaultEntityType)) {
-            ServiceBusMessage serviceBusMessage = messageConverter.fromMessage(message, ServiceBusMessage.class);
-            return senderAsyncClient.sendMessage(serviceBusMessage);
-        }
+        ServiceBusSenderAsyncClient senderAsyncClient =
+                     this.producerFactory.createProducer(destination, defaultEntityType);
+        ServiceBusMessage serviceBusMessage = messageConverter.fromMessage(message, ServiceBusMessage.class);
+        return senderAsyncClient.sendMessage(serviceBusMessage)
+                .doFinally(t -> senderAsyncClient.close());
     }
 
     /**