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..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 @@ -30,8 +30,10 @@ 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; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -67,6 +69,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; @@ -79,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"; @@ -92,9 +96,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; @@ -107,6 +111,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 +127,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 +137,170 @@ public SubscriptionHelper(final SalesforceComponent component) { } private MessageListener createHandshakeListener() { - return (channel, message) -> component.getHttpClient().getWorkerPool().execute(() -> { - LOG.debug("[CHANNEL:META_HANDSHAKE]: {}", message); + 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); + } + }); + } - 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(); + 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; + + 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); + } + } + } + } + }); + } + + private MessageListener createDisconnectListener() { + return (channel, message) -> { + LOG.debug("[CHANNEL:META_DISCONNECT]: {}", message); + + if (isStoppingOrStopped()) { + LOG.debug("Ignoring disconnect message while stopping"); + return; + } + if (!reconnecting.compareAndSet(false, true)) { + LOG.debug("Reconnect already in progress"); + return; + } + + try { + component.getHttpClient().getWorkerPool().execute(this::reconnectAfterDisconnect); + } catch (RejectedExecutionException e) { + reconnecting.set(false); + if (!isStoppingOrStopped()) { + LOG.warn("Unable to schedule reconnect after server disconnect", e); } - 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; + private void reconnectAfterDisconnect() { + final BayeuxClient disconnectedClient = client; + try { + if (disconnectedClient == null || isStoppingOrStopped()) { + return; + } - 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); - } - } + final long waitMs = MILLISECONDS.convert(DISCONNECT_TIMEOUT_SEC, SECONDS); + if (!disconnectedClient.waitFor(waitMs, BayeuxClient.State.DISCONNECTED)) { + if (!isStoppingOrStopped()) { + LOG.warn("Timed out after {} seconds waiting for the Streaming API client to disconnect", + DISCONNECT_TIMEOUT_SEC); } + return; } - }); + if (isStoppingOrStopped() || client != disconnectedClient) { + 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); + } } private MessageListener createSubscriptionListener() { @@ -272,8 +342,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 +379,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 +420,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 +431,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 +484,7 @@ protected void doStop() throws Exception { } closeChannel(META_CONNECT); + closeChannel(META_DISCONNECT); closeChannel(META_SUBSCRIBE); closeChannel(META_HANDSHAKE); @@ -434,7 +507,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 +536,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 +584,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 +647,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 +657,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 +731,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..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 @@ -271,6 +271,86 @@ void shouldResubscribeOnConnectionFailures() throws InterruptedException { verifyNoMoreInteractions(consumer); } + @Test + void shouldResubscribeOnDisconnectMessage() { + var consumer = createConsumer("Opportunity"); + subscription.subscribe(consumer); + var resubscribeAttempts = new AtomicInteger(); + + 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")) { + 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" + }, + { + "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")); + await().during(1, SECONDS).atMost(2, SECONDS).until(() -> resubscribeAttempts.get() == 1); + + verify(consumer, atLeastOnce()).getEndpoint(); + verify(consumer, atLeastOnce()).getTopicName(); + verifyNoMoreInteractions(consumer); + } + @Test void shouldResubscribeOnSubscriptionFailure() { var consumer = createConsumer("Contact");