Skip to content

camel-salesforce: Improve error resilience in subscription failure handling #22840

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davsclaus merged 2 commits into from
Apr 30, 2026
+16 −2
Merged

camel-salesforce: Improve error resilience in subscription failure handling #22840

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d416819
camel-salesforce: Improve error resilience in subscription failure ha…
shaipan Apr 29, 2026
1ebb3b9
Address review: remove redundant ORG_LIMIT_ERROR, drop silent retry o…
shaipan Apr 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 16 additions & 2 deletions ...ain/java/org/apache/camel/component/salesforce/internal/streaming/SubscriptionHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Stream;

Expand All @@ -50,6 +51,7 @@
import org.cometd.client.BayeuxClient.State;
import org.cometd.client.http.jetty.JettyHttpClientTransport;
import org.cometd.client.transport.ClientTransport;
import org.cometd.common.TransportException;
import org.eclipse.jetty.client.Request;
import org.eclipse.jetty.http.HttpCookie;
import org.eclipse.jetty.http.HttpCookieStore;
Expand Down Expand Up @@ -82,6 +84,9 @@ public class SubscriptionHelper extends ServiceSupport {
private static final String SERVER_TOO_BUSY_ERROR = "503::";
private static final String AUTHENTICATION_INVALID = "401::Authentication invalid";
private static final String INVALID_REPLAY_ID_PATTERN = "400::The replayId \\{.*} you provided was invalid.*";
private static final String CHANNEL_INVALID_PATTERN = "400::The channel.*";
private static final String DENIED_BY_SEC_POLICY = "403:denied_by_security_policy";
private static final String AUTHORIZATION_ERROR = "403::";

BayeuxClient client;

Expand Down Expand Up @@ -219,7 +224,8 @@ private void subscriptionFailed(StreamingApiConsumer firstConsumer, Message mess
boolean abort = true;

LOG.warn(msg);
if (isTemporaryError(message)) {
if (isTemporaryError(message) || error.equals(AUTHENTICATION_INVALID) || error.startsWith(DENIED_BY_SEC_POLICY)
|| error.startsWith(AUTHORIZATION_ERROR)) {

// retry after delay
final long backoff = handshakeBackoff.getAndAdd(backoffIncrement);
Expand Down Expand Up @@ -255,6 +261,9 @@ private void subscriptionFailed(StreamingApiConsumer firstConsumer, Message mess
for (var consumer : consumers) {
subscribe(consumer);
}
} else if (error.matches(CHANNEL_INVALID_PATTERN)) {
LOG.warn("Channel invalid for channel {}, removing from subscription list", channelName);
channelsToSubscribe.remove(channelName);
Comment thread
davsclaus marked this conversation as resolved.
}

if (abort && client != null) {
Expand Down Expand Up @@ -465,7 +474,12 @@ ReplayExtension getReplayExtension() {

private static boolean isTemporaryError(Message message) {
String failureReason = getFailureReason(message);
return failureReason != null && failureReason.startsWith(SERVER_TOO_BUSY_ERROR);
if (failureReason != null && failureReason.startsWith(SERVER_TOO_BUSY_ERROR)) {
return true;
}
Exception exception = getFailure(message);
return exception instanceof IOException || exception instanceof TransportException
|| exception instanceof TimeoutException;
}

private static String getFailureReason(Message message) {
Expand Down
Loading