Failed to create CamelSalesforcesourceSourceConnector: ResolveEndpointFailedException

apoorvmintri commented 10 months ago

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 -

apoorvmintri commented 10 months ago

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 - https://github.com/apache/camel-kafka-connector/issues/1433#issuecomment-1903471389

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