<a id="salesforce-cdc-source-connector"></a>

# Salesforce Change Data Capture Source Connector for Confluent Platform

The Kafka Connect Salesforce [Change Data Capture](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm)
Source connector monitors Salesforce records. Salesforce sends a notification
when a change to a Salesforce record occurs due to a create, update, delete, or
undelete operation. The Kafka Connect Salesforce Change Data Capture Source connector can be
used to capture these change events and write them to Kafka topic. This connector
can be used with either standalone or distributed Connect workers.

#### NOTE
* The connector’s `salesforce.cdc.name` property indicates the name of the change event.
  In case of standard objects in standard channels, the property indicates the name of the
  change event, and must end with `ChangeEvent`. In case of custom objects in standard channels,
  the name of the change event must end with a `__ChangeEvent`. For example,  `salesforce.cdc.name=Employee__ChangeEvent`.
  These change events must exist in Salesforce before running the connector.
  For more information, see [Salesforce Subscription Guide](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm).
* The connector supports Salesforce up to API version 65.0.
* <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->

  Effective July 6, 2025, only self-managed connector versions that meet or exceed the minimum version listed on the
  [Supported Connector Versions](https://docs.confluent.io/platform/7.8/connect/supported-connector-version.html#supported-connector-versions-till-cp-7-8)
  page receive support from Confluent. Older, unsupported connector versions have been removed from Confluent Marketplace and
  are no longer available for download.

## Features

The Salesforce CDC Source Connector includes the following features:

- [At least once delivery](#salesforce-cdc-source-at-least-once-delivery)
- [Supports one task](#salesforce-cdc-source-supports-one-task)
- [Client-side encryption](#salesforce-cdc-source-csfle-sm)
- [Salesforce enriched events](#salesforce-enriched-events)

<a id="salesforce-cdc-source-at-least-once-delivery"></a>

### At least once delivery

This connector guarantees that records are delivered at least once to the Kafka
topic. If the connector restarts, there may be some duplicate
records in the Kafka topic.

<a id="salesforce-cdc-source-supports-one-task"></a>

### Supports one task

The Salesforce CDC Source Connector supports running only one task.

<a id="salesforce-cdc-source-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

<a id="salesforce-enriched-events"></a>

### Salesforce enriched events

The connector supports enriching Salesforce Change Data Capture (CDC) events with additional fields on both custom channels
(for example, `/data/SalesEvents__chn`) and the standard `/data/ChangeEvents` channel, provided event enrichment is
configured in Salesforce for the associated entities. For more information, see  [Salesforce’s CDC event enrichment](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm) documentation.

<a id="salesforce-cdc-source-connector-considerations-limits"></a>

## Limitations

The following limitations apply to this Salesforce connector:

* The connector is limited to one task only.
* The connector is limited by the [allocations and limits](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/limits.htm) of the Salesforce account. For example, free org accounts have a cap on the number of events that can occur within a 24-hour period.
* The connector requires restarting before publishing a new schema.
* The connector does not support real time updates for Salesforce objects schema changes.
  To synchronize Salesforce objects schema changes, restart the connector.

<a id="salesforce-cdc-source-connector-considerations"></a>

## Considerations

Note the following when using the Salesforce CDC Events source connector.

<a id="salesforce-cdc-source-connector-considerations-replay"></a>

### Replaying Messages

Change events are published to the [event bus](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/using_streaming_api_durability.htm), where they are stored temporarily for 72 hours (3 days). When you create a connector, you can control how stored events are treated when the connector starts for the first time.

1. Consume only new change events after the connector starts up, by setting `salesforce.initial.start` to `latest`. This is the default behavior, due largely to [API limits](#salesforce-cdc-source-connector-considerations-limits).
2. Consume all of the change events by setting `salesforce.initial.start` to `all`.

<a id="salesforce-cdc-source-connector-considerations-restart"></a>

### Restarting

When the connector operates, it periodically records the *replay ID* of the last record written to Kafka. When the connector is stopped and then restarted within 24 hours (for standard volume messages) or 72 hours (for high volume messages) the connector continues consuming the change events where it stopped, with no missed events. However, if the connector is stopped for more than 24 or 72 hours, some events are discarded in Salesforce before the connector can read them.

If the connector stops unexpectedly due to a failure, it may not record the *replay ID* of the last record successfully written to Kafka. When the connector restarts, it resumes from the last recorded *replay ID*. This means that some events may be duplicated in Kafka.

<a id="salesforce-cdc-source-connector-considerations-retries"></a>

### Retries

In case of failures caused by network issues, you can configure the connector to retry Salesforce requests (once authentication succeeds) using `request.max.retries.time.ms` parameters.

The backoff period for each retry attempt uses a randomization function that grows exponentially. But, if the total time spent retrying the request exceeds this duration (15 minutes by default), retries stop and the request fails. This will likely result in task failure.

### Retries Example

1. Create a `sfcdc-source.properties` file with the following contents:
   ```none
   name=RetriesExample
   topics=sfcdc-messages
   tasks.max=1
   connector.class=io.confluent.salesforce.SalesforceCdcSourceConnector
   key.converter=org.apache.kafka.connect.json.JsonConverter
   value.converter=org.apache.kafka.connect.json.JsonConverter
   confluent.topic.bootstrap.servers=localhost:9092
   confluent.topic.replication.factor=1
   salesforce.cdc.name=LeadChangeEvent
   request.max.retries.time.ms=20000
   ```
2. Run and validate the connector.
3. The Connector retries requests for an initial backoff duration of 20000 ms. Retries stop if a Salesforce request is successful.

Note that the default value for `request.max.retries.time.ms` is 900000 ms (15 minutes).

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

For license properties, see [Confluent Platform license](salesforce_cdc_source_connector_config.md#salesforce-cdc-source-license-config). For
information about the license topic, see
[License topic configuration](salesforce_cdc_source_connector_config.md#salesforce-cdc-source-license-topic-configuration).

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Salesforce Change Data Capture Source Connector Configuration Properties](salesforce_cdc_source_connector_config.md#salesforce-cdc-source-connector-config).

## Install the Salesforce Change Data Capture Connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- Salesforce Streaming API version 45.0 or later.
- The Salesforce user account configured for the connector must have permission
  to **View All Data**, in addition to the permissions listed below. For
  details, see [Required Permissions for Change Events Received by CometD
  Subscribers](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_security_perms.htm).
- Before configuring the connector, ensure your Salesforce environment meets
  the following requirements:
  * **User permissions:** The Salesforce user account must have the following
    permissions: **View All Data**, **API Enabled**, **View Setup and
    Configuration**, and **Read** access to the target objects.
  * **CDC configuration:** Explicitly enable CDC for all objects you intend to
    capture.
- Salesforce change notification subscription for Salesforce entities. For
  details, see [Salesforce Change Data Capture Setup](#salesforce-cdc-setup).
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-salesforce:latest
  ```

  You can install a specific version by replacing `latest` with a version
  number as shown in the following example:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-salesforce:3.1.0
  ```

## Quick Start

This quick start uses the Salesforce Change Data Capture Source Connector to capture change events and write them to Kafka topic.

<a id="salesforce-cdc-setup"></a>

### Salesforce Change Data Capture Setup

Following are the steps to enable **change notification** for Salesforce account.

1. Use [https://login.salesforce.com/](https://login.salesforce.com/) to log into an existing account or create a new one.
2. Create a new Connected App.
   1. Select the gear icon in the upper right hand corner and choose **Setup**.
   2. Enter **App** in the **Quick Find** search box, and choose **App Manager** in the filtered results.
   3. Click the **New Connected App** button in the upper right corner of the Setup panel.
   4. Supply a **Connected App Name**, **API Name**, and **Contact Email**.
   5. Select **Enable OAuth Settings** checkbox and select the **Enable for Device Flow** checkbox. These selections enable the connector to use the Salesforce API.
   6. Under the **Select OAuth Scopes** field, select all of the items under **Available OAuth scopes** and add them to the **Selected OAuth Scopes**.
   7. Save the new app and press **Continue** at the prompt.
   8. Look for the **Consumer Key** and **Consumer Secret** in the displayed form. Save these so you can put them in the configuration properties file for the Salesforce connect worker.
3. Find your Security Token (emailed to you from Salesforce.com). If you need to reset your token or view your profile on Salesforce.com, select **Reset My Security Token** and follow the instructions.
4. Finally, [subscribe](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_select_objects.htm) to change event notifications for Salesforce entities. Subscribe the **Leads** entity.

Following are the steps to add a Lead to Salesforce:

1. In the Salesforce UI, find and click on the **App launcher** icon in the upper left of the page.
2. Find and start the Leads application.
3. Click the **New** button, fill out the form, and save it. The Lead has been created.
4. Alternatively, the leads can be created through simple a Java application, follow the [salesforce documentation](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_objects_lead.htm).

### Start Confluent Platform

Start the Confluent services using the following [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) command:

```bash
confluent local start
```

#### IMPORTANT
Do not use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.

### Property-based example

Create a configuration file `salesforce-cdc-source.properties` with the following content. This file should be placed inside the Confluent Platform installation directory. This configuration is used typically along with [standalone workers](/platform/current/connect/concepts.html#standalone-workers).

```none
name=SalesforceCdcSourceConenctor

connector.class=io.confluent.salesforce.SalesforceCdcSourceConnector
tasks.max=1
kafka.topic=< Required Configuration >

salesforce.consumer.key=< Required Configuration >
salesforce.consumer.secret=< Required Configuration >
salesforce.username=< Required Configuration >
salesforce.password=< Required Configuration >
salesforce.password.token=< Required Configuration >

salesforce.cdc.name=< Required Configuration >
salesforce.initial.start=all

confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1
confluent.license=
```

#### NOTE
Note the following:

* `salesforce.consumer.key` and `salesforce.consumer.secret` are required properties used for OAuth2 secure authentication by Salesforce.com. Additional information and tutorials are available at [salesforce.com](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/code_sample_auth_oauth.htm).
* Change the `confluent.topic.bootstrap.servers` property to include your broker address(es) and change the `confluent.topic.replication.factor` to `3` for staging or production use.
* Set the following connector configuration properties to enable OAuth JWT bearer token support:
  - `salesforce.username`
  - `salesforce.consumer.key`
  - `salesforce.jwt.keystore.path`
  - `salesforce.jwt.keystore.password`

Run the connector with this configuration.

```bash
confluent local load SalesforceCdcSourceConenctor --config salesforce-cdc-source.properties
```

Confirm that the connector is in a `RUNNING` state.

```bash
confluent local status SalesforceCdcSourceConenctor
```

### REST-based example

Use this setting with [distributed workers](/platform/current/connect/concepts.html#distributed-workers). Write the following JSON to `config.json`, configure all of the required values, and use the following command to post the configuration to one of the distributed connect workers. Check here for more information about the Kafka Connect [REST API](/platform/current/connect/references/restapi.html)

```json
{
  "name" : "SalesforceCdcSourceConenctor",
  "config" : {
    "connector.class" : "io.confluent.salesforce.SalesforceCdcSourceConnector",
    "tasks.max" : "1",
    "kafka.topic" : "< Required Configuration >",
    "salesforce.cdc.name" : "< Required Configuration >",
    "salesforce.initial.start" : "all",
    "salesforce.consumer.key" : "< Required Configuration >",
    "salesforce.consumer.secret" : "< Required Configuration >",
    "salesforce.username" : "< Required Configuration >",
    "salesforce.password" : "< Required Configuration >",
    "salesforce.password.token" : "< Required Configuration >",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "confluent.topic.replication.factor": "1",
    "confluent.license": " Omit to enable trial mode "
  }
}
```

#### NOTE
Change the `confluent.topic.bootstrap.servers` property to include your broker address(es), and change the `confluent.topic.replication.factor` to 3 for staging or production use.

Use curl to post a configuration to one of the Kafka Connect workers. Change `http://localhost:8083/` to the endpoint of one of your Kafka Connect worker(s).

```bash
curl -s -X POST -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors
```

Use the following command to update the configuration of existing connector.

```bash
curl -s -X PUT -H 'Content-Type: application/json' --data @config.json http://localhost:8083/connectors/SalesforceCdcSourceConenctor/config
```

To consume records written by the connector to the configured Kafka topic, run the following command:

```bash
kafka-avro-console-consumer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081  --topic cdc-topic --from-beginning
```

The leads that were inserted into Salesforce will be present in the Kafka topic.

<a id="salesforce-cdc-source-connector-record-format"></a>

## Record Schema

### Change Data Capture Record Format

The following example shows the JSON document structure of a Salesforce Change Data Capture Event as received by the Salesforce Change Data Capture Connector.

```json
{
   "data":{
      "schema":"MbZFVarhwdRnx4cf-8DowQ",
      "payload":{
         "Status":"Open - Not Contacted",
         "LastModifiedDate":"2019-08-14T10:59:15.000Z",
         "Company":"Confluent1010",
         "CleanStatus":"Pending",
         "OwnerId":"0052v00000alfOXAAY",
         "CreatedById":"0052v00000alfOXAAY",
         "IsConverted":false,
         "ChangeEventHeader":{
            "commitNumber":10663250415860,
            "commitUser":"0052v00000alfOXAAY",
            "sequenceNumber":1,
            "entityName":"Lead",
            "changeType":"CREATE",
            "changeOrigin":"com/salesforce/api/rest/46.0",
            "transactionKey":"00025790-0623-3300-2724-bdf92c57a1e7",
            "commitTimestamp":1565780355000,
            "recordIds":[
               "00Q2v00001V4bqkEAB"
            ]
         },
         "IsUnreadByOwner":true,
         "CreatedDate":"2019-08-14T10:59:15.000Z",
         "LastModifiedById":"0052v00000alfOXAAY",
         "Name":{
            "LastName":"ConfluentLastName101"
         }
      },
      "event":{
         "replayId":1109257
      }
   },
   "channel":"/data/LeadChangeEvent"
}
```

### Kafka Record Value

The following is a sample record written to Kafka topic for a Change Data Capture event.

```json
{
    "Id":"00Q2v00001WOn3zEAD",
    "ReplayId":"1115950",
    "ChangeEventHeader":{
       "entityName":"Lead",
       "recordIds":["00Q2v00001WOn3zEAD"],
       "changeType":"CREATE",
       "changeOrigin":"com/salesforce/api/rest/46.0",
       "transactionKey":"00097908-d9b2-d840-5501-ae933a58ee4b",
       "sequenceNumber":1,
       "commitTimestamp":1568628454000,
       "commitUser":"0052v00000alfOXAAY",
       "commitNumber":10683824131775
    },
    "LastName":"ConfluentLastName89",
    "FirstName":null,
    "Salutation":null,
    "Name":null,
    "Title":null,
    "Company":"Confluent89",
    "City":null,
    "State":null,
    "PostalCode":null,
    "Country":null,
    "Latitude":null,
    "Longitude":null,
    "GeocodeAccuracy":null,
    "Address":null,
    "Phone":null,
    "MobilePhone":null,
    "Fax":null,
    "Email":null,
    "Website":null,
    "LeadSource":null,
    "Status":"Open - Not Contacted",
    "Industry":null,
    "Rating":null,
    "AnnualRevenue":null,
    "NumberOfEmployees":null,
    "OwnerId":"0052v00000alfOXAAY",
    "IsConverted":false,
    "ConvertedDate":null,
    "ConvertedAccountId":null,
    "ConvertedContactId":null,
    "ConvertedOpportunityId":null,
    "IsUnreadByOwner":true,
    "CreatedDate":1568628454000,
    "CreatedById":"0052v00000alfOXAAY",
    "LastModifiedDate":1568628454000,
    "LastModifiedById":"0052v00000alfOXAAY",
    "Jigsaw":null,
    "JigsawContactId":null,
    "CleanStatus":"Pending",
    "CompanyDunsNumber":null,
    "DandbCompanyId":null,
    "EmailBouncedReason":null,
    "EmailBouncedDate":null,
    "IndividualId":null,
    "SICCode__c":null,
    "ProductInterest__c":null,
    "Primary__c":null,
    "CurrentGenerators__c":null,
    "NumberofLocations__c":null,
    "_ObjectType":"LeadChangeEvent",
    "_EventType":"MbZFVarhwdRnx4cf-8DowQ"
}
```

#### NOTE
The Salesforce record fields unaltered in a Change Data Capture event are expected to have value as `null` when written to a Kafka topic.

<a id="salesforce-cdc-source-connector-troubleshooting"></a>

## Troubleshooting Connector and Task Failures

You can use the [Kafka Connect REST API](/platform/current/connect/references/restapi.html) to check
the status of the connectors and tasks. If a task or connector has failed, the
`trace` field will include a reason and a stack trace.

**Authorization Failures**

The Salesforce connector must authenticate with Salesforce and establish a
connection. If a connection fails because of authentication or authorization
errors, the connector will stop immediately. These errors require changes in
your Salesforce account which may include creating OAuth tokens. Try to rerun
your connector after you make the account changes.

**Salesforce Error Codes**

Whenever the connector or task fails, it captures a message that includes the
Salesforce error code and  error message. The message may also include a failure
reason or suggested fix. See the [Salesforce error codes](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/streaming_error_codes.htm) for more detail about each error code.

**Repeated Connection Failures**

The Salesforce connector uses a long-lived connection to the Salesforce service.
Periodically this connection is lost due to network lag, service disruptions,
or network issues. When the connector loses its connection, it automatically
attempts  to reconnect so that the connector can continue working. In some
cases, however, the [error reported by Salesforce](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/streaming_error_codes.htm) indicates that the connector is unable to continue. In these cases, the connector task  fails and includes the Salesforce error and reason in the task failure message.

If the connector stops frequently because of connection timeouts, consider
changing the following connector configuration properties and restarting the
connector:

1. The `​connection.timeout` defaults to 30 seconds (30000 milliseconds) and dictates the maximum amount of time that the connector should wait for a single connection attempt to succeed. If the initial connection takes longer than this, the connector task will fail.
2. The `​request.max.retries.time.ms` defaults to 15 minutes (900000 milliseconds) and specifies the maximum amount of time that the connector should continue to retry a failed request.
   The actual duration of the delay is random and it grows exponentially with each retry.
   If all retries take longer than this value, the connector task will fail.
   This does not apply to initial connection attempts, but it does apply to subsequent requests to reconnect.
   Any requests that result in authorization or authentication failures will not be retried.

If the failure rate is not affected by tuning these properties, it may be necessary to look at the Connect worker logs
for errors, warnings, and exceptions. The log messages typically have a more detail than the connector status.
The log messages preceding failure log messages can provide additional context and may provide you with information that identifies the failure cause.

## Enabling Debug Logging

The Connect worker log configuration controls how much detail is included in the logs.
By default, the worker logs include enough detail to identify basic functionality.
Enable DEBUG logs in the Connect worker’s log configuration to include more details.
This change must be made on each worker and only takes effect upon worker startup.
After you change the log configuration as outlined below on each Connect worker,
restart all of the Connect workers. A rolling restart can be used if necessary.

#### NOTE
Trace level logging is verbose and contains many more details,  and may be
useful to solve certain failures. Trace level logging is enabled like debug
level logging is enabled, except `TRACE` is used instead of `DEBUG`.

**On-Premises Installation**

For local or on-premises installations of Confluent Platform, the `etc/kafka/connect-log4j2.yaml` file defines the logging configuration of the Connect worker process. To enable DEBUG on just the Salesforce connector,
modify the `etc/kafka/connect-log4j2.yaml` file to include the following configuration:

```yaml
Loggers:
  Logger:
    - name: io.confluent.salesforce
      level: DEBUG
```

To enable DEBUG on all of the Connect worker’s code, including all connectors, change the root logger configuration. For example, the default log configuration for Connect includes this configuration:

```yaml
Root:
  level: INFO
```

Change this configuration to the following to enable DEBUG on all of the Connect worker code:

```yaml
Root:
  level: DEBUG
```

**Docker**

For Docker containers, the logging configuration is set using environment variables. To enable DEBUG on just the Salesforce connector, use the following environment variable when starting your Confluent Platform Connect container:

```bash
CONNECT_LOG4J2_ROOT_LOGLEVEL=DEBUG
```

To enable DEBUG log messages on all Connect worker code, including all connectors, use the following environment variable when starting your Confluent Platform Connect container:

```bash
CONNECT_LOG4J2_LOGGERS="io.confluent.salesforce=DEBUG"
```

The value of this environment variable is a comma-separated list of key-value pairs. For example, the following enables DEBUG on the Salesforce connector and the Connect framework:

```bash
CONNECT_LOG4J2_LOGGERS="io.confluent.salesforce=DEBUG,org.apache.kafka.connect=DEBUG"
```