# Salesforce CDC Source Connector for Confluent Cloud The fully managed Salesforce Change Data Capture (CDC) Source connector for Confluent Cloud captures create, update, delete, and undelete change events from Salesforce records and writes them to Apache Kafka® topics. Salesforce sends a notification when a change occurs, and the connector captures these events in real time. #### NOTE * This Quick Start is for the fully managed Confluent Cloud connector. If you are installing the connector locally for Confluent Platform, see [Salesforce Change Data Capture Source Connector for Confluent Platform](https://docs.confluent.io/kafka-connectors/salesforce/current/). * If you require private networking for fully managed connectors, make sure to set up the proper networking beforehand. For more information, see [Manage Networking for Confluent Cloud Connectors](networking/internet-resource.md#clusters-connect-cloud). * The connector supports Salesforce up to API version 65.0. ## Features The Salesforce CDC Source connector provides the following features: * **Salesforce Streaming API**: This connector uses the [Salesforce Streaming API (Change Data Capture)](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm). Changes captured include new records, updates to existing records, record deletions, and record undeletions. * **Support for single entity channels**: The connector supports single entity channels like the LeadChangeEvent channel. * **Support for multiple entity channels**: The connector supports multiple entity channels like the [ChangeEvents Standard Channel](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_subscribe_channels.htm) or a [Custom Channel](https://developer.salesforce.com/docs/atlas.en-us.238.0.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm) like LeadCustom_\_chn. * **Initial start**: Captures the latest changes *or* all changes over the last 72 hours. * **Data formats:** The connector supports Avro, JSON Schema, Protobuf, JSON (schemaless), or SF_API output data. In SF_API format the record is formatted identically to the Salesforce message received by the connector and the messages are ingested as raw bytes without any schema. [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). * **Topics created automatically**: The connector can automatically create Kafka topics. When using multiple entity channels with the connector, you can add `${_ObjectType}` to the topic name to create different topic names based on the entity name. * **Tasks per connector**: Organizations can run multiple connectors with a limit of one task per connector (that is, `"tasks.max": "1"`). * **Offset management capabilities**: Supports offset management. For more information, see [Manage custom offsets](#cc-salesforce-source-cdc-custom-offsets). * **Client-side encryption (CSFLE and CSPE) support**: The connector supports CSFLE and CSPE for sensitive data. For more information about CSFLE or CSPE setup, see the [connector configuration](#cc-salesforce-cdc-source-setup-connection). * **Supports Salesforce enriched events** : The connector supports enriching Salesforce 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. * **Supports Client Credentials flow**: The connector supports authentication using the Client Credentials flow that enables connecting to Salesforce without exposing the user credentials. To use `CLIENT_CREDENTIALS` grant type, you must enable the Client Credentials flow in your connected Salesforce application and assign an integration user. * **Supports OAuth 2.0 Authorization Code grant flow (BYOA)**: The connector supports authentication using the OAuth 2.0 Authorization Code grant flow with your own connected application or external client application (bring your own app), which helps you connect to Salesforce without exposing user credentials. To use the `OAUTH2_AUTH_CODE_BYOA` grant type, provide your application’s consumer key and consumer secret, then complete the authorization by clicking **Connect with Salesforce**. A shared application is not supported. * **Supports Salesforce External Client Apps (ECA)**: This connector supports Salesforce External Client Apps when using the `JWT_BEARER`, `OAUTH2_AUTH_CODE_BYOA`, or `CLIENT_CREDENTIALS` grant type. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Limitations Be sure to review the following information. * For connector limitations, see [Salesforce CDC Source Connector](limits.md#cc-salesforce-source-cdc-limits) limitations. * If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](single-message-transforms.md#cc-single-message-transforms-limitations). ## Manage custom offsets You can manage the offsets for this connector. Offsets provide information on the point in the system from which the connector is accessing data. For more information, see [Manage Offsets for Fully Managed Connectors in Confluent Cloud](offsets.md#connect-custom-offsets). **To manage offsets**: - Manage offsets using Confluent Cloud APIs. For more information, see [Connect offsets API reference](https://docs.confluent.io/cloud/current/ccloud/offsets-connect-v-1/). ### Get the current offset To get the current offset, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets Host: https://api.confluent.cloud ``` **Response:** Successful calls return HTTP `200` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": {}, "offset": { "replayId": 75314157 } } ], "metadata": { "observed_at": "2024-03-28T17:57:48.139635200Z" } } ``` Responses include the following information: - The position of latest offset - represented by `replayId`. - The observed time of the offset in the metadata portion of the payload. The `observed_at` time indicates a snapshot in time for when the API retrieved the offset. A running connector is always updating its offsets. Use `observed_at` to get a sense for the gap between real time and the time at which the request was made. By default, offsets are observed every minute. Calling get repeatedly will fetch more recently observed offsets. - Information about the connector. ### Update the offset To update the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies new offset and a patch type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "PATCH", "offsets": [ { "partition": {}, "offset": { "replayId": 75314147 } } ] } ``` Considerations: - You can only make one offset change at a time for a given connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - For source connectors, the connector attempts to read from the position defined by the requested offsets. - `replayID` is equal to the ReplayID value for an event. For example, assume the ReplayID for an event is `1234`. The connector would start consuming all the events after the event with ReplayID `1234`. - `replayID` is available in each Kafka record. - To consume all events within retention window, set `replayID` to `-2`. Salesforce describes the behavior of this option as subscribers receive all events, including past events that are within the retention window and new events. This means that the connector resets the offsets to pick all the events that are within the 72-hour Salesforce retention window and new events. For more information, see [Message Durability](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/using_streaming_api_durability.htm) in the Salesforce documentation. - Events outside the Salesforce retention period (72 hours) are discarded. **Response:** Successful calls return HTTP `202 Accepted` with a JSON payload that describes the offset. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": {}, "offset": { "replayId": 75314147 } } ], "requested_at": "2024-03-28T17:58:45.606796307Z", "type": "PATCH" } ``` Responses include the following information: - The requested position of the offsets in the source. - The time of the request to update the offset. - Information about the connector. ### Delete the offset To delete the offset, make a `POST` request that specifies the environment, Kafka cluster, and connector name. Include a JSON payload that specifies the delete type. ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request Host: https://api.confluent.cloud { "type": "DELETE" } ``` Considerations: - Delete requests delete the offset for the provided partition and reset to the base state. A delete request is as if you created a fresh new connector. - This is an asynchronous request. To check the status of this request, you must use the check offset status API. For more information, see **Get the status of an offset request**. - Do not issue delete and patch requests at the same time. - For source connectors, the connector attempts to read from the position defined in the base state. **Response**: Successful calls return HTTP `202 Accepted` with a JSON payload that describes the result. ```bash { "id": "lcc-example123", "name": "{connector_name}", "offsets": [], "requested_at": "2024-03-28T17:59:45.606796307Z", "type": "DELETE" } ``` Responses include the following information: - Empty offsets. - The time of the request to delete the offset. - Information about Kafka cluster and connector. - The type of request. ### Get the status of an offset request To get the status of a previous offset request, make a `GET` request that specifies the environment, Kafka cluster, and connector name. ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request/status Host: https://api.confluent.cloud ``` Considerations: - The status endpoint always shows the status of the most recent PATCH/DELETE operation. **Response**: Successful calls return HTTP `200` with a JSON payload that describes the result. The following is an example of an applied patch. ```bash { "request": { "id": "lcc-example123", "name": "{connector_name}", "offsets": [ { "partition": {}, "offset": { "replayId": 75314150 } } ], "requested_at": "2024-03-28T17:58:45.606796307Z", "type": "PATCH" }, "status": { "phase": "APPLIED", "message": "The Connect framework-managed offsets for this connector have been altered successfully. However, if this connector manages offsets externally, they will need to be manually altered in the system that the connector uses." }, "previous_offsets": [ { "partition": {}, "offset": { "replayId": 75314147 } } ], "applied_at": "2024-03-28T17:58:48.079141883Z" } ``` Responses include the following information: - The original request, including the time it was made. - The status of the request: applied, pending, or failed. - The time you issued the status request. - The previous offsets. These are the offsets that the connector last updated prior to updating the offsets. Use these to try to restore the state of your connector if a patch update causes your connector to fail or to return a connector to its previous state after rolling back. ### JSON payload The table below offers a description of the unique fields in the JSON payload for managing offsets of the Salesforce CDC Source connector. | Field | Definition | Required/Optional | |------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|---------------------| | `replayId` | The ReplayId field value, which is populated by the Salesforce system refers to the position of the event in the
event stream. Replay ID values are not guaranteed to be contiguous for consecutive events. For more information,
see [Message Durability](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/using_streaming_api_durability.htm) in the Salesforce documentation. | Required | ## Set up Salesforce authentication #### IMPORTANT In accordance with the Salesforce Winter ‘27 release, Confluent will deprecate the OAuth 2.0 username-password flow (`salesforce.grant.type=PASSWORD`) for Salesforce connectors effective September 15, 2026. Update your connector configurations to use another supported auth grant type. Salesforce connectors authenticate with Salesforce through a connected application or External Client App (ECA). A connected application is a metadata object in Salesforce that defines the protocol (OAuth), the permissions (scopes), and the security policies (for example, IP relaxation and user authorization) for an external application. The consumer key (client ID) and consumer secret (client secret) that you enter in the connector configuration are the credentials of this connected application. The connector supports the following OAuth 2.0 grant types. You select the grant type using the `salesforce.grant.type` configuration property. - `CLIENT_CREDENTIALS`: The connector authenticates with only the consumer key and consumer secret of the connected application. User credentials are not transmitted. - `JWT_BEARER`: The connector authenticates with a JSON Web Token (JWT) signed with a private key. - `OAUTH2_AUTH_CODE_BYOA`: The connector uses the OAuth 2.0 authorization code flow with an application you create and own in your Salesforce org (bring your own application, or BYOA). Confluent Cloud stores a refresh token issued during a one-time browser authorization and uses it to obtain access tokens at runtime. ### Client Credentials flow The OAuth 2.0 Client Credentials flow enables the connector to connect to Salesforce without exposing user credentials. Salesforce requires an execution user (the **Run As** user) for this flow. If no Run As user is defined in the connected application policies, authentication fails with an `invalid_grant` or `no client credentials user enabled` error. #### Configure Salesforce Complete the following steps in Salesforce before you create the connector. 1. Enable the client credentials flow: 1. If you don’t already have a connected application for the connector, enable connected app creation first. Navigate to **Setup > External Client App Settings** and enable **Allow creation of connected apps**. This setting is hidden by default in new Salesforce organizations. 2. Navigate to **Setup > App Manager**. 3. Locate the connected application used for the connector, click the drop-down arrow, and select **Edit**. 4. Under **API (Enable OAuth Settings)**, select **Enable Client Credentials Flow**. This option is often disabled by default for older applications. 5. Save the application definition. 2. Assign the execution user: 1. Navigate to **Setup > Manage Connected Apps**, which is distinct from **App Manager**. 2. Click the name of the connected application. 3. Click **Edit Policies**. 4. Scroll to the **Client Credentials Flow** section. 5. In **Run As**, click the search icon and select the dedicated integration user (for example, `integration.user@example.com`). The Run As user must have the **API Enabled** permission and enough rights to access the objects the connector reads from or writes to. #### Configure the connector Set the following authentication properties in the connector configuration. | Configuration property | Value | Notes | |------------------------------|-----------------------------------------------|--------------------------------------------------------------------------------------------------------------------------| | `salesforce.grant.type` | `CLIENT_CREDENTIALS` | Select this grant type explicitly. | | `salesforce.consumer.key` | Consumer key of the connected application. | Also called the client ID. | | `salesforce.consumer.secret` | Consumer secret of the connected application. | Also called the client secret. | | `salesforce.instance` | `https://.my.salesforce.com` | You must use your My Domain URL. The default value
`https://login.salesforce.com` does not work for this grant type. | | `salesforce.username` | Not used. | The execution user is defined in Salesforce. | | `salesforce.password` | Not used. | The execution user is defined in Salesforce. | The client credentials flow requires your My Domain URL, for example, `https://.my.salesforce.com`, in the `salesforce.instance` property. When a client credentials request is sent to the generic `https://login.salesforce.com` endpoint, the Salesforce global router might fail to identify the tenant-specific policies required to look up the Run As user mapping, and authentication fails. ### JWT Bearer flow The OAuth 2.0 JWT bearer flow establishes trust through public-key cryptography and is the recommended authentication method for production environments. The connector signs a JWT with its private key, identifying the Salesforce user in the `sub` (subject) claim. Salesforce validates the signature using the public certificate uploaded to the connected application and issues an access token. The flow requires no interactive login or password, and is not affected by password expiration policies. #### Configure JWT Bearer Complete the following steps to configure the JWT bearer flow. 1. **Generate a private key and certificate**: Generate a private key and a self-signed certificate using OpenSSL. The following example sets the certificate validity to 3,650 days (10 years) to reduce maintenance overhead. ```bash openssl req -newkey rsa:2048 -nodes -keyout private.key -x509 \ -days 3650 -out public.crt ``` The command generates the following files: - `private.key`: The private key the connector uses to sign the JWT. - `public.crt`: The public certificate you upload to Salesforce. 2. **Create the Java keystore (JKS)**: The connector requires the private key in a JKS container and cannot read the raw PEM files directly. 1. Convert the certificate and private key to PKCS12 format: ```bash openssl pkcs12 -export -in public.crt -inkey private.key \ -out keystore.p12 -name salesforce-cert ``` You are prompted to create an export password. This password becomes the keystore password used in the connector configuration. 2. Import the PKCS12 file into a JKS keystore: ```bash keytool -importkeystore -srckeystore keystore.p12 \ -srcstoretype pkcs12 -destkeystore salesforce.jks \ -deststoretype JKS ``` Upload the resulting `salesforce.jks` file when you configure the connector. #### Configure the Salesforce connected application Complete the following steps in Salesforce before you create the connector. 1. If you don’t already have a connected application for the connector, enable connected app creation first. Navigate to **Setup > External Client App Settings** and enable **Allow creation of connected apps**. This setting is hidden by default in new Salesforce organizations. 2. Navigate to **Setup > App Manager** and edit the connected application used for the connector. 3. Under **API (Enable OAuth Settings)**, select **Use digital signatures**, click **Choose File**, and upload the `public.crt` file. 4. Ensure the `api`, `refresh_token`, and `offline_access` OAuth scopes are selected. 5. Pre-authorize users. Navigate to **Setup > Manage Connected Apps**, click **Edit Policies**, and change **Permitted Users** to **Admin approved users are pre-authorized**. The JWT Bearer flow is automated, and the connector cannot approve a consent screen, so you must be pre-authorized. 6. Assign profiles. In **Manage Connected Apps**, scroll to **Profiles** (or **Permission Sets**) and add the profile of the integration user, for example, a custom integration profile. #### Configure the connector Set the following authentication properties in the connector configuration. | Configuration property | Value | Notes | |------------------------------------|--------------------------------------------|----------------------------------------------------------------------------------------------| | `salesforce.grant.type` | `JWT_BEARER` | Select this grant type explicitly. | | `salesforce.consumer.key` | Consumer key of the connected application. | Also called the client ID. | | `salesforce.username` | Username of the integration user | Sets the `sub` claim of the JWT. The username must match a user
in the assigned profile. | | `salesforce.jwt.keystore.file` | The `salesforce.jks` file | Upload the file when you configure the connector. | | `salesforce.jwt.keystore.password` | Keystore password | The export password set during the PKCS12 conversion. | | `salesforce.instance` | `https://.my.salesforce.com` | Use your My Domain URL for best results. | | `salesforce.password` | Not used | Not required for this grant type. | | `salesforce.consumer.secret` | Not used | The signed JWT replaces the consumer secret. | ### Authorization Code flow (Bring your own application) The OAuth 2.0 authorization code flow (also called the web server flow) lets the connector access Salesforce on behalf of a user without storing the user’s password. In the bring your own application (BYOA) model, you create and own the application in your Salesforce org, authorize the connector once in a browser, and Salesforce issues a refresh token that Confluent Cloud stores and uses to obtain access tokens automatically. As of Salesforce Spring ‘26, create an ECA instead of a new connected application; existing connected applications continue to work. #### Configure the Salesforce application Complete the following steps in Salesforce before you create the connector. 1. Enable OAuth and set the callback URL: 1. Navigate to **Setup > App Manager** to edit an existing connected application, or **Setup > External Client App Manager > New External Client App** to create a new application. 2. Under **API (Enable OAuth Settings)**, select **Enable OAuth Settings**. 3. In **Callback URL**, enter `https://confluent.cloud/api/connect/oauth/callback`. 2. Select the OAuth scopes: 1. **Manage user data via APIs** (`api`). 2. **Perform requests at any time** (`refresh_token`, `offline_access`). 3. Enable the web server flow. For an ECA, under **Flow Enablement**, select **Enable Authorization Code and Credentials Flow**. 4. Set the OAuth security policies: 1. Select **Require Secret for Web Server Flow**. 2. Select **Require Secret for Refresh Token Flow**. 3. Disable **Require Proof Key for Code Exchange (PKCE)**. 4. Disable **Enable Refresh Token Rotation**. Rotation would invalidate the token and the connector would require manual re-authorization. 5. Configure the refresh token policy so the token does not expire immediately. Preferably set the refresh token policy to **Valid Until Revoked** to avoid manual re-authorization on token expiry. #### Configure the connector Set the following authentication properties in the connector configuration. | Configuration property | Value | Notes | |------------------------------|---------------------------------------------------------------------------------------------|------------------------------------------------------------------------------| | `salesforce.grant.type` | `OAUTH2_AUTH_CODE_BYOA` | Select this grant type explicitly. | | `salesforce.consumer.key` | Consumer key of the application. | Also called the client ID. | | `salesforce.consumer.secret` | Consumer secret of the application. | Also called the client secret. | | `salesforce.instance` | `https://login.salesforce.com` (production) or
`https://test.salesforce.com` (sandbox). | Using your My Domain URL is recommended, but the default URLs
also work. | #### Authorize the connector Click **Connect with Salesforce**, then complete the Salesforce login and consent prompt. Salesforce redirects back to Confluent Cloud, which stores the refresh token. You can then continue and launch the connector. ### Troubleshoot authentication errors The following table lists common authentication errors and how to resolve them. | Error | Grant type | Probable cause | Resolution | |-------------------------------------------------------------------------------------------------|-------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `invalid_grant` | `CLIENT_CREDENTIALS` | No Run As user is defined for the connected application. | Define the execution user in the connected application policies. | | `invalid_grant` | `CLIENT_CREDENTIALS` | Wrong instance URL. | Set `salesforce.instance` to your My Domain URL. | | `invalid_app_access` | `CLIENT_CREDENTIALS` | The user is not admin-approved to access the connected application. | In **Manage Connected Apps**, open the connected application, click
**Manage Profiles**, and add the profile of the integration user. | | `invalid_client` | All grant types | Wrong consumer key or consumer secret. | Verify the consumer key and consumer secret. Check for trailing
spaces. | | `invalid_grant: audience is invalid` | `JWT_BEARER` | Wrong audience in the JWT claim. | Ensure `salesforce.instance` matches the audience Salesforce
expects, which is typically your My Domain URL. | | `ConnectException: Connection refused` | All grant types | The network blocks the connection. | Check firewall and proxy settings, and verify that the Salesforce
URL is reachable. | | `400 Bad Request` | `CLIENT_CREDENTIALS` | The Client Credentials flow is disabled for the connected application. | In **App Manager**, edit the connected application and select **Enable
Client Credentials Flow**. | | `Job creation failed` | All grant types | The integration user has insufficient permissions. | Grant the integration user the **API Enabled** permission and
create and read permissions on the target objects. | | `invalid_grant` (“expired access/refresh token”) | `OAUTH2_AUTH_CODE_BYOA` | The refresh token expired, was revoked, or was rotated. | Manually re-authorize with **Connect with Salesforce**, and
preferably set the refresh token policy to **Valid Until Revoked**
and disable refresh token rotation. | | `invalid_grant` (“expired access/refresh token”) on a connector
that was previously working | `OAUTH2_AUTH_CODE_BYOA` | Salesforce allows only five unique OAuth approvals per user per
connected application. Each refresh token counts as an approval, and a new
approval beyond the limit revokes the oldest. Sharing one
Salesforce user and connected application across multiple connectors (or
re-authorizing repeatedly) exhausts this limit and revokes an
older connector’s refresh token. | Use a dedicated Salesforce user for each connector (a shared
connected application is fine as long as the user differs), and avoid
unnecessary re-authorizations. Then re-authorize the affected
connector with **Connect with Salesforce**. For more information,
see Salesforce’s [Manage OAuth-Enabled Connected Apps’ Access](https://help.salesforce.com/s/articleView?id=xcloud.remoteaccess_request_manage.htm&type=5)
documentation. | ## Quick Start Use this quick start to get up and running with the Salesforce CDC Source connector. The quick start provides the basics of selecting the connector and configuring it to monitor changes. Prerequisites : - Authorized access to a [Confluent Cloud](https://www.confluent.io/confluent-cloud/) cluster on Amazon Web Services (AWS), Microsoft Azure (Azure), or Google Cloud. - The Confluent CLI installed and configured for the cluster. See [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). - [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) must be enabled to use a Schema Registry-based format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). - Salesforce must be configured for CDC. See the [Salesforce Change Data Capture Developer Guide](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_select_objects.htm). - 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). - For networking considerations, see [Networking and DNS](overview.md#connect-internet-access-resources). To use a set of public egress IP addresses, see [Public Egress IP Addresses for Confluent Cloud Connectors](static-egress-ip.md#cc-static-egress-ips). - The connector uses the Salesforce SOAP client for multiple entity channels. Before using multiple entity channels, you must enable Salesforce [SOAP APIs](https://developer.salesforce.com/docs/atlas.en-us.api.meta/api/sforce_api_quickstart_intro.htm) using the **API Enabled** option in the Salesforce account. Be sure your organization’s firewall rules (if applicable) allow the connector to communicate with the SOAP client. - 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.
- Kafka cluster credentials. The following lists the different ways you can provide credentials. - Enter an existing [service account](service-account.md#s3-cloud-service-account) resource ID. - Create a Confluent Cloud [service account](service-account.md#s3-cloud-service-account) for the connector. Make sure to review the ACL entries required in the [service account documentation](service-account.md#s3-cloud-service-account). Some connectors have specific ACL requirements. - Create a Confluent Cloud API key and secret. To create a key and secret, you can use [confluent api-key create](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) *or* you can autogenerate the API key and secret directly in the Cloud Console when setting up the connector. ### Using the Confluent Cloud Console #### Step 1: Launch your Confluent Cloud cluster To create and launch a Kafka cluster in Confluent Cloud, see [Create a kafka cluster in Confluent Cloud](../get-started/index.md#cloud-create-kafka-cluster). #### Step 2: Add a connector In the left navigation menu, click **Connectors**. If you already have connectors in your cluster, click **+ Add connector**. #### Step 3: Select your connector Click the **Salesforce CDC Source** connector card. ![Salesforce CDC Source Connector Card](images/ccloud-salesforce-cdc-source-icon.png) #### Step 4: Enter the connector details #### NOTE * Make sure you have all your [prerequisites](#cc-salesforce-cdc-source-prereqs) completed. * An asterisk ( \* ) designates a required entry. At the **Add Salesforce CDC Source Connector** screen, complete the following: ### Topic selection Enter a topic name. The connector can automatically create Kafka topics. When using multiple entity channels (`MULTI`) with the connector, you can add `${_ObjectType}` to the topic name to create different topic names based on the entity name. ### Kafka access 1. Select the way you want to provide **Kafka Cluster credentials**. You can choose one of the following options: - **My account**: This setting allows your connector to globally access everything that you have access to. With a user account, the connector uses an API key and secret to access the Kafka cluster. This option is not recommended for production. - **Service account**: This setting limits the access for your connector by using a [service account](service-account.md#s3-cloud-service-account). This option is recommended for production. - **Use an existing API key**: This setting allows you to specify an API key and a secret pair. You can use an existing pair or create a new one. This method is not recommended for production environments. #### NOTE Freight clusters support only service accounts for Kafka authentication. 2. Click **Continue**. ### Authentication 1. Configure the authentication properties: - **Salesforce grant type**: Sets the authentication grant type to `PASSWORD` , `JWT_BEARER` ([Salesforce JSON Web Token (JWT)](https://help.salesforce.com/s/articleView?id=sf.remoteaccess_oauth_jwt_flow.htm&type=5)) or `CLIENT_CREDENTIALS`. Defaults to `PASSWORD`. **Salesforce details** - **Salesforce instance**: The URL of the Salesforce endpoint to use. The default is [https://login.salesforce.com](https://login.salesforce.com). This directs the connector to use the endpoint specified in the authentication response. - **Salesforce username**: The Salesforce username for the connector to use. - **Salesforce password**: The Salesforce password for the connector to use. - **Salesforce password token**: The Salesforce security token associated with the username. - **Salesforce consumer key**: The consumer key for the OAuth application. - **Salesforce consumer secret**: The consumer secret for the OAuth application. - **Salesforce JWT keystore file**: If using the grant type `JWT_BEARER`, upload the JWT keystore file. - **Salesforce JWT keystore password**: The password used to access the JWT keystore file. #### NOTE The following properties are used based on the **Salesforce grant type** you choose. - `JWT_BEARER`: Requires username, consumer key, JWT keystore file, and JWT keystore password. - `PASSWORD`: Requires username, password, password token, consumer key, and consumer secret. - `CLIENT_CREDENTIALS`: Requires the client ID and client secret of a Salesforce connected application as the consumer key and consumer secret, and the Salesforce domain URL in the Salesforce instance option. The default value [https://login.salesforce.com](https://login.salesforce.com) does not work for this option. To use `CLIENT_CREDENTIALS`, you must enable the Client Credentials flow in your connected Salesforce application and assign an integration user. - `OAUTH2_AUTH_CODE_BYOA`: Requires consumer key and consumer secret (the consumer key and consumer secret of your Salesforce connected application or external client application). After you enter these values, complete the authorization by clicking **Connect with Salesforce** to perform the OAuth 2.0 handshake and store the refresh token. Only bring your own application (BYOA) is supported. A shared application is not supported. Confluent recommends using the My Domain URL, although the default URLs also work. 2. Click **Continue**. ### Configuration - **Select output record value format**: Sets the output Kafka record value format. Valid values are AVRO, JSON_SR, PROTOBUF, JSON, or SF_API. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF. When SF_API is selected, the record is identical in format to the Salesforce message as received by the connector. Note that in SF_API, messages are ingested as raw bytes without any schema. - **Salesforce Channel Type**: The type of Salesforce CDC channel from which the connector consumes events. The value can be `SINGLE` or `MULTI`. Use `SINGLE` for a single entity channel like `LeadChangeEvent`. Use `MULTI` for the Standard (ChangeEvents) channel or a Custom channel like `LeadCustom__chn`. - **Salesforce Channel Entities List**: The comma-separated list of entities in the standard or custom channel. For example, LeadChangeEvent, AccountChangeEvent. **Salesforce details** - **Salesforce CDC name**: The Salesforce Change Data Capture event name to subscribe to. **Data encryption** - Enable **Client-Side Field Level Encryption** for data encryption. Specify a **Service Account** to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](csfle.md#connect-csfle). ### **Show advanced configurations** - **Schema context**: Select a schema context to use for this connector, if using a schema-based data format. This property defaults to the **Default** context, which configures the connector to use the default schema set up for Schema Registry in your Confluent Cloud environment. A schema context allows you to use separate schemas (like schema sub-registries) tied to topics in different Kafka clusters that share the same Schema Registry environment. For example, if you select a non-default context, a **Source** connector uses only that schema context to register a schema and a **Sink** connector uses only that schema context to read from. For more information about setting up a schema context, see [What are schema contexts and when should you use them?](../sr/faqs-cc.md#faq-schema-contexts). **Additional Configs** - **value.converter.replace.null.with.default**: Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter. - **Value Converter Reference Subject Name Strategy**: Sets the subject reference name strategy for values. Valid entries are `DefaultReferenceSubjectNameStrategy` or `QualifiedReferenceSubjectNameStrategy`. You can use this strategy only with `PROTOBUF` format; the default strategy is `DefaultReferenceSubjectNameStrategy`. - **value.converter.schemas.enable**: Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter. - **errors.tolerance**: Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records. - **value.converter.ignore.default.for.nullables**: When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters. - **Value Converter Decimal Format**: Specifies the `JSON` or `JSON_SR` serialization format for Connect `DECIMAL` logical type values with two allowed literals: `BASE64` to serialize `DECIMAL` logical types as base64 encoded binary data, and `NUMERIC` to serialize `DECIMAL` logical type values in `JSON` or `JSON_SR` as a number representing the decimal value. - **Key Converter Schema ID Serializer**: The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers. - **Value Converter Connect Meta Data**: Enables the Connect converter to add its metadata to the output schema. Applies to Avro converters. - **Value Converter Value Subject Name Strategy**: Determines how to construct the subject name under which the value schema is registered with Schema Registry. - **Key Converter Key Subject Name Strategy**: Determines how to construct the subject name for key schema registration. - **Value Converter Schema ID Serializer**: The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers. - **invalid.replay.id.behaviour**: Determine if the connector should fallback to fetching all or latest events if invalid or expired replayId is provided. : NOTE: Salesforce only retains events for 24 hours (standard-volume) or 72 hours (high-volume or CDC), and events outside this window are unrecoverable regardless of fallback mode. **Auto-restart policy** - **Enable Connector Auto-restart**: Enables the auto-restart behavior of the connector and its task in the event of user-actionable errors. Defaults to `true`, enabling the connector to automatically restart in case of user-actionable errors. Set this property to `false` to disable auto-restart for failed connectors. If disabled, you must manually restart the connector. **Connection details** - **Salesforce initial start**: Specifies the initial starting point for the connector. Allowed values are `latest` and `all`. The default value is `latest`. - **Max retry time (ms)**: Maximum time in milliseconds until the connector stops retrying failed Salesforce requests. Default is `30000` and minimum is `1000`. - **Connection Max Message Size**: The maximum message size in bytes that is accepted during a poll on the Salesforce streaming endpoint. **Transforms** - **Single Message Transformations**: To add a new SMT, see [Add transforms](single-message-transforms.md#cc-single-message-transforms-ui). For more information about unsupported SMTs, see [Unsupported transformations](single-message-transforms.md#cc-single-message-transforms-unsupported-transforms). **Processing position** - **Set offsets**: Click **Set offsets** to define a specific offset for this connector to begin procession data from. For more information on managing offsets, see [Manage offsets](offsets.md#connect-custom-offsets). For all property values and definitions, see [Configuration Properties](#cc-salesforce-source-config-properties) . - Click **Continue**. ### Sizing Based on the number of topic partitions you select, you will be provided with a recommended number of tasks. 1. To change the number of tasks, use the Range Slider to select the desired number of tasks. 2. Click **Continue**. ### Review and Launch 1. Verify the connection details by previewing the running configuration. 2. After you’ve validated that the properties are configured to your satisfaction, click **Launch**. The status for the connector should go from **Provisioning** to **Running**. #### Step 5: Check the Kafka topic After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ### Using the Confluent CLI Complete the following steps to set up and run the connector using the Confluent CLI. #### NOTE Make sure you have all your [prerequisites](#cc-salesforce-cdc-source-prereqs) completed. #### Step 1: List the available connectors Enter the following command to list available connectors: ```none confluent connect plugin list ``` #### Step 2: List the connector configuration properties Enter the following command to show the connector configuration properties: ```none confluent connect plugin describe ``` The command output shows the required and optional configuration properties. #### Step 3: Create the connector configuration file Create a JSON file that contains the connector configuration properties. The following example shows the required connector properties. ```none { "connector.class": "SalesforceCdcSource", "name": "SalesforceCdcSourceConnector_0", "kafka.auth.mode": "KAFKA_API_KEY", "kafka.api.key": "****************", "kafka.api.secret": "****************************************************************", "kafka.topic": "AccountChangeEvent", "salesforce.grant.type": "PASSWORD", "salesforce.instance": "https://login.salesforce.com", "salesforce.username": "", "salesforce.password": "**************", "salesforce.password.token": "************************", "salesforce.consumer.key": "*************************************************************************************", "salesforce.consumer.secret": "****************************************************************", "salesforce.cdc.name": "AccountChangeEvent", "output.data.format": "JSON", "tasks.max": "1" } ``` Note the following property definitions: * `"connector.class"`: Identifies the connector plugin name. * `"name"`: Sets a name for your new connector. * `"kafka.auth.mode"`: Identifies the connector authentication mode you want to use. There are two options: `SERVICE_ACCOUNT` or `KAFKA_API_KEY` (the default). To use an API key and secret, specify the configuration properties `kafka.api.key` and `kafka.api.secret`, as shown in the example configuration (above). To use a [service account](service-account.md#s3-cloud-service-account), specify the **Resource ID** in the property `kafka.service.account.id=`. To list the available service account resource IDs, use the following command: ```bash confluent iam service-account list ``` For example: ```bash confluent iam service-account list Id | Resource ID | Name | Description +---------+-------------+-------------------+------------------- 123456 | sa-l1r23m | sa-1 | Service account 1 789101 | sa-l4d56p | sa-2 | Service account 2 ``` * `""kafka.topic"`: Enter a Kafka topic name. When using multiple entity channels with the connector, you can add ${_ObjectType} to the topic name to create different topic names based on the entity name. * `"salesforce.grant.type"`: Sets the authentication grant type to `PASSWORD` (username+password) , `JWT_BEARER` ([Salesforce JSON Web Token (JWT)](https://help.salesforce.com/s/articleView?id=sf.remoteaccess_oauth_jwt_flow.htm&type=5)) or `CLIENT_CREDENTIALS`. Defaults to `PASSWORD`. #### NOTE The following properties are used based on the **Salesforce grant type** you choose. - `JWT_BEARER`: Requires username, consumer key, JWT keystore file, and JWT keystore password. - `PASSWORD`: Requires username, password, password token, consumer key, and consumer secret. - `CLIENT_CREDENTIALS`: Requires the client ID and client secret of a Salesforce connected application as the consumer key and consumer secret, and the Salesforce domain URL in the Salesforce instance option. The default value [https://login.salesforce.com](https://login.salesforce.com) does not work for this option. To use `CLIENT_CREDENTIALS`, you must enable the Client Credentials flow in your connected Salesforce application and assign an integration user. - `OAUTH2_AUTH_CODE_BYOA`: Requires consumer key and consumer secret (the consumer key and consumer secret of your Salesforce connected application or external client application). After you enter these values, complete the authorization by clicking **Connect with Salesforce** to perform the OAuth 2.0 handshake and store the refresh token. Only bring your own application (BYOA) is supported. A shared application is not supported. Confluent recommends using the My Domain URL, although the default URLs also work. * `"salesforce.username"`: The Salesforce username for the connector to use. * `"salesforce.password"`: The Salesforce username password. * `"salesforce.password.token"`: The Salesforce security token associated with the username. * `"salesforce.consumer.key"`: The consumer key for the OAuth application. * `"salesforce.consumer.secret"`: The consumer secret for the OAuth application. * `"salesforce.jwt.keystore.file"`: Salesforce JWT keystore file. The JWT keystore file is a binary file and you supply the contents of the file in the property encoded in Base64. To use the `salesforce.jwt.keystore.file` property, encode the keystore contents in Base64, take the encoded string, add the `data:text/plain:base64` prefix, and then use the entire string as the property entry. For example: ```properties "salesforce.jwt.keystore.file" : "data:text/plain;base64,/u3+7QAAAAIAAAACAAAAGY2xpZ...==", "salesforce.jwt.keystore.password" : "", ``` #### NOTE You only need to convert to Base64 when deploying the connector using the Confluent REST API and Confluent CLI. If you update the JKS file directly from the Confluent Cloud Console, Base64 conversion is not required. * `"salesforce.jwt.keystore.password"`: Enter the password used to access the JWT keystore file. * `"salesforce.cdc.name"`: The Salesforce Change Data Capture event name to subscribe to. * `"output.data.format"`: Sets the output Kafka record value format (data coming from the connector). Valid entries are **AVRO**, **JSON_SR**, **PROTOBUF**, **JSON**, or **SF_API**. You must have Confluent Cloud Schema Registry configured if using a schema-based message format (for example, Avro, JSON_SR (JSON Schema), or Protobuf). Note that if you select `SF_API`, records are ingested as raw bytes and the record format is identical to the salesforce message format. * `"tasks.max"`: Enter the number of [tasks](/platform/current/connect/concepts.html#tasks) in use by the connector. Organizations can run multiple connectors with a limit of one task per connector (that is, `"tasks.max": "1"`). #### NOTE To enable CSFLE or CSPE for data encryption, specify the following properties: * `csfle.enabled`: Flag to indicate whether the connector honors CSFLE or CSPE rules. * `sr.service.account.id`: A Service Account to access the Schema Registry and associated encryption rules or keys with that schema. For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](csfle.md#connect-csfle). **SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](single-message-transforms.md#cc-single-message-transforms) documentation. See [Configuration Properties](#cc-salesforce-source-config-properties) for all property values and definitions. #### Step 4: Load the properties file and create the connector Enter the following command to load the configuration and start the connector: ```none confluent connect cluster create --config-file .json ``` For example: ```none confluent connect cluster create --config-file salesforce-cdc-source.json ``` Example output: ```none Created connector SalesforceCdcSourceConnector_0 lcc-ix4dl ``` #### Step 5: Check the connector status Enter the following command to check the connector status: ```none confluent connect cluster list ``` Example output: ```none ID | Name | Status | Type +-----------+----------------------------------+---------+-------+ lcc-ix4dl | SalesforceCdcSourceConnector_0 | RUNNING | source ``` #### Step 6: Check the Kafka topic. After the connector is running, verify that messages are populating your Kafka topic. For more information and examples to use with the Confluent Cloud API for Connect, see the [Confluent Cloud API for Connect Usage Examples](connect-api-section.md#ccloud-connect-api) section. ## Configuration Properties Use the following configuration properties with the fully managed connector. For self-managed connector property definitions and other details, see the connector docs in [Self-managed connectors for Confluent Platform](/platform/current/connect/kafka_connectors.html). ### How should we connect to your data? `name` : Sets a name for your connector.
* Type: string * Valid Values: A string at most 64 characters long * Importance: high ### Kafka Cluster credentials `kafka.auth.mode` : Kafka Authentication mode. It can be one of KAFKA_API_KEY or SERVICE_ACCOUNT. It defaults to KAFKA_API_KEY mode, whenever possible.
* Type: string * Valid Values: SERVICE_ACCOUNT, KAFKA_API_KEY * Importance: high `kafka.api.key` : Kafka API Key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high `kafka.service.account.id` : The Service Account that will be used to generate the API keys to communicate with Kafka Cluster.
* Type: string * Importance: high `kafka.api.secret` : Secret associated with Kafka API key. Required when kafka.auth.mode==KAFKA_API_KEY.
* Type: password * Importance: high ### Which topic do you want to send data to? `kafka.topic` : Identifies the topic name to write the data to.
* Type: string * Importance: high ### Schema Config `schema.context.name` : Add a schema context name. A schema context represents an independent scope in Schema Registry. It is a separate sub-schema tied to topics in different Kafka clusters that share the same Schema Registry instance. If not used, the connector uses the default schema configured for Schema Registry in your Confluent Cloud environment.
* Type: string * Default: default * Importance: medium ### How should we connect to Salesforce? `salesforce.grant.type` : Grant type the connector uses for Salesforce authentication: PASSWORD, CLIENT_CREDENTIALS, JWT_BEARER, or OAUTH2_AUTH_CODE_BYOA (OAuth 2.0 Authorization Code with your own connected app)
* Type: string * Default: PASSWORD * Importance: high `salesforce.instance` : Salesforce endpoint URL the connector uses for API requests. For CLIENT_CREDENTIALS and the OAuth 2.0 Authorization Code grant, set this to your org’s My Domain URL (for example [https://MyDomainName.my.salesforce.com](https://MyDomainName.my.salesforce.com)). Defaults to [https://login.salesforce.com](https://login.salesforce.com).
* Type: string * Default: [https://login.salesforce.com](https://login.salesforce.com) * Importance: high `salesforce.username` : Salesforce username for connector authentication
* Type: string * Importance: high `salesforce.channel.type` : Indicates the type of Salesforce CDC channel from which the connector shall consume the events. The value can be SINGLE or MULTI. SINGLE should be used for a single entity channel like LeadChangeEvent. MULTI should be used for the Standard (ChangeEvents) channel or a Custom channel like LeadCustom_\_chn.
* Type: string * Importance: high `salesforce.password` : Salesforce account password for connector authentication
* Type: password * Importance: high `salesforce.cdc.name` : The Salesforce Change Data Capture event name to subscribe to.
* Type: string * Importance: high `salesforce.password.token` : Security token associated with the Salesforce username
* Type: password * Importance: high `salesforce.consumer.key` : Client ID (consumer key) for the Salesforce connected app
* Type: password * Importance: high `salesforce.channel.entities` : Comma seperated list of entities in the standard or custom channel. Eg LeadChangeEvent, AccountChangeEvent.
* Type: list * Importance: medium `salesforce.consumer.secret` : Client secret (consumer secret) for the Salesforce connected app
* Type: password * Importance: medium `salesforce.jwt.keystore.file` : Keystore file that stores the private key for JWT authentication
* Type: password * Default: [hidden] * Importance: medium `salesforce.jwt.keystore.password` : Password that unlocks the JWT keystore file
* Type: password * Importance: medium `oauth.refresh.token` : OAuth 2.0 refresh token. Populated by Confluent Cloud after authorization.
* Type: password * Importance: high ### Connection details `salesforce.initial.start` : Specify the initial starting point for the connector for replaying events.
* Type: string * Default: latest * Importance: high `connection.timeout` : The amount of time to wait in milliseconds while connecting to the Salesforce streaming endpoint.
* Type: long * Default: 30000 * Importance: low `request.max.retries.time.ms` : Maximum time in milliseconds until the connector stops retrying failed Salesforce requests. Default is 30000 and minimum is 1000.
* Type: long * Default: 30000 (30 seconds) * Valid Values: [1000,…,250000] * Importance: low `connection.max.message.size` : The maximum message size in bytes that is accepted during a long poll on the Salesforce streaming endpoint.
* Type: int * Default: 1048576 * Valid Values: [1048576,…,104857600] * Importance: low ### Output messages `output.data.format` : Sets the output Kafka record value format. Valid entries are AVRO, JSON_SR, PROTOBUF, JSON or SF_API. Note that you need to have Confluent Cloud Schema Registry configured if using a schema-based message format like AVRO, JSON_SR, and PROTOBUF. When SF_API is selected, the record will be identical in format to the salesforce message as received by the connector. Note that in SF_API, messages are ingested as raw bytes without any schema.
* Type: string * Default: JSON * Importance: high `convert.changed.fields` : Whether to convert field names within changed fields section of the ChangeEventHeader to match field names present on the Kafka record.
* Type: boolean * Default: false * Importance: low ### Number of tasks for this connector `tasks.max` : Maximum number of tasks for the connector.
* Type: int * Valid Values: [1,…,1] * Importance: high ### Additional Configs `header.converter` : The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string * Importance: low `producer.override.compression.type` : The compression type for all data generated by the producer. Valid values are none, gzip, snappy, lz4, and zstd.
* Type: string * Importance: low `producer.override.linger.ms` : The producer groups together any records that arrive in between request transmissions into a single batched request. More details can be found in the documentation: [https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms](https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html#linger-ms).
* Type: long * Valid Values: [100,…,1000] * Importance: low `value.converter.allow.optional.map.keys` : Allow optional string map key when converting from Connect Schema to Avro Schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.auto.register.schemas` : Specify if the Serializer should attempt to register the Schema.
* Type: boolean * Importance: low `value.converter.connect.meta.data` : Allow the Connect converter to add its metadata to the output schema. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.avro.schema.support` : Enable enhanced schema support to preserve package information and Enums. Applicable for Avro Converters.
* Type: boolean * Importance: low `value.converter.enhanced.protobuf.schema.support` : Enable enhanced schema support to preserve package information. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.flatten.unions` : Whether to flatten unions (oneofs). Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.index.for.unions` : Whether to generate an index suffix for unions. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.generate.struct.for.nulls` : Whether to generate a struct variable for null values. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.int.for.enums` : Whether to represent enums as integers. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.latest.compatibility.strict` : Verify latest subject version is backward compatible when use.latest.version is true.
* Type: boolean * Importance: low `value.converter.object.additional.properties` : Whether to allow additional properties for object schemas. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.optional.for.nullables` : Whether nullable fields should be specified with an optional label. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.optional.for.proto2` : Whether proto2 optionals are supported. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.scrub.invalid.names` : Whether to scrub invalid names by replacing invalid characters with valid characters. Applicable for Avro and Protobuf Converters.
* Type: boolean * Importance: low `value.converter.use.latest.version` : Use latest version of schema in subject for serialization when auto.register.schemas is false.
* Type: boolean * Importance: low `value.converter.use.optional.for.nonrequired` : Whether to set non-required properties to be optional. Applicable for JSON_SR Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.nullables` : Whether nullable fields should use primitive wrapper messages. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `value.converter.wrapper.for.raw.primitives` : Whether a wrapper message should be interpreted as a raw primitive at root level. Applicable for Protobuf Converters.
* Type: boolean * Importance: low `errors.tolerance` : Use this property if you would like to configure the connector’s error handling behavior. WARNING: This property should be used with CAUTION for SOURCE CONNECTORS as it may lead to dataloss. If you set this property to ‘all’, the connector will not fail on errant records, but will instead log them (and send to DLQ for Sink Connectors) and continue processing. If you set this property to ‘none’, the connector task will fail on errant records.
* Type: string * Default: none * Importance: low `key.converter.key.schema.id.serializer` : The class name of the schema ID serializer for keys. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `key.converter.key.subject.name.strategy` : How to construct the subject name for key schema registration.
* Type: string * Default: TopicNameStrategy * Importance: low `value.converter.decimal.format` : Specify the JSON/JSON_SR serialization format for Connect DECIMAL logical type values with two allowed literals:
BASE64 to serialize DECIMAL logical types as base64 encoded binary data and
NUMERIC to serialize Connect DECIMAL logical type values in JSON/JSON_SR as a number representing the decimal value.
* Type: string * Default: BASE64 * Importance: low `value.converter.flatten.singleton.unions` : Whether to flatten singleton unions. Applicable for Avro and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.ignore.default.for.nullables` : When set to true, this property ensures that the corresponding record in Kafka is NULL, instead of showing the default column value. Applicable for AVRO,PROTOBUF and JSON_SR Converters.
* Type: boolean * Default: false * Importance: low `value.converter.reference.subject.name.strategy` : Set the subject reference name strategy for value. Valid entries are DefaultReferenceSubjectNameStrategy or QualifiedReferenceSubjectNameStrategy. Note that the subject reference name strategy can be selected only for PROTOBUF format with the default strategy being DefaultReferenceSubjectNameStrategy.
* Type: string * Default: DefaultReferenceSubjectNameStrategy * Importance: low `value.converter.replace.null.with.default` : Whether to replace fields that have a default value and that are null to the default value. When set to true, the default value is used, otherwise null is used. Applicable for JSON Converter.
* Type: boolean * Default: true * Importance: low `value.converter.schemas.enable` : Include schemas within each of the serialized values. Input messages must contain schema and payload fields and may not contain additional fields. For plain JSON data, set this to false. Applicable for JSON Converter.
* Type: boolean * Default: false * Importance: low `value.converter.value.schema.id.serializer` : The class name of the schema ID serializer for values. This is used to serialize schema IDs in the message headers.
* Type: string * Default: io.confluent.kafka.serializers.schema.id.PrefixSchemaIdSerializer * Importance: low `value.converter.value.subject.name.strategy` : Determines how to construct the subject name under which the value schema is registered with Schema Registry.
* Type: string * Default: TopicNameStrategy * Importance: low `invalid.replay.id.behaviour` : Determine if the connector should fallback to fetching all or latest events if invalid or expired replayId is provided.
> NOTE: Salesforce only retains events for 24 hours (standard-volume) or 72 hours (high-volume/CDC), and events outside this window are unrecoverable regardless of fallback mode. * Type: string * Default: all * Importance: medium ### Auto-restart policy `auto.restart.on.user.error` : Enable connector to automatically restart on user-actionable errors.
* Type: boolean * Default: true * Importance: medium ## Frequently asked questions Find answers to frequently asked questions about the Salesforce CDC Source connector for Confluent Cloud. ### Authentication and connection #### How do I configure `CLIENT_CREDENTIALS` authentication for the connector? To use the `CLIENT_CREDENTIALS` grant type, you must: 1. Enable the Client Credentials flow in your Salesforce connected application. 2. Assign an integration user as the **Run As** user in the connected application policies. 3. Provide your Salesforce domain URL (My Domain URL) in the `salesforce.instance` configuration. For example, `https://sample.my.salesforce.com`. The default value `https://login.salesforce.com` does not work for `CLIENT_CREDENTIALS`. 4. Configure the connector with: * `salesforce.grant.type`: `CLIENT_CREDENTIALS` * `salesforce.consumer.key`: Client ID of the connected application * `salesforce.consumer.secret`: Client secret of the connected application * `salesforce.instance`: Your Salesforce My Domain URL If you encounter `invalid_grant` or `no client credentials user enabled` errors, verify that the **Run As** user is properly configured in your connected application’s OAuth policies. For more information, see [Salesforce OAuth 2.0 Client Credentials Flow](https://help.salesforce.com/s/articleView?id=sf.remoteaccess_oauth_client_credentials_flow.htm&type=5). #### Why am I getting `INVALID_SESSION_ID` errors with multiple connectors? This error occurs when multiple Salesforce CDC Source connectors (version 2.x or later) use the same `salesforce.username` and one connector logs out. To resolve this issue: **Option 1 (Recommended)**: Use a unique Salesforce user account (`salesforce.username`) for each connector. This prevents session conflicts. **Option 2**: If you cannot use separate user accounts, consider using the `CLIENT_CREDENTIALS` grant type, which uses application credentials instead of user sessions. ### Configuration #### How do I configure the connector for multiple entity channels? To consume from the standard `ChangeEvents` channel or a custom channel with multiple entities: 1. Set `salesforce.channel.type` to `MULTI`. 2. Set `salesforce.cdc.name` to the channel name: * For the standard channel: `ChangeEvents` * For a custom channel: Use the FullName (for example, `MyCustomChannel__chn`) 3. Set `salesforce.channel.entities` to a comma-separated list of change event entities. For example: `LeadChangeEvent,AccountChangeEvent,ContactChangeEvent`. 4. Enable change data capture for the corresponding objects in Salesforce (**Setup > Change Data Capture > select objects**). 5. Optionally, add `${_ObjectType}` to the topic name to create separate topics for each entity. #### NOTE For custom channels, pre-configure the channel in Salesforce and reference it by its `FullName` in the connector configuration. #### How do I configure the connector to use custom channels? To use a custom channel for change data capture: 1. In Salesforce, create a custom channel and add the desired change event entities. Note the channel’s FullName (for example, `MyCustomChannel__chn`). 2. In the connector configuration: * Set `salesforce.channel.type` to `MULTI`. * Set `salesforce.cdc.name` to the custom channel’s FullName (for example, `MyCustomChannel__chn`). * Set `salesforce.channel.entities` to a comma-separated list of change event entities in the channel (for example, `LeadChangeEvent,AccountChangeEvent`). 3. Ensure your Salesforce user has the required permissions: * API Enabled * View Setup and Configuration * View All Data * Read access on the objects For more information, see [Salesforce Custom Channels](https://developer.salesforce.com/docs/atlas.en-us.238.0.platform_events.meta/platform_events/platform_events_subscribe_custom_channels.htm). #### What permissions does the Salesforce user need for the connector? The Salesforce user configured for the connector must have the following permissions: For single entity channels: : * **API Enabled**: Required for all connector operations. * **View All Data**: Required to receive Change Data Capture events. See [Required Permissions for Change Events](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_security_perms.htm). * Read access on the objects being tracked. For multiple entity channels (standard or custom): : * **API Enabled**: Required for all connector operations. * **View Setup and Configuration**: Required for the connector to validate channel configuration using the Tooling API. * **View All Data**: Required to receive Change Data Capture events. * Read access on the objects being tracked. For creating or modifying custom channels: : * **Customize Application**: Required to create custom channels or add/remove entities in a custom channel. Ensure that Change Data Capture is enabled for the objects in Salesforce (**Setup > Change Data Capture**). ### Data ingestion and ReplayID management #### What does the `replayID was invalid or no longer available` warning mean? This warning appears when the connector attempts to consume events using a replayID that Salesforce no longer recognizes. Common causes include: * The replayID is older than the Salesforce retention window (72 hours for Change Data Capture events). * The connector was stopped for longer than the retention period. * The replayID was manually set to an invalid value. When this occurs, the connector automatically falls back based on the `invalid.replay.id.behaviour` configuration: * `all` (default): Resets to replayID `-2` to consume all events within the retention window. * `latest`: Resets to replayID `-1` to consume only new events from the current point forward. No action is required unless you want to change this fallback behavior. To customize it, set the `invalid.replay.id.behaviour` property in your connector configuration. #### How do I manage connector offsets using the replayID? You can manage offsets for the Salesforce CDC Source connector using the Confluent Cloud APIs: To get the current offset: ```bash GET /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets ``` To update the offset: ```bash POST /connect/v1/environments/{environment_id}/clusters/{kafka_cluster_id}/connectors/{connector_name}/offsets/request { "type": "PATCH", "offsets": [ { "partition": {}, "offset": { "replayId": 75314147 } } ] } ``` Key considerations: * The connector consumes all events *after* the specified replayID. For example, if replayID is `1234`, the connector starts from the next event. * To consume all events within the retention window, set replayID to `-2`. * To consume only new events, set replayID to `-1`. * Events outside the 72-hour Salesforce retention window are discarded. * The replayID value is available in each Kafka record. For more information, see [Manage custom offsets](#cc-salesforce-source-cdc-custom-offsets). #### What happens if the connector stops for more than 72 hours? Salesforce retains Change Data Capture events for 72 hours (3 days). If the connector is stopped for longer than this retention period: 1. Events older than 72 hours are discarded by Salesforce and cannot be recovered. 2. When the connector restarts, it attempts to resume from the last recorded replayID. 3. If the replayID is outside the retention window, the connector issues a warning: “The replayID was invalid or no longer available.” 4. The connector automatically falls back based on the `invalid.replay.id.behaviour` configuration: * `all` (default): Consumes all events within the current 72-hour retention window (replayID `-2`). * `latest`: Consumes only new events from the current point forward (replayID `-1`). To avoid data loss, ensure the connector runs continuously or is restarted within the 72-hour retention window. #### NOTE The connector periodically records the replayID of the last event written to Kafka. If the connector stops unexpectedly, some events may be duplicated when it restarts, as it resumes from the last recorded replayID. #### Can the connector handle enriched Change Data Capture events? Yes, the connector supports Salesforce enriched Change Data Capture events. You can enrich CDC events with additional fields on both: * Custom channels (for example, `/data/SalesEvents__chn`) * The standard `/data/ChangeEvents` channel To use enriched events: 1. Configure event enrichment in Salesforce for the entities you want to monitor. See [Salesforce’s CDC event enrichment documentation](https://developer.salesforce.com/docs/atlas.en-us.change_data_capture.meta/change_data_capture/cdc_intro.htm). 2. Configure the connector to consume from the appropriate channel. 3. The connector automatically includes the enriched fields in the Kafka records. The enriched fields are added to the change event payload and are available in the Kafka messages produced by the connector. ### Troubleshooting #### What does `Exception while validating change event channel` mean? This error typically appears in one of these forms: * `400 Bad Request`: No such column `FullName` on entity `PlatformEventChannel`. * sObject type `PlatformEventChannel` is not supported. Common causes and resolutions: 1. **Using the wrong entity names**: For Change Data Capture, you must specify change event entities (for example, `AccountChangeEvent`, `ContactChangeEvent`), not object names (for example, `Account`, `Contact`). Verify that `salesforce.channel.entities` contains the correct change event entity names. 2. **CDC not enabled for objects**: Ensure that Change Data Capture is enabled for the objects in Salesforce (**Setup > Change Data Capture > select objects**). 3. **Missing permissions**: Verify that your Salesforce user has **View Setup and Configuration** and **API Enabled** permissions when using multiple entity channels or custom channels. 4. **Custom channel not properly configured**: If using a custom channel, verify that it is properly created in Salesforce with the correct `FullName` and that the specified entities are added to the channel. #### How does the connector handle Salesforce API errors? The connector has different retry behaviors based on the error code: Errors that trigger retries with exponential backoff: : * `403`: “Organization concurrent user limit exceeded” * `403`: “Organization total events daily limit exceeded” * `403`: “To protect all customers” * `503`: “Server is too busy” Errors that cause the connector to fail immediately: : * `400`: Bad Request (validation errors) * `413`: Request Entity Too Large * `403`: “Unable to create channel dynamically” * `404`: “channel names may not vary only by case” * `404`: “Unknown channel” If you encounter persistent API errors, verify your Salesforce API limits and quotas. Contact Salesforce support if you need to increase limits. The connector retries failed requests for up to 15 minutes (default) using exponential backoff. You can adjust this timeout, but it may cause task failures if exceeded. ## Next Steps For an example that shows fully managed Confluent Cloud connectors in action with Confluent Cloud for Apache Flink, see the [Cloud ETL Demo](/platform/current/tutorials/examples/cloud-etl/docs/index.html). This example also shows how to use Confluent CLI to manage your resources in Confluent Cloud. [![image](images/topology.png)](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)