You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2026/07/17 14:13:34 UTC
(camel) branch fix/CAMEL-24161 created (now 9ce5ba2ed7ed)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24161
in repository https://gitbox.apache.org/repos/asf/camel.git
at 9ce5ba2ed7ed CAMEL-24161: camel-salesforce - fix medium-severity bugs from code review
This branch includes the following new commits:
new 9ce5ba2ed7ed CAMEL-24161: camel-salesforce - fix medium-severity bugs from code review
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(camel) 01/01: CAMEL-24161: camel-salesforce - fix medium-severity bugs from code review
Posted by da...@apache.org.
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 <cl...@gmail.com>
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 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../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) {