From 72b81ad54994d1bf7313b5ec2856d46367f39d67 Mon Sep 17 00:00:00 2001 From: sunil-solace Date: Tue, 23 Jun 2026 18:40:58 -0400 Subject: [PATCH] DATAGO-137655: time-bound producer close to fix unbind hang on reconnect When the JCSMP session is reconnecting (e.g. a broker / Message VPN outage with unlimited reconnect retries), XMLMessageProducer.close() and TransactedSession.close() block indefinitely in JCSMPBasicSession.waitUntilSessionReconnectDone. These closes run on the binding-lifecycle thread during Binding.unbind() -> JCSMPOutboundMessageHandler.stop(), so the unbind never returns and the producer binding lifecycle stays wedged until the process is restarted. Run the producer-side closes (the per-handler producer/transacted session and the shared producer) on a daemon-thread executor with an upper time bound. If a close does not complete within the timeout, the worker is interrupted and the stop proceeds, so a producer unbind can no longer hang while the session is reconnecting. The bound is configurable via spring.cloud.stream.solace.binder.producer-close-timeout-in-millis (default 10000 ms). Adds JCSMPCloseFlowReconnectHangIT, which reproduces the hang against a broker by disabling the Message VPN to force a perpetual reconnect, then asserts the unbind completes. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../outbound/JCSMPOutboundMessageHandler.java | 29 ++- .../SolaceBinderConfigurationProperties.java | 17 +- .../util/JCSMPSessionProducerManager.java | 86 ++++++- .../binder/util/SharedResourceManager.java | 39 +++- .../binder/SolaceMessageChannelBinder.java | 18 +- ...laceMessageChannelBinderConfiguration.java | 3 +- .../binder/JCSMPCloseFlowReconnectHangIT.java | 212 ++++++++++++++++++ 7 files changed, 376 insertions(+), 28 deletions(-) create mode 100644 solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/test/java/com/solace/spring/cloud/stream/binder/JCSMPCloseFlowReconnectHangIT.java diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/outbound/JCSMPOutboundMessageHandler.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/outbound/JCSMPOutboundMessageHandler.java index 0baf6bb99..975fbb551 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/outbound/JCSMPOutboundMessageHandler.java +++ b/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 ", @@ -340,13 +341,23 @@ private void closeResources() { synchronized (lifecycleLock) { LOGGER.info("Stopping producer to {} {} ", configDestinationType, configDestination.getName(), id); recreateProducer = false; - if (producer != null) { - LOGGER.info("Closing producer ", id); - producer.close(); - } - if (transactedSession != null) { - LOGGER.info("Closing transacted session ", 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 ", id); + producerToClose.close(); + } + if (transactedSessionToClose != null) { + LOGGER.info("Closing transacted session ", id); + transactedSessionToClose.close(); + } + }, "producer/transacted session "); + // Drop references even if the close timed out; the handler no longer owns them. + producer = null; + transactedSession = null; } producerManager.release(id); } diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/properties/SolaceBinderConfigurationProperties.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/properties/SolaceBinderConfigurationProperties.java index 5778d3eef..005ce5bd1 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/properties/SolaceBinderConfigurationProperties.java +++ b/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,6 +11,12 @@ 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; } @@ -17,4 +24,12 @@ public SessionInitializationMode getSessionInitializationMode() { public void setSessionInitializationMode(SessionInitializationMode sessionInitializationMode) { this.sessionInitializationMode = sessionInitializationMode; } -} \ No newline at end of file + + public long getProducerCloseTimeoutInMillis() { + return producerCloseTimeoutInMillis; + } + + public void setProducerCloseTimeoutInMillis(long producerCloseTimeoutInMillis) { + this.producerCloseTimeoutInMillis = producerCloseTimeoutInMillis; + } +} diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/JCSMPSessionProducerManager.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/JCSMPSessionProducerManager.java index 9419e237a..dc3074a2c 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/JCSMPSessionProducerManager.java +++ b/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 { + /** 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 { diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/SharedResourceManager.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/SharedResourceManager.java index f0735d6fb..cafa8d791 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder-core/src/main/java/com/solace/spring/cloud/stream/binder/util/SharedResourceManager.java +++ b/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 { } 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. - *

