This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24161 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 9ce5ba2ed7edcdebdbbcd38e899ae0c7df9cc949 Author: Claus Ibsen <[email protected]> AuthorDate: Fri Jul 17 16:12:49 2026 +0200 CAMEL-24161: camel-salesforce - fix medium-severity bugs from code review Fixes 11 bugs found during a deep code review of camel-salesforce: - SalesforceLoginConfig: mask refreshToken in toString() to prevent credential leak to debug logs (other secrets were already masked) - SalesforceSecurityHandler: check body.rewind() return value to detect unreplayable multipart content on 401 retry - AbstractClientBase: detect Jetty failure chunks in replay-buffering loop to prevent silently truncated requests - DefaultRawClient: use put() instead of add() for auth headers to prevent duplicate X-SFDC-Session/Authorization on retry - SalesforceSecurityHandler: coordinate re-logins via session.login() on worker pool so concurrent 401s share one login request instead of each firing its own - SubscriptionHelper: use getOrDefault() to prevent NPE when channel is concurrently unsubscribed, and remove from channelsToSubscribe on unsubscribe to prevent stale entries - PubSubApiClient: invoke consumer exception handler on per-event decode failures instead of silently swallowing them - PubSubApiClient: stop infinite retry on corrupted replay ID when fallbackToLatestReplayId=false, and make replay state per-observer to prevent preset overwrite across consumers - JsonRestProcessor: clean up RESPONSE_CLASS_DEFERRED and RESPONSE_CLASS_PREFIX exchange properties in processResponse() to match processStreamResultResponse() cleanup - CompositeApiProcessor: preserve exception when Salesforce returns both error body and exception (raw, batch, and composite response) - AbstractRestProcessor: fix deleteSObjectWithId to restore DTO fields before completing exchange (matching all sibling operations), and null-check external-id field value to prevent NPE Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../salesforce/SalesforceLoginConfig.java | 2 +- .../internal/client/AbstractClientBase.java | 5 ++++ .../internal/client/DefaultRawClient.java | 4 +-- .../internal/client/PubSubApiClient.java | 25 ++++++++++-------- .../internal/client/SalesforceSecurityHandler.java | 30 ++++++++++++++-------- .../internal/processor/AbstractRestProcessor.java | 20 ++++++++++++--- .../internal/processor/CompositeApiProcessor.java | 9 +++++++ .../internal/processor/JsonRestProcessor.java | 4 ++- .../internal/streaming/SubscriptionHelper.java | 4 ++- .../camel/component/salesforce/PubSubApiTest.java | 15 +++++------ 10 files changed, 79 insertions(+), 39 deletions(-) diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceLoginConfig.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceLoginConfig.java index e1e68ab2cfc5..0f86c911dfbd 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceLoginConfig.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceLoginConfig.java @@ -270,7 +270,7 @@ public class SalesforceLoginConfig { public String toString() { return "SalesforceLoginConfig[" + "instanceUrl= '" + instanceUrl + "', loginUrl='" + loginUrl + '\'' + "," + "clientId='" + clientId + '\'' + ", clientSecret='********'" - + ", refreshToken='" + refreshToken + '\'' + ", userName='" + userName + '\'' + ", password=********'" + + ", refreshToken='********'" + ", userName='" + userName + '\'' + ", password=********'" + ", keystore=********', audience='" + jwtAudience + '\'' + "," + ", lazyLogin=" + lazyLogin + ']'; } diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/AbstractClientBase.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/AbstractClientBase.java index 546b6977385e..af142cf6773f 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/AbstractClientBase.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/AbstractClientBase.java @@ -190,6 +190,11 @@ public abstract class AbstractClientBase extends ServiceSupport final List<ByteBuffer> buffers = new ArrayList<>(); while (true) { Content.Chunk chunk = inputStreamRequestContent.read(); + if (Content.Chunk.isFailure(chunk)) { + chunk.release(); + throw new RuntimeException( + new IOException("Failed to buffer request content", chunk.getFailure())); + } if (chunk.isLast()) { break; } else { diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRawClient.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRawClient.java index 610b8c320e63..27f59e4a9cff 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRawClient.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultRawClient.java @@ -48,8 +48,8 @@ public class DefaultRawClient extends AbstractClientBase implements RawClient { @Override protected void setAccessToken(Request request) { // replace old token - request.headers(h -> h.add(BULK_TOKEN_HEADER, accessToken)); - request.headers(h -> h.add(REST_TOKEN_HEADER, TOKEN_PREFIX + accessToken)); + request.headers(h -> h.put(BULK_TOKEN_HEADER, accessToken)); + request.headers(h -> h.put(REST_TOKEN_HEADER, TOKEN_PREFIX + accessToken)); } @Override diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java index 461e24e3bb9b..edff9c8c854a 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/PubSubApiClient.java @@ -101,10 +101,6 @@ public class PubSubApiClient extends ServiceSupport { private ManagedChannel channel; private boolean usePlainTextConnection = false; - private ReplayPreset initialReplayPreset; - private String initialReplayId; - private boolean fallbackToLatestReplayId; - public PubSubApiClient(SalesforceSession session, SalesforceLoginConfig loginConfig, String pubSubHost, int pubSubPort, long backoffIncrement, long maxBackoff, boolean allowUseProxyServer) { this.session = session; @@ -151,9 +147,6 @@ public class PubSubApiClient extends ServiceSupport { public void subscribe( PubSubApiConsumer consumer, ReplayPreset replayPreset, String initialReplayId, boolean fallbackToLatestReplayId) { LOG.debug("Starting subscribe {}", consumer.getTopic()); - this.initialReplayPreset = replayPreset; - this.initialReplayId = initialReplayId; - this.fallbackToLatestReplayId = fallbackToLatestReplayId; if (replayPreset == ReplayPreset.CUSTOM && initialReplayId == null) { throw new RuntimeException("initialReplayId is required for ReplayPreset.CUSTOM"); } @@ -164,7 +157,8 @@ public class PubSubApiClient extends ServiceSupport { replayId = base64DecodeToByteString(initialReplayId); } LOG.info("Subscribing to topic: {}.", topic); - final FetchResponseObserver responseObserver = new FetchResponseObserver(consumer); + final FetchResponseObserver responseObserver + = new FetchResponseObserver(consumer, replayPreset, initialReplayId, fallbackToLatestReplayId); StreamObserver<FetchRequest> serverStream = asyncStub.subscribe(responseObserver); LOG.info("Subscribe successful."); responseObserver.setServerStream(serverStream); @@ -292,13 +286,20 @@ public class PubSubApiClient extends ServiceSupport { private final PubSubApiConsumer consumer; private final Map<String, Class<?>> eventClassMap; private final Class<?> pojoClass; + private final String initialReplayId; + private final boolean fallbackToLatestReplayId; + private ReplayPreset initialReplayPreset; private String replayId; private StreamObserver<FetchRequest> serverStream; - public FetchResponseObserver(PubSubApiConsumer consumer) { + public FetchResponseObserver(PubSubApiConsumer consumer, ReplayPreset initialReplayPreset, + String initialReplayId, boolean fallbackToLatestReplayId) { this.consumer = consumer; this.eventClassMap = consumer.getEventClassMap(); this.pojoClass = consumer.getPojoClass(); + this.initialReplayPreset = initialReplayPreset; + this.initialReplayId = initialReplayId; + this.fallbackToLatestReplayId = fallbackToLatestReplayId; } @Override @@ -315,7 +316,9 @@ public class PubSubApiClient extends ServiceSupport { try { processEvent(fetchResponse.getRpcId(), ce); } catch (Exception e) { - LOG.error(e.toString(), e); + LOG.error("Failed to process event on topic {}: {}", topic, e.getMessage(), e); + consumer.getExceptionHandler().handleException( + "Failed to process Pub/Sub event on topic " + topic, e); } } replayId = base64EncodeByteString(fetchResponse.getLatestReplayId()); @@ -366,6 +369,8 @@ public class PubSubApiClient extends ServiceSupport { consumer.getExceptionHandler().handleException(new InvalidReplayIdException( "Corrupt replay id: " + currReplayId, currReplayId)); + replayId = null; + return; } } default -> LOG.error("unexpected errorCode: {}", errorCode); diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java index e2d2d533633c..054168e21e0a 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/SalesforceSecurityHandler.java @@ -186,9 +186,8 @@ public class SalesforceSecurityHandler implements ProtocolHandler { // REST token expiry LOG.warn("Retrying on Salesforce authentication error [{}]: [{}]", status, reason); - // remember original request and send a relogin request in - // current conversation - retryLogin(request, retries); + // coordinated re-login and retry via session.login() on worker pool + retryLogin(request, retries, client); } else if (status < HttpStatus.OK_200 || status >= HttpStatus.MULTIPLE_CHOICES_300) { @@ -202,7 +201,7 @@ public class SalesforceSecurityHandler implements ProtocolHandler { // retry Bulk API call LOG.warn("Retrying on Bulk API Salesforce authentication error [{}]: [{}]", status, reason); - retryLogin(request, retries); + retryLogin(request, retries, client); } else { @@ -236,13 +235,21 @@ public class SalesforceSecurityHandler implements ProtocolHandler { && "InvalidSessionId".equals(e.getErrors().get(0).getErrorCode()); } - private void retryLogin(HttpRequest request, Integer retries) { - + private void retryLogin(HttpRequest request, Integer retries, AbstractClientBase client) { final HttpConversation conversation = request.getConversation(); - // remember the original request to resend - conversation.setAttribute(AUTHENTICATION_REQUEST_ATTRIBUTE, request); - retryRequest((HttpRequest) session.getLoginRequest(conversation), null, retries, conversation, false); + // use session.login() on worker pool for single-flight coordination: + // concurrent 401s share one login instead of each firing its own + httpClient.getWorkerPool().execute(() -> { + try { + session.login(session.getAccessToken()); + } catch (SalesforceException e) { + LOG.error("Login failed during authentication retry", e); + forwardFailureComplete(request, null, null, e); + return; + } + retryRequest(request, client, retries, conversation, true); + }); } private void retryRequest( @@ -255,8 +262,9 @@ public class SalesforceSecurityHandler implements ProtocolHandler { if (copy) { newRequest = httpClient.copyRequest(request, request.getURI()); final Request.Content body = newRequest.getBody(); - if (body != null) { - body.rewind(); + if (body != null && !body.rewind()) { + LOG.warn("Request body cannot be replayed for authentication retry (content type: {})", + body.getContentType()); } newRequest.method(request.getMethod()); newRequest.headers(headers -> { diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java index 3158c2772ef5..aff931270648 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/AbstractRestProcessor.java @@ -479,7 +479,11 @@ public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class); if (sObjectBase != null) { oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName); - sObjectExtIdValue = oldValue.toString(); + if (oldValue != null) { + sObjectExtIdValue = oldValue.toString(); + } else { + sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL); + } } else { sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL); } @@ -508,7 +512,11 @@ public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor final AbstractSObjectBase sObjectBase = exchange.getIn().getBody(AbstractSObjectBase.class); if (sObjectBase != null) { oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName); - sObjectExtIdValue = oldValue.toString(); + if (oldValue != null) { + sObjectExtIdValue = oldValue.toString(); + } else { + sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, IGNORE_BODY, NOT_OPTIONAL); + } // clear base object fields, which cannot be updated sObjectBase.clearBaseFields(); } else { @@ -536,7 +544,11 @@ public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor String sObjectExtIdValue; if (sObjectBase != null) { oldValue = getAndClearPropertyValue(sObjectBase, sObjectExtIdName); - sObjectExtIdValue = oldValue.toString(); + if (oldValue != null) { + sObjectExtIdValue = oldValue.toString(); + } else { + sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL); + } } else { sObjectExtIdValue = getParameter(SOBJECT_EXT_ID_VALUE, exchange, USE_BODY, NOT_OPTIONAL); } @@ -546,8 +558,8 @@ public abstract class AbstractRestProcessor extends AbstractSalesforceProcessor new RestClient.ResponseCallback() { @Override public void onResponse(InputStream response, Map<String, String> headers, SalesforceException exception) { - processResponse(exchange, response, headers, exception, callback); restoreFields(exchange, sObjectBase, null, sObjectExtIdName, finalOldValue); + processResponse(exchange, response, headers, exception, callback); } }); } diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/CompositeApiProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/CompositeApiProcessor.java index eb28d8722df9..584448170db6 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/CompositeApiProcessor.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/CompositeApiProcessor.java @@ -115,6 +115,9 @@ public final class CompositeApiProcessor extends AbstractSalesforceProcessor { out.copyFromWithNewBody(in, response); out.getHeaders().putAll(headers); + if (exception != null) { + exchange.setException(exception); + } } } finally { // notify callback that exchange is done @@ -136,6 +139,9 @@ public final class CompositeApiProcessor extends AbstractSalesforceProcessor { out.copyFromWithNewBody(in, response); out.getHeaders().putAll(headers); + if (exception != null) { + exchange.setException(exception); + } } } finally { // notify callback that exchange is done @@ -157,6 +163,9 @@ public final class CompositeApiProcessor extends AbstractSalesforceProcessor { out.copyFromWithNewBody(in, response); out.getHeaders().putAll(headers); + if (exception != null) { + exchange.setException(exception); + } } } finally { // notify callback that exchange is done diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java index a06da2bfdcf9..2542ee8d5e1a 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java +++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/processor/JsonRestProcessor.java @@ -219,8 +219,10 @@ public class JsonRestProcessor extends AbstractRestProcessor { String msg = "Error parsing JSON response: " + e.getMessage(); exchange.setException(new SalesforceException(msg, e)); } finally { - // cleanup temporary exchange headers + // cleanup temporary exchange properties exchange.removeProperty(RESPONSE_CLASS); + exchange.removeProperty(RESPONSE_CLASS_DEFERRED); + exchange.removeProperty(RESPONSE_CLASS_PREFIX); exchange.removeProperty(RESPONSE_TYPE); // consume response entity 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 ab77195d963b..3e8cef329cde 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 @@ -203,7 +203,8 @@ public class SubscriptionHelper extends ServiceSupport { if (toSubscribe != null) { LOG.info("Subscribing to channels: {}", toSubscribe); for (var channelName : toSubscribe) { - for (var consumer : channelToConsumers.get(channelName)) { + var consumers = channelToConsumers.getOrDefault(channelName, emptySet()); + for (var consumer : consumers) { subscribe(consumer); } } @@ -592,6 +593,7 @@ public class SubscriptionHelper extends ServiceSupport { consumers.remove(consumer); if (consumers.isEmpty()) { channelToConsumers.remove(channelName); + channelsToSubscribe.remove(channelName); } } final ClientSessionChannel.MessageListener listener = consumerToListener.remove(consumer); diff --git a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/PubSubApiTest.java b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/PubSubApiTest.java index 9ca33f2d46d9..cb191c89c303 100644 --- a/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/PubSubApiTest.java +++ b/components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/PubSubApiTest.java @@ -204,17 +204,14 @@ public class PubSubApiTest { final String replayId = encodeReplayId("123"); client.subscribe(consumer, ReplayPreset.CUSTOM, replayId, false); - Thread.sleep(1000); - - InOrder inOrder = Mockito.inOrder(client); - inOrder.verify(client, timeout(5000).times(3)).subscribe(consumer, ReplayPreset.CUSTOM, replayId, false); - inOrder.verify(client, never()).subscribe(consumer, ReplayPreset.LATEST, null, false); + // with the fix, corrupted replay ID with fallback disabled stops after one attempt + // (no infinite retry with the same corrupt ID) + verify(client, timeout(5000).times(1)).subscribe(consumer, ReplayPreset.CUSTOM, replayId, false); + verify(client, never()).subscribe(consumer, ReplayPreset.LATEST, null, false); ArgumentCaptor<InvalidReplayIdException> captor = ArgumentCaptor.forClass(InvalidReplayIdException.class); - verify(exceptionHandler, timeout(5000).times(3)).handleException(captor.capture()); - for (InvalidReplayIdException exception : captor.getAllValues()) { - Assertions.assertEquals(replayId, exception.getReplayId()); - } + verify(exceptionHandler, timeout(5000).times(1)).handleException(captor.capture()); + Assertions.assertEquals(replayId, captor.getValue().getReplayId()); } private String encodeReplayId(String replayId) {
