## Camel Salesforce kafka source connector

#### Docs: https://camel.apache.org/components/3.21.x/salesforce-component.html

### We are using Salesforce Change Data Capture feature.

Change Data Capture publishes change events, which represent changes to Salesforce records. Changes include creation of a new record, updates to an existing record, deletion of a record, and undeletion of a record.
Message structure: https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_message_structure.htm


### Run it locally :

```
cp .env.shadow .env
# update .env file with creds.
awsume dev
aws ecr get-login-password --region us-east-1 | docker login --username AWS --password-stdin 086679231553.dkr.ecr.us-east-1.amazonaws.com
docker-compose up --build -d
```


### Configuration options to run for a new type:
- `KAFKA_TOPICS` - name of kafka topic where it will send the cdc events.
- `SF_TOPIC` - name of salesforce CDC channel that this connecter will listen to. If you want all CDC events then use `/data/ChangeEvents` instead of `/data/Fan__ChangeEvent` which is for Fan sObject only. Ref: [Salesforce Subscription Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm)
- `SF_SOBJECT_QUERY_FILE` - filter the events based on this SOQL query and send to kafka topic. Refer [Filter Your Stream of Change Events with Channels](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_filter_section.htm)
- `SF_SOBJECT_QUERY` - Instead of specifyign a query file you can directly specify the SOQL query as a string. If both SF_SOBJECT_QUERY_FILE and SF_SOBJECT_QUERY env are specified, SF_SOBJECT_QUERY takes preference.
- `SF_INSTANCE` - it can be `https://test.salesforce.com` or `https://sony--fullcopysb.sandbox.my.salesforce.com`
- `SF_REPLAY_ID` - to replay old messages. You can find this in messages headers in kafka topic. Refer: [Event Replay Process](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/using_streaming_api_durability.htm#generic_replay_event_replay_details) .
- `CONNECT_VALUE_CONVERTER` - control how message body is converted. Use ByteArrayConverter for CDC messages so you get a json instead of a base64 encoded payload.
- `RAW_PAYLOAD` - Controls if you want to use raw payload String for request and response. Set this to true if you want raw message with headers etc as as string in kafka message. Eg:
    ````
    {data={schema=8pgFT5npbFPH6CEgjcgdqA, payload={LastModifiedDate=2024-02-20T08:54:58.000Z, ChangeEventHeader={commitNumber=1708419299123765249, commitUser=005Fg000002qeMrIAI, sequenceNumber=1, entityName=Fan__c, changeType=UPDATE, changedFields=[Ljava.lang.Object;@5c965d4a, changeOrigin=, transactionKey=00000be8-388d-9057-a1e6-e0528859ed76, commitTimestamp=1708419298000, recordIds=[Ljava.lang.Object;@2ad7ba7f}, Email__c=rambo@fighter.com}, event={replayId=4401}}, channel=/data/Fan__ChangeEvent}
    ````

- `SECRETS_MANAGER_SF_CREDS`


### Some other things to note with current setup:
- API Rate limits - https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_allocations.htm
- SF_REPLAY_ID - This config does not work with 3.18.2 as it is. You have to also send it with `SF_TOPIC: "/data/Fan__ChangeEvent?replayId=4497&"` for it to work. This is an open issue with the connector.
- clientId and secrets are specified multiple times in config because of config naming mismatch between diff libraries.
- different varients of userName/ username in config. userName is mandatory param but without username it will give login error invalid_grant. There is some inconsistency in their package.


### We are on Version 3.18.2 instead of 4.0.0 because of some issues that will be resolved in V4.0.4
- https://github.com/apache/camel-kafka-connector/issues/1592
- https://github.com/apache/camel-kafka-connector/issues/1433
- https://github.com/apache/camel-kafka-connector/issues/1587
- https://github.com/apache/camel-kafka-connector/issues/1598
- Also all the configuration names are different from 4.x. They start with `camel.kamelet`.