If this is the last {@code key} associated to the shared resource, {@link #close()} the resource. + *

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); + } } } diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/SolaceMessageChannelBinder.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/SolaceMessageChannelBinder.java index db484225b..da95eeabf 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/SolaceMessageChannelBinder.java +++ b/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 diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/config/SolaceMessageChannelBinderConfiguration.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/config/SolaceMessageChannelBinderConfiguration.java index 21ef1bad8..9743e4e27 100644 --- a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/main/java/com/solace/spring/cloud/stream/binder/config/SolaceMessageChannelBinderConfiguration.java +++ b/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 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); diff --git a/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/test/java/com/solace/spring/cloud/stream/binder/JCSMPCloseFlowReconnectHangIT.java b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/test/java/com/solace/spring/cloud/stream/binder/JCSMPCloseFlowReconnectHangIT.java new file mode 100644 index 000000000..a0508eb85 --- /dev/null +++ b/solace-spring-cloud-stream-binder/solace-spring-cloud-stream-binder/src/test/java/com/solace/spring/cloud/stream/binder/JCSMPCloseFlowReconnectHangIT.java @@ -0,0 +1,212 @@ +package com.solace.spring.cloud.stream.binder; + +import static org.assertj.core.api.Assertions.assertThatCode; + +import com.solace.it.util.semp.config.BrokerConfiguratorBuilder; +import com.solace.it.util.semp.config.BrokerConfiguratorBuilder.BrokerConfigurator; +import com.solace.spring.boot.autoconfigure.SolaceJavaAutoConfiguration; +import com.solace.spring.cloud.stream.binder.properties.SolaceConsumerProperties; +import com.solace.spring.cloud.stream.binder.properties.SolaceProducerProperties; +import com.solace.spring.cloud.stream.binder.test.junit.extension.SpringCloudStreamExtension; +import com.solace.spring.cloud.stream.binder.test.spring.SpringCloudStreamContext; +import com.solace.spring.cloud.stream.binder.test.util.SolaceTestBinder; +import com.solace.spring.cloud.stream.binder.util.DestinationType; +import com.solace.test.integration.junit.jupiter.extension.PubSubPlusExtension; +import com.solace.test.integration.semp.v2.SempV2Api; +import com.solace.test.integration.semp.v2.config.model.ConfigMsgVpn; +import com.solacesystems.jcsmp.JCSMPChannelProperties; +import com.solacesystems.jcsmp.JCSMPProperties; +import org.apache.commons.lang3.RandomStringUtils; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInfo; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.boot.test.context.ConfigDataApplicationContextInitializer; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +import org.springframework.cloud.stream.config.BindingProperties; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.MessageChannel; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import java.time.Duration; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +/** + * DATAGO-137655 / SOL-50004: closing a bound flow while the JCSMP session is reconnecting blocks in + * {@code JCSMPBasicSession.waitUntilSessionReconnectDone} until the thread is interrupted. + * + *

Reproduction: bind a flow against a session with {@code reconnectRetries = -1}, disable the + * Message VPN to force a perpetual {@code RECONNECTING} loop, then unbind. The producer close is + * bounded by {@code JCSMPSessionProducerManager.closeSafely}; the consumer close by the existing + * {@code JCSMPInboundChannelAdapter.stopAllConsumers()} interrupt. Both unbinds must return within + * {@link #EXPECTED_CLOSE_COMPLETION}. + * + *

Requires a broker (Testcontainers / {@link PubSubPlusExtension}); runs SAME_THREAD because it + * disables a shared, broker-side Message VPN. + */ +@SpringJUnitConfig(classes = SolaceJavaAutoConfiguration.class, + initializers = ConfigDataApplicationContextInitializer.class) +@ExtendWith(PubSubPlusExtension.class) +@ExtendWith(SpringCloudStreamExtension.class) +@Execution(ExecutionMode.SAME_THREAD) +class JCSMPCloseFlowReconnectHangIT { + private static final Logger logger = LoggerFactory.getLogger(JCSMPCloseFlowReconnectHangIT.class); + + /** Max time an {@code unbind()} may take during reconnect (the producer can close two producers at ~10s each). */ + private static final Duration EXPECTED_CLOSE_COMPLETION = Duration.ofSeconds(30); + + private SempV2Api sempV2Api; + private SpringCloudStreamContext context; + private String vpnName; + private BrokerConfigurator brokerConfig; + + // Per-test artifacts cleaned up in @AfterEach. + private Binding producerBinding; + private Binding consumerBinding; + private boolean vpnDisabled; + + @BeforeEach + void setUp(JCSMPProperties jcsmpProperties, SempV2Api sempV2Api, SpringCloudStreamContext context) { + this.sempV2Api = sempV2Api; + this.context = context; + this.vpnName = (String) jcsmpProperties.getProperty(JCSMPProperties.VPN_NAME); + this.brokerConfig = BrokerConfiguratorBuilder.create(sempV2Api).build(); + + // Unlimited reconnect-retries is the SOL-50004 precondition: the session must stay in + // RECONNECTING (never transition to DOWN) while the VPN is disabled, otherwise close() + // would eventually return on its own and the hang would not reproduce. + JCSMPChannelProperties channelProperties = (JCSMPChannelProperties) jcsmpProperties + .getProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES); + if (channelProperties == null) { + channelProperties = new JCSMPChannelProperties(); + } + channelProperties.setReconnectRetries(-1); + channelProperties.setReconnectRetryWaitInMillis(2000); + channelProperties.setConnectRetries(1); + jcsmpProperties.setProperty(JCSMPProperties.CLIENT_CHANNEL_PROPERTIES, channelProperties); + context.setJcsmpProperties(jcsmpProperties); + } + + @AfterEach + void tearDown() { + // Re-enable the VPN first so any still-blocked close can drain and resources can be freed. + if (vpnDisabled) { + try { + brokerConfig.vpns().updateVpn(new ConfigMsgVpn().msgVpnName(vpnName).enabled(true)); + } catch (Exception e) { + logger.warn("Failed to re-enable VPN '{}' during cleanup", vpnName, e); + } + } + if (producerBinding != null) { + try { + producerBinding.unbind(); + } catch (Exception e) { + logger.warn("Failed to unbind producer binding during cleanup", e); + } + } + if (consumerBinding != null) { + try { + consumerBinding.unbind(); + } catch (Exception e) { + logger.warn("Failed to unbind consumer binding during cleanup", e); + } + } + } + + /** Producer (DATAGO-137655): unbinding while the session is reconnecting must not hang. */ + @Test + @Timeout(value = 90, unit = TimeUnit.SECONDS) + void testProducerStopDoesNotHangWhileSessionReconnecting(TestInfo testInfo) throws Exception { + SolaceTestBinder binder = context.getBinder(); + ExtendedProducerProperties producerProperties = + context.createProducerProperties(testInfo); + producerProperties.getExtension().setDestinationType(DestinationType.TOPIC); + BindingProperties bindingProperties = new BindingProperties(); + bindingProperties.setProducer(producerProperties); + + DirectChannel outputChannel = context.createBindableChannel("output", bindingProperties); + String destination = RandomStringUtils.randomAlphanumeric(10); + producerBinding = binder.bindProducer(destination, outputChannel, producerProperties); + + // Confirm the producer is healthy (forces creation of the underlying JCSMP producer). + outputChannel.send(MessageBuilder.withPayload("before-disruption").build()); + + disableVpnAndAwaitReconnecting(); + + Binding bindingToStop = producerBinding; + assertClosesWithinTimeout("producer binding unbind()", bindingToStop::unbind); + producerBinding = null; // already unbound + } + + /** Message-driven consumer: unbinding while the session is reconnecting must not hang. */ + @Test + @Timeout(value = 90, unit = TimeUnit.SECONDS) + void testConsumerStopDoesNotHangWhileSessionReconnecting() throws Exception { + SolaceTestBinder binder = context.getBinder(); + ExtendedConsumerProperties consumerProperties = + context.createConsumerProperties(); + + DirectChannel inputChannel = context.createBindableChannel("input", new BindingProperties(), true); + String destination = RandomStringUtils.randomAlphanumeric(10); + String group = RandomStringUtils.randomAlphanumeric(10); + // Binding a consumer provisions the queue and starts an active flow. + consumerBinding = binder.bindConsumer(destination, group, inputChannel, consumerProperties); + + disableVpnAndAwaitReconnecting(); + + Binding bindingToStop = consumerBinding; + assertClosesWithinTimeout("consumer binding unbind()", bindingToStop::unbind); + consumerBinding = null; // already unbound + } + + /** + * Disables the Message VPN and waits until the broker has dropped its clients, so the binder's + * session is reconnecting when the subsequent close is issued. + */ + private void disableVpnAndAwaitReconnecting() { + logger.info("Disabling VPN '{}' to force the session into perpetual RECONNECTING", vpnName); + brokerConfig.vpns().updateVpn(new ConfigMsgVpn().msgVpnName(vpnName).enabled(false)); + vpnDisabled = true; + + // Wait until the broker reports the VPN disabled; disabling drops its clients, so the binder's + // session has then lost its connection and entered the (perpetual) reconnect loop. + Awaitility.await("VPN '" + vpnName + "' reported disabled by broker") + .atMost(Duration.ofSeconds(30)) + .pollInterval(Duration.ofMillis(250)) + .ignoreExceptions() + .until(() -> Boolean.FALSE.equals( + sempV2Api.monitor().getMsgVpn(vpnName, null).getData().isEnabled())); + } + + /** Runs {@code close} on a daemon thread and asserts it returns within {@link #EXPECTED_CLOSE_COMPLETION} (pre-fix it hangs). */ + private void assertClosesWithinTimeout(String description, Runnable close) { + ExecutorService executor = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable, "reconnect-hang-close-invoker"); + thread.setDaemon(true); + return thread; + }); + try { + Future future = executor.submit(close); + assertThatCode(() -> future.get(EXPECTED_CLOSE_COMPLETION.toMillis(), TimeUnit.MILLISECONDS)) + .as("%s must complete within %s while the session is reconnecting; pre-fix it hangs " + + "forever in JCSMPBasicSession.waitUntilSessionReconnectDone", description, + EXPECTED_CLOSE_COMPLETION) + .doesNotThrowAnyException(); + } finally { + executor.shutdownNow(); + } + } +}