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/18 07:59:59 UTC
(camel) branch fix/CAMEL-24160 updated (4d3f6d8f78c0 -> c66a2cdf8fe5)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a change to branch fix/CAMEL-24160
in repository https://gitbox.apache.org/repos/asf/camel.git
discard 4d3f6d8f78c0 CAMEL-24160: camel-salesforce - fix high-severity bugs from code review
add 13938eea6048 Upgrade Netty from 4.2.15 to 4.2.16
add 3d0619fd14a1 CAMEL-24155: camel-salesforce - Fix streamQueryResult iterator hang when queryMore fails
add f8a59b9cd954 CAMEL-24175: Remove incorrect option counts from documentation page headers
add 1367ce157a74 CAMEL-24163: Guard Jackson-only options in JsonDataFormatReifier
add 93950e50adf0 CAMEL-24141: Aggregate EIP - guard background tasks against exceptions
add ecbcbb004d1b (chores): fix SonarCloud S1845 naming case issues (#24853)
add 56cd8dfdbf4b CAMEL-24154: camel-aws2-ddb - Fix DDB Streams consumer silent data loss on resharding
add 3aaa8037a18c CAMEL-24153: camel-aws2-s3 - Fix streaming upload truncation and orphaned multipart uploads
add 701eb46e477e CAMEL-24161: camel-salesforce - fix medium-severity bugs from code review
add 73fe3089db15 (chores): fix SonarCloud S1845 naming convention issues in salesforce and crypto-pgp (#24860)
add fd94df444f86 CAMEL-24194: camel-aws-config - removeConformancePack reads the correct header (#24854)
add 4f887ea6d264 CAMEL-24157: camel-aws2 - Fix medium-severity findings from July 2026 review (#24864)
add edf03f86d84e CAMEL-24159: camel-azure-servicebus - fix medium-severity bugs from code review (#24865)
add cd8268220566 CAMEL-24159: camel-azure-eventhubs - fix medium-severity bugs from code review (#24867)
add f05c571a44b8 CAMEL-24159: camel-azure-storage-blob - Fix medium-severity bugs from code review (#24869)
add dd1d7cafed24 CAMEL-24196: camel-aws2-sqs - Fix QueueDoesNotExist swallowed, thread pool leak, and batch chunking
add d8be2daaa324 CAMEL-24195: camel-aws2-s3 - Fix StreamUploadProducer race condition and timeout task exception handling
add f6ae9ecad39b CAMEL-24159: camel-azure-storage-datalake - Fix medium-severity bugs from code review
add 054611d9be95 CAMEL-24150, CAMEL-24151: Saga EIP fixes — shared service lifecycle, code review findings
add eceeeaaf9d9c CAMEL-24006: Fix flaky JpaPollingConsumerLockEntityTest under concurrency
add 27d0663fc626 CAMEL-23465: camel-aws-bedrock - align agent runtime modelId enum with BedrockConfiguration (#24851)
add 48a28da60ef1 CAMEL-24197: camel-aws2-kinesis - Fix fixed-shardId DescribeStream on every poll and batch partition key (#24878)
add 21b8be5aa494 CAMEL-24159: camel-azure-storage-queue - Fix medium-severity bugs from code review (#24870)
add 941c2fe2313c CAMEL-23955: Fix conversation memory user turns and history reset
add 6ab243b20b31 CAMEL-23962: Preserve all MCP inputSchema keywords in tool conversion
add 5fbcf2365a8a CAMEL-23956: Handle malformed tool arguments in agentic loop
new c66a2cdf8fe5 CAMEL-24160: camel-salesforce - fix high-severity bugs from code review
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (4d3f6d8f78c0)
\
N -- N -- N refs/heads/fix/CAMEL-24160 (c66a2cdf8fe5)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
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.
Summary of changes:
.../components/aws-bedrock-agent-runtime.json | 4 +-
.../camel/catalog/components/aws2-kinesis.json | 5 +-
.../apache/camel/catalog/components/aws2-s3.json | 2 +-
.../apache/camel/catalog/dataformats/fastjson.json | 5 +-
.../camel/component/openai/McpToolConverter.java | 15 +-
.../camel/component/openai/OpenAIProducer.java | 30 ++-
.../openai/OpenAIToolExecutionProducer.java | 6 +-
.../component/openai/McpToolConverterTest.java | 45 +++++
...penAIAgenticLoopMalformedToolArgumentsTest.java | 163 ++++++++++++++++
.../openai/OpenAIConversationMemoryResetTest.java | 94 +++++++++
.../OpenAIConversationMemoryUserMessageTest.java | 95 +++++++++
.../agentruntime/aws-bedrock-agent-runtime.json | 4 +-
.../BedrockAgentRuntimeConfiguration.java | 2 +-
components/camel-aws/camel-aws-config/pom.xml | 6 +
.../component/aws/config/AWSConfigProducer.java | 3 +-
.../config/AWSConfigRemoveConformancePackTest.java | 66 +++++++
.../camel/component/aws2/ddb/Ddb2Endpoint.java | 6 +-
.../camel/component/aws2/ddb/ScanCommand.java | 3 +-
.../ddb/transform/Ddb2JsonDataTypeTransformer.java | 4 +-
.../aws2/ddbstream/ShardIteratorHandler.java | 48 ++++-
.../aws2/ddbstream/AmazonDDBStreamsClientMock.java | 9 +
.../aws2/ddbstream/ShardIteratorHandlerTest.java | 105 ++++++++++
.../camel/component/aws2/kinesis/aws2-kinesis.json | 5 +-
.../component/aws2/kinesis/Kinesis2Constants.java | 8 +
.../component/aws2/kinesis/Kinesis2Consumer.java | 108 +++++-----
.../component/aws2/kinesis/Kinesis2Producer.java | 26 ++-
.../aws2/kinesis/Kinesis2ProducerTest.java | 166 ++++++++++++++++
.../KinesisConsumerClosedShardWithSilentTest.java | 17 +-
.../apache/camel/component/aws2/s3/aws2-s3.json | 2 +-
.../camel/component/aws2/s3/AWS2S3Consumer.java | 38 +++-
.../camel/component/aws2/s3/AWS2S3Endpoint.java | 13 +-
.../camel/component/aws2/s3/AWS2S3Producer.java | 47 ++++-
.../aws2/s3/stream/AWS2S3StreamUploadProducer.java | 45 ++++-
.../s3/stream/AWS2S3StreamUploadMultipartTest.java | 217 +++++++++++++++++++++
.../camel/component/aws2/sns/Sns2Endpoint.java | 9 +-
.../aws2/sns/Sns2ProducerHealthCheck.java | 10 +-
.../camel/component/aws2/sqs/Sqs2Consumer.java | 26 +++
.../camel/component/aws2/sqs/Sqs2Producer.java | 65 +++---
.../eventhubs/EventHubsCheckpointUpdaterTask.java | 4 +-
.../azure/eventhubs/EventHubsComponent.java | 8 +-
.../azure/eventhubs/EventHubsConsumer.java | 4 +-
.../azure/eventhubs/EventHubsProducer.java | 8 +-
.../azure/servicebus/ServiceBusComponent.java | 8 +-
.../azure/servicebus/ServiceBusConsumer.java | 24 ++-
.../azure/servicebus/ServiceBusProducer.java | 19 +-
.../ServicebusCloudEventDataTypeTransformer.java | 2 +-
.../azure/servicebus/ServiceBusConsumerTest.java | 29 ++-
.../storage/blob/BlobCommonRequestOptions.java | 5 +-
.../azure/storage/blob/BlobComponent.java | 4 +-
.../blob/operations/BlobChangeFeedOperations.java | 2 +-
.../storage/blob/operations/BlobOperations.java | 53 ++++-
.../azure/storage/datalake/DataLakeComponent.java | 14 +-
.../operations/DataLakeFileOperations.java | 9 +-
.../azure/storage/queue/QueueComponent.java | 4 +-
.../storage/queue/client/QueueClientWrapper.java | 4 +
.../storage/queue/operations/QueueOperations.java | 4 +-
.../crypto/DefaultPGPSecretKeyAccessor.java | 2 +-
.../camel/converter/crypto/PGPDataFormatUtil.java | 4 +-
.../fastjson/FastjsonDataFormatConfigurer.java | 7 +
.../apache/camel/component/fastjson/fastjson.json | 5 +-
.../component/fastjson/FastjsonDataFormat.java | 2 +-
.../jpa/JpaPollingConsumerLockEntityTest.java | 22 +--
.../org/apache/camel/service/lra/LRAClient.java | 21 +-
.../camel/service/lra/LRASagaCoordinator.java | 4 +-
.../apache/camel/service/lra/LRASagaService.java | 24 +--
.../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/processor/PubSubApiProcessor.java | 5 +-
.../internal/processor/QueryResultIterator.java | 21 +-
.../internal/streaming/SubscriptionHelper.java | 4 +-
.../camel/component/salesforce/PubSubApiTest.java | 15 +-
.../processor/QueryResultIteratorTest.java | 190 ++++++++++++++++++
.../src/main/docs/modules/eips/pages/saga-eip.adoc | 4 +-
.../eips/partials/eip-exchangeProperties.adoc | 2 +-
.../docs/modules/eips/partials/eip-options.adoc | 9 +-
.../processor/aggregate/AggregateProcessor.java | 67 ++++---
.../processor/saga/MandatorySagaProcessor.java | 5 +-
.../camel/processor/saga/NeverSagaProcessor.java | 5 +-
.../processor/saga/NotSupportedSagaProcessor.java | 5 +-
.../processor/saga/RequiredSagaProcessor.java | 5 +-
.../processor/saga/RequiresNewSagaProcessor.java | 5 +-
.../apache/camel/processor/saga/SagaProcessor.java | 24 +--
.../camel/processor/saga/SagaProcessorBuilder.java | 20 +-
.../processor/saga/SupportsSagaProcessor.java | 5 +-
.../java/org/apache/camel/reifier/SagaReifier.java | 12 +-
.../reifier/dataformat/JsonDataFormatReifier.java | 35 ++--
.../processor/SagaSharedServiceRouteStopTest.java | 86 ++++++++
.../AggregateIntervalTaskExceptionTest.java | 112 +++++++++++
.../dataformat/JsonDataFormatReifierTest.java | 96 +++++++++
.../apache/camel/saga/InMemorySagaCoordinator.java | 34 +++-
.../ROOT/partials/component-endpoint-headers.adoc | 2 +-
.../ROOT/partials/component-endpoint-options.adoc | 8 +-
.../dataformats/partials/dataformat-options.adoc | 8 +-
.../languages/partials/language-options.adoc | 8 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 32 +++
.../endpoint/dsl/AWS2S3EndpointBuilderFactory.java | 12 +-
.../dsl/Kinesis2EndpointBuilderFactory.java | 16 ++
.../java/org/apache/camel/main/KameletMain.java | 2 +-
.../apache/camel/main/download/SagaDownloader.java | 14 +-
parent/pom.xml | 2 +-
106 files changed, 2345 insertions(+), 442 deletions(-)
create mode 100644 components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIAgenticLoopMalformedToolArgumentsTest.java
create mode 100644 components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIConversationMemoryResetTest.java
create mode 100644 components/camel-ai/camel-openai/src/test/java/org/apache/camel/component/openai/OpenAIConversationMemoryUserMessageTest.java
create mode 100644 components/camel-aws/camel-aws-config/src/test/java/org/apache/camel/component/aws/config/AWSConfigRemoveConformancePackTest.java
create mode 100644 components/camel-aws/camel-aws2-kinesis/src/test/java/org/apache/camel/component/aws2/kinesis/Kinesis2ProducerTest.java
create mode 100644 components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/stream/AWS2S3StreamUploadMultipartTest.java
create mode 100644 components/camel-salesforce/camel-salesforce-component/src/test/java/org/apache/camel/component/salesforce/internal/processor/QueryResultIteratorTest.java
create mode 100644 core/camel-core/src/test/java/org/apache/camel/processor/SagaSharedServiceRouteStopTest.java
create mode 100644 core/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateIntervalTaskExceptionTest.java
create mode 100644 core/camel-core/src/test/java/org/apache/camel/reifier/dataformat/JsonDataFormatReifierTest.java
(camel) 01/01: CAMEL-24160: camel-salesforce - fix high-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-24160
in repository https://gitbox.apache.org/repos/asf/camel.git
commit c66a2cdf8fe5570e86490f16555770e74330b21e
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Fri Jul 17 15:49:19 2026 +0200
CAMEL-24160: camel-salesforce - fix high-severity bugs from code review
Fixes 9 bugs found during a deep code review of camel-salesforce:
- JsonRestProcessor: fix infinite loop on non-object JSON in deferred
SObject type detection, and fix RuntimeException escaping IOException
catch block causing silent data corruption
- DefaultBulkApiV2Client: add missing return after error callback in
changeJobState/changeQueryJobState to prevent double callback
- SalesforceEndpointConfig: add allOrNone to toValueMap() so the URI
option takes effect, and fix REPLAY_PRESET mapped to wrong field
- PubSubApiClient: skip binary trailer keys in onError to prevent
IllegalArgumentException killing reconnection, add stop/channel
checks in resubscribeOnError to prevent infinite loop on shutdown,
cap reconnect delay at maxBackoff, and restore thread interrupt
- SubscriptionHelper: add backoff delay to handshake retry to prevent
tight retry loop hammering the Salesforce API
- SalesforceSecurityHandler: guard against null client in 401 handling
to prevent NPE for CometD streaming requests
- SalesforceSession: make latch volatile and remove dead spin loop to
fix race condition in attemptLoginUntilSuccessful
Co-Authored-By: Claude Opus 4.6 <no...@anthropic.com>
Signed-off-by: Claus Ibsen <cl...@gmail.com>
---
.../salesforce/SalesforceEndpointConfig.java | 3 +-
.../salesforce/internal/SalesforceSession.java | 6 +---
.../internal/client/DefaultBulkApiV2Client.java | 2 ++
.../internal/client/PubSubApiClient.java | 21 +++++++++++---
.../internal/client/SalesforceSecurityHandler.java | 33 +++++++++++++---------
.../internal/processor/JsonRestProcessor.java | 10 +++++--
.../internal/streaming/SubscriptionHelper.java | 22 +++++++++++++--
7 files changed, 67 insertions(+), 30 deletions(-)
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
index 6b8f3f4d0778..843d7e42a04d 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/SalesforceEndpointConfig.java
@@ -834,6 +834,7 @@ public class SalesforceEndpointConfig implements Cloneable {
valueMap.put(APEX_URL, apexUrl);
// apexQueryParams are handled explicitly in AbstractRestProcessor
valueMap.put(COMPOSITE_METHOD, compositeMethod);
+ valueMap.put(ALL_OR_NONE, allOrNone);
valueMap.put(LIMIT, limit);
valueMap.put(APPROVAL, approval);
valueMap.put(EVENT_NAME, eventName);
@@ -869,7 +870,7 @@ public class SalesforceEndpointConfig implements Cloneable {
valueMap.put(INITIAL_REPLAY_ID_MAP, initialReplayIdMap);
// add Pub/Sub API properties
- valueMap.put(REPLAY_PRESET, initialReplayIdMap);
+ valueMap.put(REPLAY_PRESET, replayPreset);
valueMap.put(PUB_SUB_DESERIALIZE_TYPE, pubSubDeserializeType);
valueMap.put(PUB_SUB_POJO_CLASS, pubSubPojoClass);
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/SalesforceSession.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/SalesforceSession.java
index e810917ce066..fbbb9ca40649 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/SalesforceSession.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/SalesforceSession.java
@@ -91,7 +91,7 @@ public class SalesforceSession extends ServiceSupport {
private final CamelContext camelContext;
private final AtomicBoolean loggingIn = new AtomicBoolean();
- private CountDownLatch latch = new CountDownLatch(1);
+ private volatile CountDownLatch latch = new CountDownLatch(1);
public SalesforceSession(CamelContext camelContext, SalesforceHttpClient httpClient, long timeout,
SalesforceLoginConfig config) {
@@ -113,11 +113,7 @@ public class SalesforceSession extends ServiceSupport {
// if another thread is logging in, we will just wait until it's successful
if (!loggingIn.compareAndSet(false, true)) {
LOG.debug("waiting on login from another thread");
- // TODO: This is janky
try {
- while (latch == null) {
- Thread.sleep(100);
- }
latch.await();
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
diff --git a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
index 3fcd74054bd7..7bbbe07d169f 100644
--- a/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
+++ b/components/camel-salesforce/camel-salesforce-component/src/main/java/org/apache/camel/component/salesforce/internal/client/DefaultBulkApiV2Client.java
@@ -113,6 +113,7 @@ public class DefaultBulkApiV2Client extends AbstractClientBase implements BulkAp
public void onResponse(InputStream response, Map<String, String> headers, SalesforceException ex) {
if (ex != null) {
callback.onResponse(null, headers, ex);
+ return;
}
Job responseJob = null;
try {
@@ -237,6 +238,7 @@ public class DefaultBulkApiV2Client extends AbstractClientBase implements BulkAp
public void onResponse(InputStream response, Map<String, String> headers, SalesforceException ex) {
if (ex != null) {
callback.onResponse(null, headers, ex);
+ return;
}
QueryJob responseJob = null;
try {
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 edff9c8c854a..aafcdd084360 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
@@ -342,8 +342,12 @@ public class PubSubApiClient extends ServiceSupport {
String errorCode = "";
LOG.error("Trailers:");
if (trailers != null) {
- trailers.keys().forEach(trailer -> LOG.error("Trailer: {}, Value: {}", trailer,
- trailers.get(Metadata.Key.of(trailer, Metadata.ASCII_STRING_MARSHALLER))));
+ trailers.keys().forEach(trailer -> {
+ if (!trailer.endsWith(Metadata.BINARY_HEADER_SUFFIX)) {
+ LOG.error("Trailer: {}, Value: {}", trailer,
+ trailers.get(Metadata.Key.of(trailer, Metadata.ASCII_STRING_MARSHALLER)));
+ }
+ });
errorCode = trailers.get(Metadata.Key.of("error-code", Metadata.ASCII_STRING_MARSHALLER));
}
if (errorCode != null) {
@@ -383,12 +387,21 @@ public class PubSubApiClient extends ServiceSupport {
}
private void resubscribeOnError() {
+ if (isStoppingOrStopped() || channel == null || channel.isShutdown()) {
+ LOG.debug("Client is stopping or channel is shut down, skipping resubscribe");
+ return;
+ }
try {
LOG.debug("Will attempt resubscribe in {} ms", reconnectDelay);
Thread.sleep(reconnectDelay);
- reconnectDelay = reconnectDelay + backoffIncrement;
+ reconnectDelay = Math.min(reconnectDelay + backoffIncrement, maxBackoff);
} catch (InterruptedException e) {
- throw new RuntimeException(e);
+ Thread.currentThread().interrupt();
+ return;
+ }
+ if (isStoppingOrStopped() || channel == null || channel.isShutdown()) {
+ LOG.debug("Client is stopping or channel is shut down, skipping resubscribe");
+ return;
}
if (replayId != null) {
subscribe(consumer, ReplayPreset.CUSTOM, replayId, fallbackToLatestReplayId);
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 054168e21e0a..4345234aeb76 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
@@ -171,16 +171,19 @@ public class SalesforceSecurityHandler implements ProtocolHandler {
// Salesforce will allow successful login with an expired password, but any subsequent
// API calls will fail with a 401 and message about expired password.
// It's fatal. User must reset password.
- List<RestError> errors = Collections.emptyList();
- try {
- errors = client.readErrorsFrom(getContentAsInputStream(), objectMapper);
- } catch (IOException e) {
- LOG.warn("Unable to deserialize errors from response body.");
- }
- if (errors.stream().anyMatch(error -> EXPIRED_PASSWORD_CODE.equals(error.getErrorCode()))) {
- SalesforceException salesforceException = createSalesforceException(client, status);
- forwardFailureComplete(request, null, response, salesforceException);
- return;
+ // Note: client may be null for CometD streaming requests
+ if (client != null) {
+ List<RestError> errors = Collections.emptyList();
+ try {
+ errors = client.readErrorsFrom(getContentAsInputStream(), objectMapper);
+ } catch (IOException e) {
+ LOG.warn("Unable to deserialize errors from response body.");
+ }
+ if (errors.stream().anyMatch(error -> EXPIRED_PASSWORD_CODE.equals(error.getErrorCode()))) {
+ SalesforceException salesforceException = createSalesforceException(client, status);
+ forwardFailureComplete(request, null, response, salesforceException);
+ return;
+ }
}
// REST token expiry
@@ -213,10 +216,12 @@ public class SalesforceSecurityHandler implements ProtocolHandler {
private SalesforceException createSalesforceException(AbstractClientBase client, int statusCode) {
List<RestError> restErrors = Collections.emptyList();
- try {
- restErrors = client.readErrorsFrom(getContentAsInputStream(), new ObjectMapper());
- } catch (IOException e) {
- LOG.warn("Unable to deserialize errors from response body.");
+ if (client != null) {
+ try {
+ restErrors = client.readErrorsFrom(getContentAsInputStream(), new ObjectMapper());
+ } catch (IOException e) {
+ LOG.warn("Unable to deserialize errors from response body.");
+ }
}
return new SalesforceException(restErrors, statusCode);
}
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 2542ee8d5e1a..5803a05fbd03 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
@@ -243,7 +243,8 @@ public class JsonRestProcessor extends AbstractRestProcessor {
Class<?> responseClass;
try (final JsonParser parser = new JsonFactory().createParser(responseEntity)) {
String type = null;
- while (parser.nextToken() != JsonToken.END_OBJECT) {
+ JsonToken token;
+ while ((token = parser.nextToken()) != null && token != JsonToken.END_OBJECT) {
String propName = parser.currentName();
if ("type".equals(propName)) {
parser.nextToken();
@@ -251,10 +252,13 @@ public class JsonRestProcessor extends AbstractRestProcessor {
break;
}
}
+ if (type == null) {
+ return null;
+ }
String prefix = exchange.getProperty(RESPONSE_CLASS_PREFIX, "", String.class);
responseClass = getSObjectClass(prefix + type, null);
- } catch (IOException | SalesforceException exc) {
- throw new RuntimeException(exc);
+ } catch (SalesforceException exc) {
+ throw new IOException("Failed to detect SObject response class", exc);
} finally {
responseEntity.reset();
}
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 3e8cef329cde..9a6d4b13e209 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
@@ -144,9 +144,25 @@ public class SubscriptionHelper extends ServiceSupport {
}
}
}
- // failed, so keep trying
- LOG.debug("Handshake failed, so try again.");
- client.handshake();
+ // 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.foregroundTask()
+ .withBudget(Budgets.iterationBudget()
+ .withMaxIterations(1)
+ .withInitialDelay(Duration.ofMillis(backoff))
+ .withInterval(Duration.ZERO)
+ .build())
+ .withName("SalesforceHandshakeRetryDelay")
+ .build()
+ .run(component.getCamelContext(), () -> true);
+ }
+ client.handshake();
+ }
} else if (!channelToConsumers.isEmpty()) {
channelsLock.lock();
try {