DATAGO-137655: time-bound producer close to fix unbind hang on reconnect

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/outbound/JCSMPOutboundMessageHandler.java

@@ -303,11 +303,12 @@ private void createProducerInternal() {
 	}
 
 	private void recreateProducerIfNeeded(ErrorChannelSendingCorrelationKey correlationKey) throws MessagingException {
-		if (!recreateProducer && !producer.isClosed()) {
+		// producer may be null after closeResources() (e.g. a concurrent stop()); null means "recreate".
+		if (!recreateProducer && producer != null && !producer.isClosed()) {
 			return;
 		}
 		synchronized (lifecycleLock) {
-			if (!recreateProducer && !producer.isClosed()) {
+			if (!recreateProducer && producer != null && !producer.isClosed()) {
 				return;
 			}
 			LOGGER.debug("Recreating JCSMP producer for binding {} after stale-flow detection <message handler ID: {}>",
@@ -340,13 +341,23 @@ private void closeResources() {
 		synchronized (lifecycleLock) {
 			LOGGER.info("Stopping producer to {} {} <message handler ID: {}>", configDestinationType, configDestination.getName(), id);
 			recreateProducer = false;
-			if (producer != null) {
-				LOGGER.info("Closing producer <message handler ID: {}>", id);
-				producer.close();
-			}
-			if (transactedSession != null) {
-				LOGGER.info("Closing transacted session <message handler ID: {}>", id);
-				transactedSession.close();
+			final XMLMessageProducer producerToClose = producer;
+			final TransactedSession transactedSessionToClose = transactedSession;
+			if (producerToClose != null || transactedSessionToClose != null) {
+				// Time-bounded close (DATAGO-137655): one budget for both resources, not 2x.
+				producerManager.closeSafely(() -> {
+					if (producerToClose != null) {
+						LOGGER.info("Closing producer <message handler ID: {}>", id);
+						producerToClose.close();
+					}
+					if (transactedSessionToClose != null) {
+						LOGGER.info("Closing transacted session <message handler ID: {}>", id);
+						transactedSessionToClose.close();
+					}
+				}, "producer/transacted session <message handler ID: " + id + ">");
+				// Drop references even if the close timed out; the handler no longer owns them.
+				producer = null;
+				transactedSession = null;
 			}
 			producerManager.release(id);
 		}

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

@@ -1,6 +1,7 @@
 package com.solace.spring.cloud.stream.binder.properties;
 
 
+import com.solace.spring.cloud.stream.binder.util.JCSMPSessionProducerManager;
 import com.solace.spring.cloud.stream.binder.util.SessionInitializationMode;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
@@ -10,11 +11,25 @@ public class SolaceBinderConfigurationProperties {
 
   private SessionInitializationMode sessionInitializationMode = SessionInitializationMode.EAGER;
 
+  /**
+   * Milliseconds to wait for a JCSMP producer close on binding stop before interrupting it
+   * (DATAGO-137655); {@code producer.close()} blocks indefinitely while the session is reconnecting.
+   */
+  private long producerCloseTimeoutInMillis = JCSMPSessionProducerManager.DEFAULT_CLOSE_TIMEOUT_IN_MILLIS;
+
   public SessionInitializationMode getSessionInitializationMode() {
     return sessionInitializationMode;
   }
 
   public void setSessionInitializationMode(SessionInitializationMode sessionInitializationMode) {
     this.sessionInitializationMode = sessionInitializationMode;
   }
-}
+
+  public long getProducerCloseTimeoutInMillis() {
+    return producerCloseTimeoutInMillis;
+  }
+
+  public void setProducerCloseTimeoutInMillis(long producerCloseTimeoutInMillis) {
+    this.producerCloseTimeoutInMillis = producerCloseTimeoutInMillis;
+  }
+}

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

@@ -10,18 +10,50 @@
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
 import java.util.Optional;
 import java.util.UUID;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 public class JCSMPSessionProducerManager extends SharedResourceManager<XMLMessageProducer> {
+	/** Default producer-close timeout used when none is supplied (e.g. by tests). */
+	public static final long DEFAULT_CLOSE_TIMEOUT_IN_MILLIS = 10000L;
+
 	private final SolaceSessionManager solaceSessionManager;
 	private final CloudStreamEventHandler publisherEventHandler = new CloudStreamEventHandler();
+	private final long closeTimeoutInMillis;
+	private final ExecutorService closeExecutor = Executors.newCachedThreadPool(newCloseThreadFactory());
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(JCSMPSessionProducerManager.class);
 
 	public JCSMPSessionProducerManager(SolaceSessionManager solaceSessionManager) {
+		this(solaceSessionManager, DEFAULT_CLOSE_TIMEOUT_IN_MILLIS);
+	}
+
+	public JCSMPSessionProducerManager(SolaceSessionManager solaceSessionManager, long closeTimeoutInMillis) {
 		super("producer");
 		this.solaceSessionManager = solaceSessionManager;
+		if (closeTimeoutInMillis <= 0) {
+			LOGGER.warn("Invalid producer close timeout {} ms; falling back to default {} ms",
+					closeTimeoutInMillis, DEFAULT_CLOSE_TIMEOUT_IN_MILLIS);
+			this.closeTimeoutInMillis = DEFAULT_CLOSE_TIMEOUT_IN_MILLIS;
+		} else {
+			this.closeTimeoutInMillis = closeTimeoutInMillis;
+		}
+	}
+
+	private static ThreadFactory newCloseThreadFactory() {
+		CustomizableThreadFactory threadFactory = new CustomizableThreadFactory("solace-producer-close-");
+		threadFactory.setDaemon(true);
+		return threadFactory;
 	}
 
 	@Override
@@ -30,8 +62,58 @@ XMLMessageProducer create() throws JCSMPException {
 	}
 
 	@Override
-	void close() {
-		sharedResource.close();
+	void close(XMLMessageProducer resource) {
+		// Guard the shared-producer close (last-user close via release()); see closeSafely().
+		closeSafely(resource::close, "shared producer");
+	}
+
+	/**
+	 * Closes a JCSMP producer-side resource with an upper time bound. {@code close()} parks in
+	 * {@code JCSMPBasicSession.waitUntilSessionReconnectDone} while the session is reconnecting
+	 * (DATAGO-137655); the close runs on a daemon thread and is interrupted after
+	 * {@link #closeTimeoutInMillis} so a producer stop cannot wedge the binding lifecycle. Never
+	 * propagates an exception.
+	 *
+	 * @param closeAction the blocking {@code close()} to run (e.g. {@code producer::close})
+	 * @param description human-readable resource description, used in logging
+	 */
+	public void closeSafely(Runnable closeAction, String description) {
+		Future<?> future;
+		try {
+			future = closeExecutor.submit(closeAction);
+		} catch (RejectedExecutionException e) {
+			// Executor already shut down (binder destroyed); nothing left to bound the close against.
+			LOGGER.warn("Close executor unavailable (already shut down); skipping bounded close of {}", description);
+			return;
+		}
+		try {
+			future.get(closeTimeoutInMillis, TimeUnit.MILLISECONDS);
+		} catch (TimeoutException e) {
+			LOGGER.warn("Timed out after {} ms closing {}; interrupting and proceeding "
+					+ "(session is likely reconnecting)", closeTimeoutInMillis, description);
+			future.cancel(true);
+		} catch (ExecutionException e) {
+			LOGGER.warn("Error while closing {}", description, e.getCause());
+		} catch (InterruptedException e) {
+			future.cancel(true);
+			Thread.currentThread().interrupt();
+			LOGGER.warn("Interrupted while closing {}; proceeding", description);
+		}
+	}
+
+	/** Shuts down the close executor. Should be called when the owning binder is destroyed. */
+	public void shutdown() {
+		closeExecutor.shutdownNow();
+		try {
+			if (!closeExecutor.awaitTermination(closeTimeoutInMillis, TimeUnit.MILLISECONDS)) {
+				// A worker still blocked in JCSMP (interrupt ignored) survives as a daemon thread; surface it.
+				LOGGER.warn("Close executor did not terminate within {} ms; one or more producer-close "
+						+ "workers may still be blocked (session likely still reconnecting)", closeTimeoutInMillis);
+			}
+		} catch (InterruptedException e) {
+			Thread.currentThread().interrupt();
+			LOGGER.warn("Interrupted while awaiting close-executor termination");
+		}
 	}
 
 	public static class CloudStreamEventHandler implements JCSMPStreamingPublishCorrelatingEventHandler {

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

@@ -19,7 +19,14 @@ abstract class SharedResourceManager<T> {
 	}
 
 	abstract T create() throws Exception;
-	abstract void close();
+
+	/**
+	 * Close a resource previously returned by {@link #create()}. Invoked after {@link #lock} is
+	 * released, so a blocking {@code close} cannot stall concurrent {@code get()}/{@code release()}.
+	 *
+	 * @param resource the resource to close (never {@code null})
+	 */
+	abstract void close(T resource);
 
 	/**
 	 * Register {@code key} to the shared resource.
@@ -53,39 +60,49 @@ public T get(String key) throws Exception {
 	 * @throws Exception whatever {@link #create()} may throw
 	 */
 	public T forceRecreate(T expected) throws Exception {
+		T toClose;
+		T current;
 		synchronized (lock) {
 			if (sharedResource != expected) {
 				return sharedResource;
 			}
-			if (sharedResource != null) {
-				try {
-					close();
-				} catch (Exception e) {
-					LOGGER.debug("Failed to close current {} during forceRecreate", type, e);
-				}
-			}
+			toClose = sharedResource;
 			sharedResource = create();
-			return sharedResource;
+			current = sharedResource;
+		}
+		// Close the previous resource outside the lock; the replacement is already installed.
+		if (toClose != null) {
+			try {
+				close(toClose);
+			} catch (Exception e) {
+				LOGGER.debug("Failed to close previous {} during forceRecreate", type, e);
+			}
 		}
+		return current;
 	}
 
 	/**
 	 * De-register {@code key} from the shared resource.
-	 * <p>If this is the last {@code key} associated to the shared resource, {@link #close()} the resource.
+	 * <p>If this is the last {@code key} associated to the shared resource, {@link #close(Object)} the resource.
 	 * @param key the registration key of the caller that is using the resource
 	 */
 	public void release(String key) {
+		T toClose = null;
 		synchronized (lock) {
 			if (!registeredIds.contains(key)) return;
 
 			if (registeredIds.size() <= 1) {
 				LOGGER.info("{} is the last user, closing {}...", key, type);
-				close();
+				toClose = sharedResource;
 				sharedResource = null;
 			} else {
 				LOGGER.info("{} is not the last user, persisting {}...", key, type);
 			}
 			registeredIds.remove(key);
 		}
+		// Close the last-user resource outside the lock so a blocking close can't stall other callers.
+		if (toClose != null) {
+			close(toClose);
+		}
 	}
 }

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/SolaceMessageChannelBinder.java

@@ -61,9 +61,14 @@ public class SolaceMessageChannelBinder
 	@Nullable private SolaceBinderHealthAccessor solaceBinderHealthAccessor;
 
 	public SolaceMessageChannelBinder(SolaceSessionManager solaceSessionManager, SolaceEndpointProvisioner solaceEndpointProvisioner) {
+		this(solaceSessionManager, solaceEndpointProvisioner, JCSMPSessionProducerManager.DEFAULT_CLOSE_TIMEOUT_IN_MILLIS);
+	}
+
+	public SolaceMessageChannelBinder(SolaceSessionManager solaceSessionManager, SolaceEndpointProvisioner solaceEndpointProvisioner,
+									  long producerCloseTimeoutInMillis) {
 		super(new String[0], solaceEndpointProvisioner);
 		this.solaceSessionManager = solaceSessionManager;
-		this.sessionProducerManager = new JCSMPSessionProducerManager(solaceSessionManager);
+		this.sessionProducerManager = new JCSMPSessionProducerManager(solaceSessionManager, producerCloseTimeoutInMillis);
 	}
 
 	@Override
@@ -73,9 +78,14 @@ public String getBinderIdentity() {
 
 	@Override
 	public void destroy() {
-		sessionProducerManager.release(errorHandlerProducerKey);
-		consumersRemoteStopFlag.set(true);
-		solaceSessionManager.close();
+		try {
+			sessionProducerManager.release(errorHandlerProducerKey);
+			consumersRemoteStopFlag.set(true);
+			solaceSessionManager.close();
+		} finally {
+			// Always reap the close executor, even if release()/session close throws.
+			sessionProducerManager.shutdown();
+		}
 	}
 
 	@Override

solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/config/SolaceMessageChannelBinderConfiguration.java

@@ -57,7 +57,8 @@ SolaceMessageChannelBinder solaceMessageChannelBinder(
 			@Nullable ProducerMessageHandlerCustomizer<JCSMPOutboundMessageHandler> producerCustomizer,
 			@Nullable SolaceBinderHealthAccessor solaceBinderHealthAccessor,
 			@Nullable SolaceMeterAccessor solaceMeterAccessor) {
-		SolaceMessageChannelBinder binder = new SolaceMessageChannelBinder(solaceSessionManager, solaceEndpointProvisioner);
+		SolaceMessageChannelBinder binder = new SolaceMessageChannelBinder(solaceSessionManager, solaceEndpointProvisioner,
+				solaceBinderConfigurationProperties.getProducerCloseTimeoutInMillis());
 		binder.setExtendedBindingProperties(solaceExtendedBindingProperties);
 		binder.setProducerMessageHandlerCustomizer(producerCustomizer);
 		binder.setSolaceMeterAccessor(solaceMeterAccessor);