From 60dc49450bdec4d0b04f9b6d73258d6abfae81ab Mon Sep 17 00:00:00 2001 From: Bartosz Popiela Date: Sun, 30 Aug 2026 23:52:45 +0200 Subject: [PATCH 1/3] CAMEL-24569: Support Salesforce Streaming API disconnect messages Handle server-initiated /meta/disconnect messages introduced in Salesforce Streaming API 64.0 and reconnect to keep subscriptions active. --- .../streaming/SubscriptionHelper.java | 337 ++++++++++++------ .../streaming/SubscriptionHelperManualIT.java | 76 ++++ 2 files changed, 295 insertions(+), 118 deletions(-) diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java index 0960afa4b8d19..a004e0b3db6d2 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java @@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -67,6 +68,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import static org.cometd.bayeux.Channel.META_CONNECT; +import static org.cometd.bayeux.Channel.META_DISCONNECT; import static org.cometd.bayeux.Channel.META_HANDSHAKE; import static org.cometd.bayeux.Channel.META_SUBSCRIBE; import static org.cometd.bayeux.Message.ERROR_FIELD; @@ -107,6 +109,8 @@ public class SubscriptionHelper extends ServiceSupport { private final AtomicLong handshakeBackoff; + private final AtomicBoolean reconnecting = new AtomicBoolean(false); + private final Map> channelToConsumers = new ConcurrentHashMap<>(); private final Map consumerToListener @@ -121,6 +125,8 @@ public class SubscriptionHelper extends ServiceSupport { private final ClientSessionChannel.MessageListener connectListener = createConnectionListener(); + private final ClientSessionChannel.MessageListener disconnectListener = createDisconnectListener(); + public SubscriptionHelper(final SalesforceComponent component) { this.component = component; handshakeBackoff = new AtomicLong(); @@ -129,108 +135,184 @@ public SubscriptionHelper(final SalesforceComponent component) { } private MessageListener createHandshakeListener() { - return (channel, message) -> component.getHttpClient().getWorkerPool().execute(() -> { - LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message); - - if (!message.isSuccessful()) { - LOG.warn("Handshake failure: {}", message); - handshakeError = (String) message.get(ERROR_FIELD); - handshakeException = getFailure(message); - if (handshakeError != null) { - if (handshakeError.startsWith("403::")) { - String failureReason = getFailureReason(message); - if (AUTHENTICATION_INVALID.equals(failureReason)) { - LOG.debug( - "attempting login due to handshake error: 403 -> 401::Authentication invalid"); - session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff); - } - } - } - // failed, so keep trying with backoff - final long backoff = handshakeBackoff.getAndAdd(backoffIncrement); - if (backoff > maxBackoff) { - LOG.error("Handshake retry aborted after exceeding {} msecs backoff", maxBackoff); - } else { - LOG.debug("Pausing for {} msecs before handshake retry", backoff); - if (backoff > 0) { - Tasks.backgroundTask() - .withBudget(Budgets.iterationTimeBudget() - .withMaxIterations(1) - .withInitialDelay(Duration.ofMillis(backoff)) - .withInterval(Duration.ofMillis(1)) - .withUnlimitedDuration() - .build()) - .withScheduledExecutor(taskExecutor) - .withName("SalesforceHandshakeRetryDelay") - .build() - .run(component.getCamelContext(), () -> true); - } - client.handshake(); - } - } else if (!channelToConsumers.isEmpty()) { - channelsLock.lock(); - try { - channelsToSubscribe.clear(); - channelsToSubscribe.addAll(channelToConsumers.keySet()); - } finally { - channelsLock.unlock(); - } - LOG.info("Handshake successful. Channels to subscribe: {}", channelsToSubscribe); - } - }); + return (channel, message) -> component + .getHttpClient() + .getWorkerPool() + .execute( + () -> { + LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message); + + if (!message.isSuccessful()) { + LOG.warn("Handshake failure: {}", message); + handshakeError = (String) message.get(ERROR_FIELD); + handshakeException = getFailure(message); + if (handshakeError != null) { + if (handshakeError.startsWith("403::")) { + String failureReason = getFailureReason(message); + if (AUTHENTICATION_INVALID.equals(failureReason)) { + LOG.debug( + "attempting login due to handshake error: 403 -> 401::Authentication invalid"); + session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff); + } + } + } + // failed, so keep trying with backoff + final long backoff = handshakeBackoff.getAndAdd(backoffIncrement); + if (backoff > maxBackoff) { + LOG.error("Handshake retry aborted after exceeding {} msecs backoff", maxBackoff); + } else { + LOG.debug("Pausing for {} msecs before handshake retry", backoff); + if (backoff > 0) { + Tasks.backgroundTask() + .withBudget(Budgets.iterationTimeBudget() + .withMaxIterations(1) + .withInitialDelay(Duration.ofMillis(backoff)) + .withInterval(Duration.ofMillis(1)) + .withUnlimitedDuration() + .build()) + .withScheduledExecutor(taskExecutor) + .withName("SalesforceHandshakeRetryDelay") + .build() + .run(component.getCamelContext(), () -> true); + } + client.handshake(); + } + } else if (!channelToConsumers.isEmpty()) { + channelsLock.lock(); + try { + channelsToSubscribe.clear(); + channelsToSubscribe.addAll(channelToConsumers.keySet()); + } finally { + channelsLock.unlock(); + } + LOG.info("Handshake successful. Channels to subscribe: {}", channelsToSubscribe); + } + }); } private MessageListener createConnectionListener() { - return (channel, message) -> component.getHttpClient().getWorkerPool().execute(() -> { - LOG.debug("[CHANNEL:META_CONNECT]: {}", message); - String reconnectAdvice = message.getAdvice() != null - ? (String) message.getAdvice().get("reconnect") - : null; + return (channel, message) -> component + .getHttpClient() + .getWorkerPool() + .execute( + () -> { + LOG.debug("[CHANNEL:META_CONNECT]: {}", message); + String reconnectAdvice = message.getAdvice() != null + ? (String) message.getAdvice().get("reconnect") + : null; + + if (!message.isSuccessful()) { + LOG.warn("Connect failure: {}", message); + connectError = (String) message.get(ERROR_FIELD); + connectException = getFailure(message); + + if (connectError != null && connectError.equals(AUTHENTICATION_INVALID)) { + LOG.debug("connectError: {}", connectError); + LOG.debug("Attempting login..."); + session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff); + } + // Per Bayeux spec: handshake on null advice, "none", "handshake", or any non-"retry" value. + // When advice is "retry", the CometD client handles reconnection automatically. + if (reconnectAdvice == null || !"retry".equals(reconnectAdvice)) { + LOG.debug("Reconnect advice [{}] on failed connect, initiating handshake", reconnectAdvice); + client.handshake(); + } else if (isTemporaryError(message)) { + LOG.debug("Initiating handshake after temporary error: {}", message); + client.handshake(); + } + } else if (reconnectAdvice != null && !"retry".equals(reconnectAdvice)) { + LOG.warn("Reconnect advice [{}] on successful connect, initiating handshake", reconnectAdvice); + client.handshake(); + } else { + Set toSubscribe = null; + channelsLock.lock(); + try { + if (!channelsToSubscribe.isEmpty()) { + toSubscribe = new HashSet<>(channelsToSubscribe); + channelsToSubscribe.clear(); + } + } finally { + channelsLock.unlock(); + } + if (toSubscribe != null) { + LOG.info("Subscribing to channels: {}", toSubscribe); + for (var channelName : toSubscribe) { + var consumers = channelToConsumers.getOrDefault(channelName, emptySet()); + for (var consumer : consumers) { + subscribe(consumer); + } + } + } + } + }); + } - if (!message.isSuccessful()) { - LOG.warn("Connect failure: {}", message); - connectError = (String) message.get(ERROR_FIELD); - connectException = getFailure(message); - - if (connectError != null && connectError.equals(AUTHENTICATION_INVALID)) { - LOG.debug("connectError: {}", connectError); - LOG.debug("Attempting login..."); - session.attemptLoginUntilSuccessful(backoffIncrement, maxBackoff); + private MessageListener createDisconnectListener() { + return (channel, message) -> component + .getHttpClient() + .getWorkerPool() + .execute( + () -> { + LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message); + + if (isStoppingOrStopped()) { + LOG.debug("Ignoring disconnect message while stopping"); + return; + } + + LOG.info("Server disconnect received, reconnecting to Streaming API"); + attemptReconnectUntilSuccessful(); + }); + } + + private void attemptReconnectUntilSuccessful() { + if (isStoppingOrStopped()) { + return; + } + if (!reconnecting.compareAndSet(false, true)) { + LOG.debug("Reconnect already in progress"); + return; + } + + try { + if (!channelToConsumers.isEmpty()) { + channelsToSubscribe.clear(); + channelsToSubscribe.addAll(channelToConsumers.keySet()); + LOG.info("Channels to resubscribe after reconnect: {}", channelsToSubscribe); + } + + long backoff = 0; + while (!isStoppingOrStopped()) { + handshakeError = null; + handshakeException = null; + connectError = null; + connectException = null; + + client.handshake(); + final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC, SECONDS); + if (client.waitFor(waitMs, BayeuxClient.State.CONNECTED)) { + LOG.info("Reconnect successful"); + handshakeBackoff.set(0); + return; } - // Per Bayeux spec: handshake on null advice, "none", "handshake", or any non-"retry" value. - // When advice is "retry", the CometD client handles reconnection automatically. - if (reconnectAdvice == null || !"retry".equals(reconnectAdvice)) { - LOG.debug("Reconnect advice [{}] on failed connect, initiating handshake", reconnectAdvice); - client.handshake(); - } else if (isTemporaryError(message)) { - LOG.debug("Initiating handshake after temporary error: {}", message); - client.handshake(); + + LOG.warn("Reconnect attempt failed, retrying..."); + backoff = backoff + backoffIncrement; + if (backoff > maxBackoff) { + backoff = maxBackoff; } - } else if (reconnectAdvice != null && !"retry".equals(reconnectAdvice)) { - LOG.warn("Reconnect advice [{}] on successful connect, initiating handshake", reconnectAdvice); - client.handshake(); - } else { - Set toSubscribe = null; - channelsLock.lock(); try { - if (!channelsToSubscribe.isEmpty()) { - toSubscribe = new HashSet<>(channelsToSubscribe); - channelsToSubscribe.clear(); - } - } finally { - channelsLock.unlock(); - } - if (toSubscribe != null) { - LOG.info("Subscribing to channels: {}", toSubscribe); - for (var channelName : toSubscribe) { - var consumers = channelToConsumers.getOrDefault(channelName, emptySet()); - for (var consumer : consumers) { - subscribe(consumer); - } - } + LOG.debug("Pausing for {} msecs before reconnect attempt", backoff); + Thread.sleep(backoff); + } catch (InterruptedException e) { + LOG.warn("Aborting reconnect on interrupt!", e); + Thread.currentThread().interrupt(); + return; } } - }); + } finally { + reconnecting.set(false); + } } private MessageListener createSubscriptionListener() { @@ -272,8 +354,9 @@ private void subscriptionFailed(StreamingApiConsumer firstConsumer, Message mess } Exception failure = getFailure(message); - String msg = String.format("Error subscribing to %s: %s", firstConsumer.getTopicName(), - failure != null ? failure.getMessage() : error); + String msg = String.format( + "Error subscribing to %s: %s", + firstConsumer.getTopicName(), failure != null ? failure.getMessage() : error); boolean abort = true; LOG.warn(msg); @@ -308,8 +391,7 @@ private void subscriptionFailed(StreamingApiConsumer firstConsumer, Message mess } } else if (error.matches(INVALID_REPLAY_ID_PATTERN)) { abort = false; - long fallBackReplayId - = firstConsumer.getEndpoint().getConfiguration().getFallBackReplayId(); + long fallBackReplayId = firstConsumer.getEndpoint().getConfiguration().getFallBackReplayId(); LOG.warn(error); LOG.warn("Falling back to replayId {} for channel {}", fallBackReplayId, channelName); replayExtension.setReplayId(channelName, fallBackReplayId); @@ -350,6 +432,7 @@ private void initMessageListeners() { client.getChannel(META_HANDSHAKE).addListener(handshakeListener); client.getChannel(META_SUBSCRIBE).addListener(subscriptionListener); client.getChannel(META_CONNECT).addListener(connectListener); + client.getChannel(META_DISCONNECT).addListener(disconnectListener); } private void handshake() throws CamelException { @@ -360,18 +443,19 @@ private void handshake() throws CamelException { if (!client.waitFor(waitMs, BayeuxClient.State.CONNECTED)) { if (handshakeException != null) { throw new CamelException( - String.format("Exception during HANDSHAKE: %s", handshakeException.getMessage()), handshakeException); + String.format("Exception during HANDSHAKE: %s", handshakeException.getMessage()), + handshakeException); } else if (handshakeError != null) { throw new CamelException(String.format("Error during HANDSHAKE: %s", handshakeError)); } else if (connectException != null) { throw new CamelException( - String.format("Exception during CONNECT: %s", connectException.getMessage()), connectException); + String.format("Exception during CONNECT: %s", connectException.getMessage()), + connectException); } else if (connectError != null) { throw new CamelException(String.format("Error during CONNECT: %s", connectError)); } else { throw new CamelException( - String.format("Handshake request timeout after %s seconds", - HANDSHAKE_TIMEOUT_SEC)); + String.format("Handshake request timeout after %s seconds", HANDSHAKE_TIMEOUT_SEC)); } } } @@ -412,6 +496,7 @@ protected void doStop() throws Exception { } closeChannel(META_CONNECT); + closeChannel(META_DISCONNECT); closeChannel(META_SUBSCRIBE); closeChannel(META_HANDSHAKE); @@ -434,7 +519,8 @@ protected void doStop() throws Exception { LOG.debug("Stopped the helper and destroyed the client"); } - static BayeuxClient createClient(final SalesforceComponent component, final SalesforceSession session) + static BayeuxClient createClient( + final SalesforceComponent component, final SalesforceSession session) throws SalesforceException { // use default Jetty client from SalesforceComponent, it's shared by all consumers final SalesforceHttpClient httpClient = component.getConfig().getHttpClient(); @@ -462,7 +548,7 @@ static BayeuxClient createClient(final SalesforceComponent component, final Sale protected void customize(Request request) { super.customize(request); - //accessToken might be null due to lazy login + // accessToken might be null due to lazy login String accessToken = session.getAccessToken(); if (accessToken == null) { try { @@ -510,17 +596,21 @@ public void subscribe(StreamingApiConsumer consumer) { try { // create subscription for consumer final String channelName = getChannelName(consumer.getTopicName()); - channelToConsumers.computeIfAbsent(channelName, key -> ConcurrentHashMap.newKeySet()).add(consumer); + channelToConsumers + .computeIfAbsent(channelName, key -> ConcurrentHashMap.newKeySet()) + .add(consumer); setReplayIdIfAbsent(consumer.getEndpoint()); // channel message listener LOG.info("Subscribing to channel {}...", channelName); - var messageListener = consumerToListener.computeIfAbsent(consumer, key -> (channel, message) -> { - LOG.debug("Received Message: {}", message); - // convert CometD message to Camel Message - consumer.processMessage(channel, message); - }); + var messageListener = consumerToListener.computeIfAbsent( + consumer, + key -> (channel, message) -> { + LOG.debug("Received Message: {}", message); + // convert CometD message to Camel Message + consumer.processMessage(channel, message); + }); // subscribe asynchronously final ClientSessionChannel clientChannel = client.getChannel(channelName); @@ -569,7 +659,8 @@ private void setReplayIdIfAbsent(final SalesforceEndpoint endpoint) { } } - static Optional determineReplayIdFor(final SalesforceEndpoint endpoint, final String topicName) { + static Optional determineReplayIdFor( + final SalesforceEndpoint endpoint, final String topicName) { final String channelName = getChannelName(topicName); final Long replayId = endpoint.getReplayId(); @@ -578,21 +669,27 @@ static Optional determineReplayIdFor(final SalesforceEndpoint endpoint, fi final SalesforceEndpointConfig endpointConfiguration = endpoint.getConfiguration(); final Map endpointInitialReplayIdMap = endpointConfiguration.getInitialReplayIdMap(); - final Long endpointReplayId - = endpointInitialReplayIdMap.getOrDefault(topicName, endpointInitialReplayIdMap.get(channelName)); + final Long endpointReplayId = endpointInitialReplayIdMap.getOrDefault( + topicName, endpointInitialReplayIdMap.get(channelName)); final Long endpointDefaultReplayId = endpointConfiguration.getDefaultReplayId(); final SalesforceEndpointConfig componentConfiguration = component.getConfig(); final Map componentInitialReplayIdMap = componentConfiguration.getInitialReplayIdMap(); - final Long componentReplayId - = componentInitialReplayIdMap.getOrDefault(topicName, componentInitialReplayIdMap.get(channelName)); + final Long componentReplayId = componentInitialReplayIdMap.getOrDefault( + topicName, componentInitialReplayIdMap.get(channelName)); final Long componentDefaultReplayId = componentConfiguration.getDefaultReplayId(); // the endpoint values have priority over component values, and the // default values priority // over give topic values - return Stream.of(replayId, endpointReplayId, componentReplayId, endpointDefaultReplayId, componentDefaultReplayId) - .filter(Objects::nonNull).findFirst(); + return Stream.of( + replayId, + endpointReplayId, + componentReplayId, + endpointDefaultReplayId, + componentDefaultReplayId) + .filter(Objects::nonNull) + .findFirst(); } static String getChannelName(final String topicName) { @@ -646,9 +743,13 @@ static String getEndpointUrl(final SalesforceComponent component) { boolean replayOptionsPresent = component.getConfig().getDefaultReplayId() != null || !component.getConfig().getInitialReplayIdMap().isEmpty(); if (replayOptionsPresent) { - return component.getSession().getInstanceUrl() + "/cometd/replay/" + component.getConfig().getApiVersion(); + return component.getSession().getInstanceUrl() + + "/cometd/replay/" + + component.getConfig().getApiVersion(); } } - return component.getSession().getInstanceUrl() + "/cometd/" + component.getConfig().getApiVersion(); + return component.getSession().getInstanceUrl() + + "/cometd/" + + component.getConfig().getApiVersion(); } } diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java index abbdcbacfabb7..b56f2952eb00e 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java +++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java @@ -271,6 +271,82 @@ void shouldResubscribeOnConnectionFailures() throws InterruptedException { verifyNoMoreInteractions(consumer); } + @Test + void shouldResubscribeOnDisconnectMessage() { + var consumer = createConsumer("Opportunity"); + subscription.subscribe(consumer); + + messages.add(""" + [ + { + "data": { + "event": { + "createdDate": "2020-12-11T13:44:56.891Z", + "replayId": 1, + "type": "created" + }, + "sobject": { + "Id": "0061n00002XWMgVAAX", + "Name": "shouldResubscribeOnDisconnectMessage 1" + } + }, + "channel": "/topic/Opportunity" + }, + { + "clientId": "5ra4927ikfky6cb12juthkpofeu8", + "channel": "/meta/connect", + "id": "$id", + "successful": true + } + ]"""); + verify(consumer, timeout(10000)).processMessage(any(ClientSessionChannel.class), + messageWithName("shouldResubscribeOnDisconnectMessage 1")); + + subscription.client.getChannel("/meta/subscribe").addListener( + (MessageListener) (clientSessionChannel, message) -> { + var channel = (String) message.get("subscription"); + if (channel != null && channel.contains("Opportunity")) { + messages.add(""" + [ + { + "data": { + "event": { + "createdDate": "2020-12-11T13:44:57.891Z", + "replayId": 2, + "type": "created" + }, + "sobject": { + "Id": "0061n00002XWMgVAAX", + "Name": "shouldResubscribeOnDisconnectMessage 2" + } + }, + "channel": "/topic/Opportunity" + }, + { + "clientId": "5ra4927ikfky6cb12juthkpofeu8", + "channel": "/meta/connect", + "id": "$id", + "successful": true + } + ]"""); + } + }); + messages.add(""" + [ + { + "channel": "/meta/disconnect", + "clientId": "5ra4927ikfky6cb12juthkpofeu8" + } + ]"""); + + verify(consumer, timeout(20000)).processMessage(any(ClientSessionChannel.class), + messageWithName("shouldResubscribeOnDisconnectMessage 2")); + + verify(consumer, atLeastOnce()).getEndpoint(); + verify(consumer, atLeastOnce()).getTopicName(); + verifyNoMoreInteractions(consumer); + } + @Test void shouldResubscribeOnSubscriptionFailure() { var consumer = createConsumer("Contact"); From 5037649564cbabe2a28fc5d658701263e28e4ea9 Mon Sep 17 00:00:00 2001 From: Bartosz Popiela Date: Mon, 31 Aug 2026 00:56:17 +0200 Subject: [PATCH 2/3] CAMEL-24569: Improve Salesforce component to be state-safe when processing a disconnect message - Wait for CometD to reach the disconnected state before reconnecting. - Use the managed executor and verify that channels are resubscribed only once. --- .../streaming/SubscriptionHelper.java | 100 ++++++++---------- .../streaming/SubscriptionHelperManualIT.java | 46 ++++---- 2 files changed, 71 insertions(+), 75 deletions(-) diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java index a004e0b3db6d2..f8c3a325ecded 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java @@ -30,6 +30,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; @@ -94,9 +95,9 @@ public class SubscriptionHelper extends ServiceSupport { private static final String DENIED_BY_SEC_POLICY = "403:denied_by_security_policy"; private static final String AUTHORIZATION_ERROR = "403::"; - BayeuxClient client; + volatile BayeuxClient client; - private ScheduledExecutorService taskExecutor; + private volatile ScheduledExecutorService taskExecutor; private final SalesforceComponent component; private SalesforceSession session; @@ -248,67 +249,58 @@ private MessageListener createConnectionListener() { } private MessageListener createDisconnectListener() { - return (channel, message) -> component - .getHttpClient() - .getWorkerPool() - .execute( - () -> { - LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message); + return (channel, message) -> { + LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message); - if (isStoppingOrStopped()) { - LOG.debug("Ignoring disconnect message while stopping"); - return; - } + if (isStoppingOrStopped()) { + LOG.debug("Ignoring disconnect message while stopping"); + return; + } + if (!reconnecting.compareAndSet(false, true)) { + LOG.debug("Reconnect already in progress"); + return; + } - LOG.info("Server disconnect received, reconnecting to Streaming API"); - attemptReconnectUntilSuccessful(); - }); - } + final ScheduledExecutorService executor = taskExecutor; + if (executor == null) { + reconnecting.set(false); + return; + } - private void attemptReconnectUntilSuccessful() { - if (isStoppingOrStopped()) { - return; - } - if (!reconnecting.compareAndSet(false, true)) { - LOG.debug("Reconnect already in progress"); - return; - } + try { + executor.execute(this::reconnectAfterDisconnect); + } catch (RejectedExecutionException e) { + reconnecting.set(false); + if (!isStoppingOrStopped()) { + LOG.warn("Unable to schedule reconnect after server disconnect", e); + } + } + }; + } + private void reconnectAfterDisconnect() { + final BayeuxClient disconnectedClient = client; try { - if (!channelToConsumers.isEmpty()) { - channelsToSubscribe.clear(); - channelsToSubscribe.addAll(channelToConsumers.keySet()); - LOG.info("Channels to resubscribe after reconnect: {}", channelsToSubscribe); + if (disconnectedClient == null || isStoppingOrStopped()) { + return; } - long backoff = 0; - while (!isStoppingOrStopped()) { - handshakeError = null; - handshakeException = null; - connectError = null; - connectException = null; - - client.handshake(); - final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC, SECONDS); - if (client.waitFor(waitMs, BayeuxClient.State.CONNECTED)) { - LOG.info("Reconnect successful"); - handshakeBackoff.set(0); - return; + final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC, SECONDS); + if (!disconnectedClient.waitFor(waitMs, BayeuxClient.State.DISCONNECTED)) { + if (!isStoppingOrStopped()) { + LOG.warn("Timed out waiting for the Streaming API client to disconnect"); } + return; + } + if (isStoppingOrStopped() || client != disconnectedClient) { + return; + } - LOG.warn("Reconnect attempt failed, retrying..."); - backoff = backoff + backoffIncrement; - if (backoff > maxBackoff) { - backoff = maxBackoff; - } - try { - LOG.debug("Pausing for {} msecs before reconnect attempt", backoff); - Thread.sleep(backoff); - } catch (InterruptedException e) { - LOG.warn("Aborting reconnect on interrupt!", e); - Thread.currentThread().interrupt(); - return; - } + LOG.info("Server disconnect received, reconnecting to Streaming API"); + disconnectedClient.handshake(); + } catch (IllegalStateException e) { + if (!isStoppingOrStopped()) { + LOG.debug("Streaming API reconnect was superseded by another state transition", e); } } finally { reconnecting.set(false); diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java index b56f2952eb00e..2e73104a4a68b 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java +++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelperManualIT.java @@ -275,6 +275,7 @@ void shouldResubscribeOnConnectionFailures() throws InterruptedException { void shouldResubscribeOnDisconnectMessage() { var consumer = createConsumer("Opportunity"); subscription.subscribe(consumer); + var resubscribeAttempts = new AtomicInteger(); messages.add(""" [ @@ -306,29 +307,31 @@ void shouldResubscribeOnDisconnectMessage() { (MessageListener) (clientSessionChannel, message) -> { var channel = (String) message.get("subscription"); if (channel != null && channel.contains("Opportunity")) { - messages.add(""" - [ - { - "data": { - "event": { - "createdDate": "2020-12-11T13:44:57.891Z", - "replayId": 2, - "type": "created" + if (resubscribeAttempts.incrementAndGet() == 1) { + messages.add(""" + [ + { + "data": { + "event": { + "createdDate": "2020-12-11T13:44:57.891Z", + "replayId": 2, + "type": "created" + }, + "sobject": { + "Id": "0061n00002XWMgVAAX", + "Name": "shouldResubscribeOnDisconnectMessage 2" + } + }, + "channel": "/topic/Opportunity" }, - "sobject": { - "Id": "0061n00002XWMgVAAX", - "Name": "shouldResubscribeOnDisconnectMessage 2" + { + "clientId": "5ra4927ikfky6cb12juthkpofeu8", + "channel": "/meta/connect", + "id": "$id", + "successful": true } - }, - "channel": "/topic/Opportunity" - }, - { - "clientId": "5ra4927ikfky6cb12juthkpofeu8", - "channel": "/meta/connect", - "id": "$id", - "successful": true - } - ]"""); + ]"""); + } } }); messages.add(""" @@ -341,6 +344,7 @@ void shouldResubscribeOnDisconnectMessage() { verify(consumer, timeout(20000)).processMessage(any(ClientSessionChannel.class), messageWithName("shouldResubscribeOnDisconnectMessage 2")); + await().during(1, SECONDS).atMost(2, SECONDS).until(() -> resubscribeAttempts.get() == 1); verify(consumer, atLeastOnce()).getEndpoint(); verify(consumer, atLeastOnce()).getTopicName(); From 649f04f191137f0214c60950a42eddc171981c8b Mon Sep 17 00:00:00 2001 From: Bartosz Popiela Date: Mon, 31 Aug 2026 17:57:20 +0200 Subject: [PATCH 3/3] CAMEL-24569: Improve SalesforceComponent to wait for the disconnect state in the http worker thread instead of a single-threaded executor service not to block the latter. Also, disconnect timeout has been reduced to 10 seconds --- .../internal/streaming/SubscriptionHelper.java | 14 +++++--------- 1 file changed, 5 insertions(+), 9 deletions(-) diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java index f8c3a325ecded..1e842fbe80aa0 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java @@ -82,6 +82,7 @@ public class SubscriptionHelper extends ServiceSupport { private static final Logger LOG = LoggerFactory.getLogger(SubscriptionHelper.class); private static final int HANDSHAKE_TIMEOUT_SEC = 120; + private static final int DISCONNECT_TIMEOUT_SEC = 10; private static final String FAILURE_FIELD = "failure"; private static final String EXCEPTION_FIELD = "exception"; @@ -261,14 +262,8 @@ private MessageListener createDisconnectListener() { return; } - final ScheduledExecutorService executor = taskExecutor; - if (executor == null) { - reconnecting.set(false); - return; - } - try { - executor.execute(this::reconnectAfterDisconnect); + component.getHttpClient().getWorkerPool().execute(this::reconnectAfterDisconnect); } catch (RejectedExecutionException e) { reconnecting.set(false); if (!isStoppingOrStopped()) { @@ -285,10 +280,11 @@ private void reconnectAfterDisconnect() { return; } - final long waitMs = MILLISECONDS.convert(HANDSHAKE_TIMEOUT_SEC, SECONDS); + final long waitMs = MILLISECONDS.convert(DISCONNECT_TIMEOUT_SEC, SECONDS); if (!disconnectedClient.waitFor(waitMs, BayeuxClient.State.DISCONNECTED)) { if (!isStoppingOrStopped()) { - LOG.warn("Timed out waiting for the Streaming API client to disconnect"); + LOG.warn("Timed out after {} seconds waiting for the Streaming API client to disconnect", + DISCONNECT_TIMEOUT_SEC); } return; }