# Salesforce Bulk API Sink Connector for Confluent Platform The Kafka Connect Salesforce Bulk API Sink connector integrates [Salesforce](https://developer.salesforce.com) with Apache Kafka®. The connector performs CRUD operations (insert, update, delete) on Salesforce SObjects using records available in Kafka topics and writes them to Salesforce. These records can be written into Kafka topic using Salesforce PushTopic Source connector. For example, given two Salesforce.com organizations or Instances: Instance A and Instance B. If a Data Engineer wants to synchronize Salesforce `Lead` objects from A to B, the Data Engineer can configure and deploy the PushTopic Source connector to stream a Salesforce `Lead` from Instance A into a single Kafka topic, while the sink connector may be configured to stream a `Lead` from that topic into Instance B. Depending upon the configuration, all changes to `Lead` SObjects may be synchronized across organizations. This connector can be used with standalone and distributed Connect workers. #### NOTE * The connector’s `salesforce.object` property indicates the name of the SObject to operate on. The structure and format of input messages for the sink connector are identical to the output format of the [Salesforce PushTopic Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/index.html). * The connector supports Salesforce up to API version 64.0. * 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. ## Considerations Note the following when using the Salesforce Bulk API Sink connector. ### Unexpected errors When the connector is performing operations on Salesforce SObjects, unexpected errors can occur that will be reported. The following lists several reasons why errors may occur: - Attempting to insert a duplicate record. Rules for determining duplicates are [configurable in Salesforce](https://help.salesforce.com/articleView?id=duplicate_rules_overview.htm&type=5). - Attempting to delete, update, or upsert a record that does not exist because the `Id` field does not match. - Attempting an operation on a field where the `Id` field value matches a previously deleted `Id` field value. ### ID field semantics When the Salesforce Bulk API Sink connector consumes records on Kafka topics which originated from the [Salesforce PushTopic Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/index.html), an `Id` field is included that is a sibling of the other fields in the body of the SObject. Note that the `Id` is only valid within the Salesforce organization from which the record was streamed. For upsert, delete, and update operations, attempting to rely on the `Id` field causes failures when used on different Salesforce organizations. Inserts always ignore the `Id` field because `Id` fields are internally managed in Salesforce. Upsert operations must be used with the external id config options `salesforce.use.custom.id.field=true` and `salesforce.custom.id.field.name=`. ### Input topic record format The input topic record format is expected to be the same as the record format written to output topics by the [Salesforce PushTopic Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/index.html). The Kafka key value is not required. For an example, see the [Salesforce PushTopic Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/index.html). ### Read-Only fields Salesforce SObject fields may not be writable by insert, update, or upsert operation because the fields are set with `creatable=false` or `updatable=false` attributes within Salesforce. If a write is attempted to a field with these attributes set, the sink connector excludes the field in the operation rather than fail the entire operation. This behavior is **not configurable**. ### Event Type The Salesforce Bulk API Sink connector Kafka record format contains an `_EventType` field. This field describes the type of PushTopic event that generated the record, if the record was created by the [Salesforce PushTopic Source connector for Confluent Platform](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/index.html). Types are `created`, `updated`, and `deleted`. When processing records, the sink connector (by default) maps the `_EventType` to either an `insert`, `update`, or `delete` operation on the configured SObject. This behavior can be overridden using the `override.event.type=true` and `salesforce.sink.object.operation=` fields. Overriding the event type ignores the `_EventType` field in the record and obeys the `salesforce.sink.object.operation` for every record. ### API Limits - The Salesforce Bulk API Sink connector is limited by number of batches to execute, records per batch, and length of the batch. For detailed limitations, see [Bulk API Limits](https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/asynch_api_concepts_limits.htm). - The connector does not support the relation fields. - The Salesforce Bulk API supports only upsert operation with External Id field. ### 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). ## 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](../source/configuration_options.md#salesforce-bulk-api-connector-license-config). For information about the license topic, see [License topic configuration](configuration_options.md#salesforce-bulk-api-license-topic-configuration). ## Configuration Properties For a complete list of configuration properties for the sink connector, see [Salesforce Bulk API Sink Connector Configuration Properties](configuration_options.md#salesforce-bulk-api-sink-connector-config). ## Install the Salesforce Bulk API Sink 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. - Java Version: 1.8. - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) to run Quick Start (requires separate installation). - 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-bulk-api: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-bulk-api:3.1.0 ``` ## Quick Start In this quick start, the Salesforce PushTopic Source connector is used to get data into Kafka and the Salesforce Bulk API Sink connector is used to export data from Kafka to Salesforce. For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see [Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster). ### Populate a Kafka topic Complete the steps in the [Example: Configure Salesforce PushTopic Source connector](https://docs.confluent.io/kafka-connect-salesforce/current/pushtopics/salesforce_pushtopic_source_connector_quickstart.html) to get data into Kafka. Note the following: - After creating the `Lead` and before checking the topic for the `Lead` object, edit the name of the `Lead`. Two records should be present when you check the contents of the Kafka topic: a record of \_EventType `created` and another of `updated`. - Complete the **Salesforce Account** steps to create a second account in a separate Salesforce organization from the sink connector. This account is used to configure the sink connector. ### Configure the connector Prerequisites : - [Confluent Platform](/platform/current/installation/index.html) - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) (requires separate installation) To configure the connector, complete the following steps: #### NOTE You can add the following Single Message Transform (SMT) to the connector configuration to process records generated by the Salesforce Bulk API Source connector. ```text "transforms" : "InsertField", "transforms.InsertField.type" : "org.apache.kafka.connect.transforms.InsertField$Value", "transforms.InsertField.static.field" : "_EventType", "transforms.InsertField.static.value" : "created" ``` 1. Create a configuration file named `salesforce-bulk-api-leads-sink-config.json` with the following contents. Ensure you enter a real username, password, security token, consumer key, and consumer secret. For details about configuration properties, see [Configuration Properties](configuration_options.md#salesforce-bulk-api-sink-connector-config). ```text { "name" : "SalesforceBulkApiSinkConnector", "config" : { "connector.class" : "io.confluent.connect.salesforce.SalesforceBulkApiSinkConnector", "tasks.max" : "1", "topics" : "sfdc-pushtopic-lead", "salesforce.object" : "Lead", "salesforce.password" : "< Required Configuration >", "salesforce.password.token" : "< Required Configuration >", "salesforce.username" : "< Required Configuration: secondary organization username >", "reporter.result.topic.replication.factor" : "1", "reporter.error.topic.replication.factor" : "1", "reporter.bootstrap.servers" : "localhost:9092", "confluent.topic.bootstrap.servers": "localhost:9092", "confluent.topic.replication.factor": "1", "confluent.license": " Omit to enable trial mode " } } ``` For details about using this connector with Kafka Connect Reporter, see [Connect Reporter](/kafka-connectors/self-managed/userguide.html#userguide-connect-reporter). 2. Enter the Confluent CLI command to start the Salesforce Sink connector. ```bash confluent local load SalesforceBulkApiSinkConnector -- -d salesforce-bulk-api-leads-sink-config.json ``` Your output should resemble: ```none { "name": "SalesforceBulkApiSinkConnector", "config": { "connector.class" : "io.confluent.connect.salesforce.SalesforceBulkApiSinkConnector", "tasks.max" : "1", "topics" : "sfdc-pushtopic-leads", "salesforce.object" : "Lead", "salesforce.username" : "" "salesforce.password" : "", "salesforce.password.token" : "", "reporter.result.topic.replication.factor" : "1", "reporter.error.topic.replication.factor" : "1", "reporter.bootstrap.servers" : "localhost:9092", "confluent.topic.bootstrap.servers": "localhost:9092", "confluent.topic.replication.factor": "1", "confluent.license": " Omit to enable trial mode " }, "tasks": [ ... ], "type": null } ``` ## View Leads in Salesforce Log into the secondary Salesforce organization and verify the `Leads` object exists with the correct name. It should match the primary Salesforce organization. ## Running Both Connectors Concurrently Running both the sink connector and the source connector in separate workers allows for synchronizing SObject changes in near real-time. If running in standalone mode, add a custom port to one of the workers using the `listeners` configuration property. Add and change leads as necessary in the primary organization. The source connector captures your changes and writes them to the same topic that the sink connector reads from, and uses as a source for operations.