Failed to create CamelSalesforcesourceSourceConnector: ResolveEndpointFailedException

[apoorvmintri]

Background: I am a python developer & have never done Java.

Camel Salesforce Source Connector: v3.20.6 ( I have tried 3.21.0 & 4.0.0 and I'm facing the same issue)
Kafka: v3.6.1

Issue:

Trying to create CamelSalesforcesourceSourceConnector for my kafka instance and it fails with the exception ResolveEndpointFailedException. I have tried a few variations of creating the connector using Kafka's REST API.

Variation 1:

{ "name": "sf-source-connector", "config": { "tasks.max":"1", "connector.class":"org.apache.camel.kafkaconnector.salesforcesource.CamelSalesforcesourceSourceConnector", "key.converter":"org.apache.kafka.connect.storage.StringConverter", "value.converter":"org.apache.kafka.connect.storage.StringConverter", "camel.kamelet.salesforce-source.query": "SELECT * FROM Account", "camel.kamelet.salesforce-source.topicName": "/data/AccountChangeEvent", "camel.kamelet.salesforce-source.loginUrl": "https://login.salesforce.com/", "camel.kamelet.salesforce-source.clientId": "<Redacted>", "camel.kamelet.salesforce-source.clientSecret": "<Redacted>", "camel.kamelet.salesforce-source.userName": "<Redacted>", "camel.kamelet.salesforce-source.password": "<Redacted>", "camel.kamelet.salesforce-source.notifyForOperationCreate": "true", "camel.kamelet.salesforce-source.notifyForOperationUpdate": "true", "camel.kamelet.salesforce-source.notifyForOperationDelete": "true", "camel.kamelet.salesforce-source.notifyForOperationUndelete": "true", "camel.source.endpoint.rawPayload": "true", "camel.kamelet.salesforce-source.operation": "subscribe", "topics": "camelsfstream" } }

Failure Exception Summary -

Starts with - org.apache.kafka.connect.errors.ConnectException: Failed to create and start Camel context\n\tat org.apache.camel.kafkaconnector.CamelSourceTask.start(CamelSourceTask.java:184)
Ends with - Caused by: java.lang.IllegalArgumentException: /data/AccountChangeEvent\n\tat org.apache.camel.component.salesforce.internal.OperationName.fromValue(OperationName.java:128)\n\tat org.apache.camel.component.salesforce.SalesforceComponent.createEndpoint(SalesforceComponent.java:303)\n\tat org.apache.camel.support.DefaultComponent.createEndpoint(DefaultComponent.java:171)\n\tat org.apache.camel.impl.engine.AbstractCamelContext.doGetEndpoint(AbstractCamelContext.java:975)

Key Issue -

IllegalArgumentException: /data/AccountChangeEvent

Variation 2 -

In my connector I replaced the topic name: "camel.kamelet.salesforce-source.topicName": "subscribe:/data/AccountChangeEvent"

The exception starts with Failed to create and start Camel context, but the final exception changes to -

due to: There are 1 parameters that couldn't be set on the endpoint. Check the uri if the parameters are spelt correctly and that they are properties of the endpoint. Unknown parameters=[{operationName=subscribe}]

I am unable to overwrite the endpoint url to test the above variation without the operationName and I am also aware that subscribe is the default. So even if I remove that property from the json, it makes no difference.

On another note the endpoint starts with local-salesforce-1:// the # increments - is that right? I couldnt find any property to change this. I imagine it gets replaced eventually.

Closest Issues I found -

• Salesforce kamelet - Adjust for Camel 3.19.x camel-kamelets#998 Closed
• CamelSalesforcesourceSourceConnector: Failed to resolve endpoint error #1433 Open

[apoorvmintri]

Workaround

Realizing that the "operationName" property was added in v3.19.x, I switched to 3.18.x (I used the latest - v3.18.2). And my json was the same as Version 2 (described in the problem) -

{ "name": "sf-source-connector", "config": { "tasks.max":"1", "connector.class":"org.apache.camel.kafkaconnector.salesforcesource.CamelSalesforcesourceSourceConnector", "key.converter":"org.apache.kafka.connect.storage.StringConverter", "value.converter":"org.apache.kafka.connect.storage.StringConverter", "camel.kamelet.salesforce-source.query": "SELECT * FROM Account", "camel.kamelet.salesforce-source.topicName": "subscribe:/data/AccountChangeEvent", "camel.kamelet.salesforce-source.loginUrl": "https://login.salesforce.com/", "camel.kamelet.salesforce-source.clientId": "<Redacted>", "camel.kamelet.salesforce-source.clientSecret": "<Redacted>", "camel.kamelet.salesforce-source.userName": "<Redacted>", "camel.kamelet.salesforce-source.password": "<Redacted>", "camel.kamelet.salesforce-source.notifyForOperationCreate": "true", "camel.kamelet.salesforce-source.notifyForOperationUpdate": "true", "camel.kamelet.salesforce-source.notifyForOperationDelete": "true", "camel.kamelet.salesforce-source.notifyForOperationUndelete": "true", "camel.source.endpoint.rawPayload": "true", "topics": "camelsfstream" } }

Note -
"camel.kamelet.salesforce-source.operation": "subscribe" is removed &
"camel.kamelet.salesforce-source.topicName": "subscribe:/data/AccountChangeEvent" prefix 'subscribe:' added to topic

I did face Authorization Failure, which was resolved by going to Salesforce Setup > OAuth and OpenId Connect Settings, and enabling 'Allow OAuth Usename-Password Flows'. Just FYI, there are other policies that should be setup in your salesforce connected app to make this connection successful.

After this, I faced the infamous InputStreamCache messages in my kafka instance instead of the actual data, as described in issue #1433 . To resolve this, the following workaround I described here works -
#1433 (comment)

@oscerd I hope this helped and if there's anything I should improve, please please do let me know.

[oscerd]

Thanks for reporting, if this is fixed please close the issue.

[apoorvmintri]

@oscerd The issue is fixed by reverting to 3.18.x, issue persists for 3.19.x and above (incl 4.0.0), unless there is something wrong with my configuration. This is just an fyi, if you still think I should close it, then I will.

[oscerd]

I'll try to investigate, but I don't have an ETA.

[oscerd]

Leave it open for the moment, thank you.

[apoorvmintri]

@oscerd In addition to the operationName, there is the issue with InputStreamCache being transported to Kafka. This could only be resolved by using ByteArrayConverter & disabling the stream caching as described in #1433 (comment) another ticket that is open.

Not sure if this is a bug as well.

[oscerd]

StreamCaching has been enabled by default in Camel. So it must be disabled with camel.main.streamCachingEnabled set to false.

[apoorvmintri]

Sorry for spamming. The salesforce source kafka connector seems to be unusable above 3.19.x as "operationName=subscribe" is added by default and the topic name does not work without the "subscribe:" prefix. This leads to either IllegalArgumentException on the topic name (if prefix not added) OR an unknown property exception for operationName.

Not an issue if devs use 3.18.x or earlier. Just letting you know in case this is supposed to be a higher priority issue as the component is unusable, assuming there isn't any other workaround for those versions.

[oscerd]

apache/camel-kamelets#1841

[oscerd]

We'll fix that in 4.4.0.

[oscerd]

We just released 4.0.3 camel-kamelets-catalog, and we're going to release 4.0.3 camel-kafka-connector based on that. So the issue will be fixed in 4.4.0 and 4.0.4. 4.4.0 is an LTS release, so we'll release camel-kafka-connector based on that, while 4.0.4 will be released later.

[aonamrata]

Hi, Any idea when 4.0.4 would be available? I am getting the same error and dont see 4.0.4 in
https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-salesforce-source-kafka-connector/ list.

[oscerd]

No ETA at this stage.