# SalesForce Sink The SalesForce Sink Connector pulls data from a Kafka topic and writes data into a SalesForce Object (sObject) using the available SalesForce API. This connector is based on the Apache Camel framework, and it's official documentation can be [found here](https://camel.netlify.app/camel-kafka-connector/latest/connectors/camel-salesforce-kafka-sink-connector.html). ## Prerequisites - **SalesForce Connected App Auth Credentials**: A new App must be created in SalesForce with the right permissions. Check out the guide [Setting Up SalesForce OAuth App in Notion](https://www.notion.so/Setting-Up-Salesforce-OAuth-App-e311551e00764749bb459aa9a3db336e). ## Configuration Properties The following are the configuration properties specific to this connector that are exposed through Environment Variables in the Connector. Note that common environment variables are described in the [Deploying Connectors doc](/docs/kafka-connect/deploying-connectors.mdx). | Environment Variable | Required | Type | Default | Description | |------------------------------- |--------- |------ |-------- |------------------------------------------------------------------------------- | | **DEBUG_MODE** | Yes | `bool`| ` ` | Enable extra debug logging on the connector. | | **ERRORS_TOLERANCE** | Yes | `str` | `all` | Allow `all` or `none` errors to happen without failing the connector. | | **RAW_PAYLOAD** | No | `bool`| `true` | Use raw payload String for request and response instead of DTOs. | | **SECRETS_MANAGER_SF_CREDS** | No | `str` | ` ` | Name of the Secret in Secrets Manager. | | **SF_AUTH_TYPE** | Yes | `str` | `REFRESH_TOKEN`| SalesForce API Authentication Type. | | **SF_CONSUMER_KEY** | No[^1] | `str` | ` ` | SalesForce API Consumer Key. **Overrides SecretsManager** | | **SF_CONSUMER_SECRET** | No[^1] | `str` | ` ` | SalesForce API Consumer Secret. **Overrides SecretsManager** | | **SF_INSTANCE** | Yes | `str` | ` ` | SalesForce instance URL. | | **SF_REFRESH_TOKEN** | No[^1] | `str` | ` ` | SalesForce API Refresh Token. **Overrides SecretsManager** | | **SOBJECT_ID_NAME** | No | `str` | ` ` | API Name of the sObject unique ID field. Used for Upsert. | | **SOBJECT_NAME** | Yes | `str` | ` ` | API Name of the SalesForce Object to sink into. | | **SOBJECT_OPERATION** | Yes | `str` | ` ` | API Operation. Current supported options: `upsertSObject`, `createSObject`, `updateSObject` and `deleteSObject`. | | **START_READING_RECORDS_FROM** | Yes | `str` | ` ` | From where to start reading records from in the topic; `latest` or `earliest`. | [^1]: Only required when Secrets Manager is not used ## Secrets This connector expects to find the following secret in Secrets Manager when their corresponding Environment Variables are **not set**. **Secret Name Structure**: `${Environment}/${SERVICE_NAME}/${SECRETS_MANAGER_SF_CREDS}` **Secret Mandatory Keys**: - **sf_consumer_key**: SalesForce API Consumer Key. - **sf_consumer_secret**: SalesForce API Consumer Secret. - **sf_refresh_token**: SalesForce API Refresh Token. ## Creating into SalesForce Object Create operations through the Connector enable us to Insert new records into the sObject. To configure the connector for create operations, the following environment variables **must be present**: - **SOBJECT_NAME**: The API name of the SalesForce Object to upsert into. - **SOBJECT_OPERATION**: `createSObject` for upsert operations. ## Upserting into SalesForce Object Upserting operations through the Connector enable us to Update or Insert new records into the sObject based on it's unique ID. When a new message is being sent to SalesForce, it will check if the sObject ID exists and update its fields; or if the sObject ID can't be found in the sObject, it will try to insert it as new. :::note SalesForce internal IDs Inserting new records through this operation when the used unique sObjectID field is SalesForce's own internally generated ID is not supported as the underlying API expects the ID to always be populated. ::: To configure the connector for upsert operations, the following environment variables **must be present**: - **SOBJECT_NAME**: The API name of the SalesForce Object to upsert into. - **SOBJECT_ID_NAME**: The API name of the field used as upsert key. - **SOBJECT_OPERATION**: `upsertSObject` for upsert operations. ### Setting the Upsert Key value This connector doesn't allow the ID to be sent as part of the message, instead this must be set as a Header in the messages sent to the topic. This header's key name has to be `CamelHeader.sObjectIdValue`, and it's value must be the ID of the object that needs to be updated or inserted (new id). ## Update a SalesForce Object To configure the connector for update operations, the following environment variables **must be present**: - **SOBJECT_NAME**: The API name of the SalesForce object. Eg `Subscription__c` - **SOBJECT_OPERATION**: `updateSObject` for update operations. The config param `camel.sink.endpoint.sObjectIdName` is ignored in case of update operation. ### Setting the Update Key value This connector doesn't allow the ID to be sent as part of the message, instead this must be set as a Header in the messages sent to the topic. For **Update** operations the messages in kafka topic must include a header with header's key name has to be `CamelHeader.sObjectId`, and it's value must be the ID of the object that needs to be updated. If a record with such key exists, the message updates an existing row in the sObject, otherwise it will throw an error and the message will be moved to DLQ topic. Error message: ``` org.apache.camel.component.salesforce.api.NoSuchSObjectException: {errors:[{"errorCode":"NOT_FOUND","message":"The requested resource does not exist","fields":null}],statusCode:404} ``` ## Dead Letter Queue and Errors You can configure a DLQ for this connector so failed messages are stored in a topic with information about their failure. To enable a DLQ on an existing topic, the following environment variables must be set: - **DLQ_TOPIC_NAME**: Name of an existing DLQ Kafka Topic. - **ERRORS_TOLERANCE**: `all` to allow the connector to pass through errors into the DLQ. Once this is set, any new messages that fail to be sent into SalesForce will appear in the defined DLQ topic, and exploring each message's headers you will be able to identify the error. ## Existing examples - [SalesForce Sink in Kafka Infra](https://github.com/theorchard/terraform-infra/tree/master/qa/kafka-infra/salesforce_sink)