# Salesforce Bulk API 2.0 Sink Connector for Confluent Cloud
The fully managed Salesforce Bulk API 2.0 Sink connector for Confluent Cloud performs
insert, update, and delete operations on [Salesforce.com](https://developer.salesforce.com/) SObjects using records from Apache Kafka®
topics and writes them to Salesforce. This
connector uses [Salesforce Bulk API 2.0](https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/bulk_common_diff_two_versions.htm).
#### NOTE
* 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 Bulk API 2.0 Sink connector provides the following features:
* **API 2.0**: Supports [Salesforce Bulk API 2.0](https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/bulk_api_2_0.htm).
* **At least once delivery**: The connector guarantees that records are
delivered at least once to the Kafka topic. If the connector restarts, there
could be duplicate records in the Kafka topic.
* **Supported data formats:** The connector supports Avro, JSON Schema
(JSON_SR), and Protobuf output data. [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, or
Protobuf).
* **Supports multiple tasks**: The connector supports running one or more
tasks. More tasks may improve performance (that is, consumer lag is reduced
with multiple tasks running).
* **Supports Salesforce relationship fields**: The connector supports Salesforce relationship fields.
For more information, see [Salesforce relationship fields](#salesforce-relationship-fields-bulk-api-v2-sink).
* **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-bulk-api-v2-sink-setup-connection).
* **Supports Multiple SObjects**: Each connector instance supports up to five SObjects, provided that every selected topic is mapped to exactly one
unique SObject. For each SObject, you must provide the object type and a comma-separated list of associated topics,
ensuring that no single topic contains records for multiple SObjects.
* **Supports Salesforce Big Objects**: The connector supports writing to Salesforce
Big Objects (custom objects with names ending in `__b`) in addition to standard
and custom SObjects. Because Big Objects do not support update or delete
operations, only `created` is applicable for the `_EventType` field on Big
Object records; if a record carries an `_EventType` of `updated` or
`deleted`, it is routed to the error topic. As a best practice, omit
`_EventType` from Big Object schemas. Non-indexed Big Object fields require
explicit field-level security (FLS) grants on the connector user’s profile or
permission set, and relationship fields are not supported for Big Objects. For
more information, see
[Salesforce Big Objects](https://developer.salesforce.com/docs/atlas.en-us.bigobjects.meta/bigobjects/big_object.htm).
* **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.
## Salesforce relationship fields
The connector supports Salesforce relationship fields that allow you to integrate external
data fields within Salesforce sObjects, enhancing data aggregation and business processes.
This is useful for populating Lookup relationships using values from external systems (External IDs).
#### NOTE
This connector does not currently support polymorphic fields.
### Prerequisites
- A Salesforce lookup field must already exist on your target sObject
(for example, `RelatedAccount__c` pointing to the `Account` object).
- The related object must have an indexed field or an External ID field
(for example, `External_Account_ID__c`) to facilitate the relationship mapping.
### Process relationship fields
Follow the steps below to process Salesforce relationship fields in your connector:
#### Enable relationship field
When you configure a sink connector, you must enable relationship field support. In your connector configuration,
set the following parameters to `false`:
* `"skip.objectN.relationship.fields": "false"`
* `"salesforce.objectN.ignore.reference.fields": "false"`
**Sample Configuration**: Below is a sample configuration for a sink connector targeting a custom object (`RelationshipDemo__c`).
```json
{
"config": {
"schema.context.name": "default",
"input.data.format": "AVRO",
"connector.class": "SalesforceBulkApiV2Sink",
"name": "SalesforceBulkApiV2SinkConnector_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "E2G767UJZGO5JSMR",
"kafka.api.secret": "",
"salesforce.grant.type": "CLIENT_CREDENTIALS",
"salesforce.instance": "https://login.salesforce.com/",
"salesforce.consumer.key": "",
"salesforce.consumer.secret": "*******************************************",
"salesforce.object.num": 1,
"salesforce.object1": "RelationshipDemo__c",
"salesforce.object1.topics": "RelationshipDemoTopic",
"salesforce.object1.use.custom.id.field": "false",
"salesforce.object1.ignore.reference.fields": "false",
"skip.object1.relationship.fields": "false",
"salesforce.object1.override.event.type": "false",
"salesforce.object1.sink.object.operation": "upsert",
"salesforce.version": "65.0",
"behavior.on.api.errors": "ignore",
"request.max.retries.time.ms": "30000",
"max.timeout.ms": "200000",
"max.poll.interval.ms": "300000",
"max.poll.records": "500",
"tasks.max": "1",
"value.converter.decimal.format": "BASE64",
"value.converter.reference.subject.name.strategy": "DefaultReferenceSubjectNameStrategy",
"errors.tolerance": "all",
"value.converter.value.subject.name.strategy": "TopicNameStrategy",
"key.converter.key.subject.name.strategy": "TopicNameStrategy",
"value.converter.ignore.default.for.nullables": "false",
"auto.restart.on.user.error": "true"
}
}
```
#### Associated Avro schema for sink topic
To use a relationship field, the Avro schema for the sink topic must define the relationship object (`RelatedAccount__r`)
and the specific external field used for the lookup.
In the schema below, `RelatedAccount__r` is used to resolve and populate the `RelatedAccount__c` field in Salesforce.
The relationship field, `RelatedAccount__r`, is controlled by the external ID field, `External_Account_ID__c`.
**Avro Schema:** The following schema defines the structure for the `RelationshipDemo__c` sink topic.
```json
{
"connect.name": "io.confluent.salesforce.RelationshipDemo__c",
"fields": [
{
"name": "Id",
"type": {
"connect.doc": "Unique identifier for the object.",
"type": "string"
}
},
{
"default": null,
"name": "OwnerId",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "IsDeleted",
"type": [
"null",
"boolean"
]
},
{
"default": null,
"name": "Name",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "CreatedDate",
"type": [
"null",
{
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"connect.version": 1,
"logicalType": "timestamp-millis",
"type": "long"
}
]
},
{
"default": null,
"name": "CreatedById",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "LastModifiedDate",
"type": [
"null",
{
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"connect.version": 1,
"logicalType": "timestamp-millis",
"type": "long"
}
]
},
{
"default": null,
"name": "LastModifiedById",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "SystemModstamp",
"type": [
"null",
{
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"connect.version": 1,
"logicalType": "timestamp-millis",
"type": "long"
}
]
},
{
"default": null,
"name": "LastViewedDate",
"type": [
"null",
{
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"connect.version": 1,
"logicalType": "timestamp-millis",
"type": "long"
}
]
},
{
"default": null,
"name": "LastReferencedDate",
"type": [
"null",
{
"connect.name": "org.apache.kafka.connect.data.Timestamp",
"connect.version": 1,
"logicalType": "timestamp-millis",
"type": "long"
}
]
},
{
"name": "RelatedAccount__r",
"type": {
"connect.name": "RelatedAccount__r",
"fields": [
{
"default": null,
"name": "External_Account_ID__c",
"type": [
"null",
"string"
]
}
],
"name": "RelatedAccount__r",
"type": "record"
}
},
{
"default": null,
"name": "_ObjectType",
"type": [
"null",
"string"
]
},
{
"default": null,
"name": "_EventType",
"type": [
"null",
"string"
]
}
],
"name": "RelationshipDemo__c",
"namespace": "io.confluent.salesforce",
"type": "record"
}
```
#### Sample sink record
When sending records to the sink connector for `RelationshipDemo__c`, use the relationship field (`RelatedAccount__r`) to specify the external ID of the related record.
```json
{
"Id": "a00gL00000P85iUQAW",
"OwnerId": {
"string": "005gL000005tlt3QAA"
},
"IsDeleted": {
"boolean": false
},
"Name": {
"string": "DemoEntity"
},
"CreatedDate": {
"long": 1761736099000
},
"CreatedById": {
"string": "005gL000005tlt3QAA"
},
"LastModifiedDate": {
"long": 1761736099000
},
"LastModifiedById": {
"string": "005gL000005tlt3QAA"
},
"SystemModstamp": {
"long": 1761736099000
},
"LastViewedDate": null,
"LastReferencedDate": null,
"RelatedAccount__r": {
"External_Account_ID__c": {
"string": "123456"
}
},
"_ObjectType": {
"string": "RelationshipDemo__c"
},
"_EventType": {
"string": "created"
}
}
```
Upon processing this record, a new `RelationshipDemo` sObject named `DemoEntity` is created.
The connector automatically populates the `RelatedAccount__c` field with the `accountId` of the
`DemoAccount` that matches the `External_Account_ID__c` value of `123456`.
### Common issues and resolutions
| Issue | Potential Cause | Resolution |
|---------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| `Cannot specify both an external ID reference RelatedAccount__r and a salesforce id, RelatedAccount__c` | The record or schema includes both the lookup field (`__c`) and the relationship reference (`__r`). Salesforce requires only one identifier for sink operations. | Ensure the record schema uses only one field type. If source data is mixed, route different field types to separate topics for processing. |
| `Field name provided, Name is not an External ID or indexed field for Account` | The field used to define the relationship is not an indexed or External ID field. | Use an indexed or External ID field when referencing relationships. `idLookup` fields are only supported if the referenced field is the same object type as the parent. Otherwise, use a field explicitly marked as an **External Id**. |
### **Auto-create missing parent records for relationship fields**
By default, if the connector writes a child record whose parent record
does not yet exist in Salesforce, Salesforce rejects the record.
Whenever possible, avoid this by ensuring the parent record already
exists in Salesforce before the connector sends the child record, for example, by
ordering or staging the data upstream in your source system.
When you cannot guarantee that order, that is, when a child record
might reach Salesforce before its parent exists, you can configure a
staging field together with a `before insert` Apex trigger. With
this setup, Salesforce creates the missing parent record automatically
and links the child record to it as the record is inserted.
**Why a direct reference fails**
A relationship field (`__r`) links a child record to a parent using
the parent’s External ID. Salesforce resolves that reference in its
API layer, before the record reaches the database. For example:
```json
"RelatedAccount__r": { "External_Account_ID__c": "12345" }
```
If the referenced parent, `Account 12345` in this example, does not
exist, the API layer rejects the record with
`FOREIGN_KEY_EXTERNAL_ID_NOT_FOUND` before it ever reaches the
database. Because the record never reaches the database, any Apex trigger
that could have created the parent never runs.
**How the trigger creates the parent record**
To avoid that validation failure, send the parent’s External ID in a
plain text field instead of in the relationship field. This extra text
field is a phantom field (a staging field). It carries no business meaning
and exists only to move the External ID past the API layer’s validation. Salesforce
does not validate a plain text field against the parent object, so the
record passes API layer validation and reaches the database. From
there, a `before insert` trigger reads that text field, creates
the parent record if it does not already exist, and populates the
real lookup field with the new parent’s Salesforce record ID before
the record is saved. A lookup field requires an 18-character
Salesforce record ID rather than text, so the trigger must resolve
the External ID to that record ID before the record is saved.
**Salesforce setup**
On the child object (example: `RelationshipDemo__c`):
* `RelatedAccount__c`: the lookup to the parent `Account` (the
real relationship field).
* `relationValue__c`: a Text (255-character) staging field that
carries the parent’s External ID. This field has no business
meaning of its own. Its only purpose is to pass the External ID to
the trigger.
* Your own data fields, for example `Name` (Text) and
`ObjectNum__c` (Number).
On the parent object (`Account`):
* `External_Account_ID__c`: an indexed, unique External ID field.
**Payload**
For each child record, put the parent’s External ID in the staging
field (`relationValue__c`) and leave the lookup field
(`RelatedAccount__c`) empty. The trigger populates the lookup field.
#### IMPORTANT
Include the `RelatedAccount__c` column in your input file, with
an empty value on every row. Do not omit it. If the column is
absent, the Bulk API 2.0 request fails with an HTTP 500 error when
the trigger populates the lookup field and changes the record’s
shape.
```text
ObjectNum__c,Name,relationValue__c,RelatedAccount__c
123,Global Nexus,12345,
124,Test Entity Two,99988,
```
**Apex trigger (for reference)**
The following trigger runs only for Bulk API transactions. For each
batch, it reads the External IDs from the staging field, looks up
which parent records already exist, and creates the ones that are
missing. If two batches try to create the same parent record at the
same time, the trigger re-queries for the record that the other batch
committed instead of failing. It then links each child record to its
parent, or fails the child record if its parent genuinely could not
be created.
```java
trigger RelationshipDemoTrigger on RelationshipDemo__c (before insert) {
// 1. Run only for Bulk API transactions. Allow Apex tests so the trigger is covered.
if (System.Request.getCurrent().getQuiddity() != System.Quiddity.BULK_API
&& !Test.isRunningTest()) {
return;
}
// 2. Collect parent External IDs from the staging field.
Set accountExtIds = new Set();
for (RelationshipDemo__c demo : Trigger.new) {
if (String.isNotBlank(demo.relationValue__c) && demo.RelatedAccount__c == null) {
accountExtIds.add(demo.relationValue__c);
}
}
if (accountExtIds.isEmpty()) return;
// 3. Find which parents already exist (single query). A lookup field needs a real
// 18-character record Id, so resolve each External ID to its Account Id here.
Map accountsByExtId = new Map();
for (Account acc : [SELECT Id, External_Account_ID__c FROM Account
WHERE External_Account_ID__c IN :accountExtIds]) {
accountsByExtId.put(acc.External_Account_ID__c, acc.Id);
}
// 4. Build placeholder parents for External IDs that do not exist yet.
Map newAccounts = new Map();
for (String extId : accountExtIds) {
if (!accountsByExtId.containsKey(extId)) {
newAccounts.put(extId, new Account(
External_Account_ID__c = extId,
Name = ('Mock - ' + extId + ' - ' + System.now().getTime()).left(255),
Description = 'Auto-generated via Bulk API'
));
}
}
// 5. Insert placeholders with partial success, and classify any failures:
// - row lock / duplicate: a parallel Bulk API batch is creating the same parent
// (transient) -> recover by re-querying.
// - any other error (for example, a validation rule): permanent -> fail the child.
Set lockedExtIds = new Set();
Map failedExtIds = new Map();
if (!newAccounts.isEmpty()) {
List toInsert = newAccounts.values();
Database.SaveResult[] results = Database.insert(toInsert, false);
for (Integer i = 0; i < results.size(); i++) {
Account acc = toInsert[i];
if (results[i].isSuccess()) {
accountsByExtId.put(acc.External_Account_ID__c, results[i].getId());
} else {
Database.Error err = results[i].getErrors()[0];
if (err.getStatusCode() == StatusCode.UNABLE_TO_LOCK_ROW
|| err.getStatusCode() == StatusCode.DUPLICATE_VALUE) {
lockedExtIds.add(acc.External_Account_ID__c);
} else {
failedExtIds.put(acc.External_Account_ID__c, err.getMessage());
}
}
}
}
// 6. Re-query the locked ones: the winning batch has likely committed the parent.
// Remove any we recover so lockedExtIds holds only those still unresolved.
if (!lockedExtIds.isEmpty()) {
for (Account acc : [SELECT Id, External_Account_ID__c FROM Account
WHERE External_Account_ID__c IN :lockedExtIds]) {
accountsByExtId.put(acc.External_Account_ID__c, acc.Id);
lockedExtIds.remove(acc.External_Account_ID__c);
}
}
// 7. Link each child, or decide how to fail it.
for (RelationshipDemo__c demo : Trigger.new) {
String extId = demo.relationValue__c;
if (String.isBlank(extId) || demo.RelatedAccount__c != null) continue;
if (accountsByExtId.containsKey(extId)) {
// Parent resolved: link the child.
demo.RelatedAccount__c = accountsByExtId.get(extId);
} else if (failedExtIds.containsKey(extId)) {
// Parent could not be created (permanent): fail this row visibly.
demo.addError('Could not create parent Account for External ID '
+ extId + ': ' + failedExtIds.get(extId));
} else if (lockedExtIds.contains(extId)) {
// Transient lock: a competing batch has not committed yet. Insert the child
// unlinked (orphan) and reconcile later rather than rejecting valid data.
}
}
}
```
The guard clause restricts the logic to Bulk API transactions (and
Apex tests), so manual UI edits and other integrations are
unaffected. All queries and DML run once per batch (bulkified) to
stay within Salesforce governor limits. The last block encodes the
intended failure policy: a row whose parent failed to be created
for a real reason (for example, a validation rule) is failed with
`addError()`, while a row blocked only by a transient lock from a
competing batch is inserted unlinked and reconciled later, so valid
data is never rejected over a temporary lock.
## Limitations
Be sure to review the following information.
* For connector limitations, see [Salesforce Bulk API 2.0 Sink Connector](limits.md#cc-salesforce-bulk-api-v2-sink-limits) limitations. For additional information, see [Considerations](#cc-salesforce-bulk-api-v2-sink-considerations).
* If you plan to use one or more Single Message Transformations (SMTs), see [SMT Limitations](single-message-transforms.md#cc-single-message-transforms-limitations).
## 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 Bulk API 2.0
Sink connector. The quick start provides the basics of selecting the connector
and configuring it to capture records and record changes from Kafka topics.
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.
- Salesforce account credentials.
- 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).
- **At least one topic must exist** in your Confluent Cloud cluster before creating
the connector.
- 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).
- 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 Bulk API 2.0 Sink** connector card.
#### IMPORTANT
At least one topic must exist in your Confluent Cloud cluster before creating the
connector.
#### Step 4: Enter the connector details
#### NOTE
* Make sure you have all your [prerequisites](#cc-salesforce-bulk-api-v2-sink-prereqs) completed.
* An asterisk ( \* ) designates a required entry.
At the **Add Salesforce Bulk API 2.0 Sink Connector** screen, complete the
following:
### Topic selection
Select the topic you want to send data to from the **Topics** list. To
create a new topic, click **+Add new topic**.
### 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
- **Input Kafka record value format**: Select the input Kafka record value format (data coming from the
Kafka topic). Valid values are AVRO, JSON_SR (JSON Schema), or PROTOBUF. A valid
schema must be available in [Schema Registry](../get-started/schema-registry.md#cloud-sr-config)
to use a schema-based message format (for example, Avro, JSON_SR
(JSON Schema), or Protobuf).
for additional information.
- **Salesforce Object Type**: Select Salesforce Object type to process. ‘STANDARD_OR_CUSTOM_OBJECT’ for regular SObjects, ‘BIG_OBJECT’ for Big Objects (only supports insert operation).
- **Number of Salesforce Objects**: Number of Salesforce Objects to write to. It must be between 1 and 5.
**Object 1 configuration**
- **Salesforce SObject1 Name**: Specifies the Salesforce SObject1 to write to.
- **Salesforce SObject1 Topics**: A comma-separated list of topics associated with Salesforce SObject1.
- **SObject1 Override Event Type**: Determines whether to override the SObject1 EventType(create, update, delete) with the configured sink operation.
- **SObject1 Sink Operation**: The Salesforce sink operation to perform for SObject1 if an override is enabled.
- **SObject1 Ignore Fields**: A comma-separated list of fields to ignore when pushing SObject1 records.
- **SObject1 Ignore Reference Fields**: Prevents reference-type fields from being updated or inserted for SObject1.
- **SObject1 Use Custom ID Field**: Determines whether to use a custom external ID field for SObject1 insert or upsert operations.
- **SObject1 Custom ID Field Name**: Specifies the custom external ID field name for SObject1.
- **Skip SObject1 Relationship Fields**: Prevents relationship fields in records from being updated or inserted in Salesforce SObjects.
**Data decryption**
- Enable **Client-Side Field Level Encryption** for
data decryption. Specify a **Service Account** to
access the Schema Registry and associated encryption rules or keys with that schema. Select the connector behavior
(`ERROR` or `NONE`) on data decryption failure. If set to `ERROR`, the connector fails and writes the encrypted data
in the DLQ. If set to `NONE`, the connector writes the encrypted data in the target system without decryption.
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).
- **Behavior on API errors**: How the connector behaves when a
Salesforce API error occurs. Valid options are `fail` and
`ignore` (the default). If set to `fail`, the connector
stops.
- **Max timeout milliseconds**: The maximum time in milliseconds
(ms) that the connector waits for all batch operations to
complete. Defaults to `200000` ms.
**Additional Configs**
- **Value Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for values. The deserializer reads schema IDs from message headers.
- **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`.
- **Schema ID For Value Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **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.
- **Key Converter Schema ID Deserializer**: Sets the class name of the schema ID deserializer for keys. The deserializer reads schema IDs from message headers.
- **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.
- **Schema GUID For Key Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **Schema GUID For Value Converter**: Sets the schema GUID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema GUID for deserializing message values. This property is applicable only when `value.converter.value.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
- **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.
- **Schema ID For Key Converter**: Sets the schema ID to use for deserialization when using `ConfigSchemaIdDeserializer`. This lets you specify a fixed schema ID for deserializing message keys. This property is applicable only when `key.converter.key.schema.id.deserializer` is set to `ConfigSchemaIdDeserializer`.
**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.
**Consumer configuration**
- **Max poll interval(ms)**: Sets the maximum delay between subsequent consume requests to Kafka. Use this property to
improve connector performance in cases when the connector cannot send records to the sink system.
The default is 300,000 milliseconds (5 minutes).
- **Max poll records**: Sets the maximum number of records to consume from Kafka in a single request. Use this property to
improve connector performance in cases when the connector cannot send records to the sink system.
The default is 500 records.
**Connection details**
- **Max retry time (ms)**: Maximum time in milliseconds until the connector stops retrying failed Salesforce requests. Default is `30000` and minimum is `1000`.
**Salesforce details**
- **Salesforce Version**: The version of the Salesforce API to use. Defaults to `latest`.
**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-bulk-api-v2-sink-config-properties).
- Click **Continue**.
### Sizing
The connector supports running one or more tasks. More tasks may improve
performance (that is, consumer lag is reduced with multiple tasks
running).
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 for records
Verify that records are being produced at the endpoint. For additional
information, see [Considerations](#cc-salesforce-bulk-api-v2-sink-considerations).
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.
#### IMPORTANT
Make sure you have all your [prerequisites](#cc-salesforce-bulk-api-v2-sink-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 for a single SObject.
```none
{
"connector.class": "SalesforceBulkApiV2Sink",
"name": "SalesforceBulkApiV2Sink_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "",
"kafka.api.secret": "",
"topics": "TestBulkAPI",
"input.data.format": "AVRO",
"salesforce.grant.type": "PASSWORD",
"salesforce.instance": "https://login.salesforce.com",
"salesforce.username": "",
"salesforce.password": "**************",
"salesforce.password.token": "************************",
"salesforce.consumer.key": "**************",
"salesforce.consumer.secret": "************************",
"salesforce.object.num": "1",
"salesforce.object1": "",
"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
```
* `""topics"`: Enter a Kafka topic name or a comma-separated list of topics. A
topic must exist before launching the connector.
* `"input.data.format"`: Sets the input data format (data coming from the
Kafka topic): AVRO, JSON_SR (JSON Schema), or PROTOBUF. A valid schema must be
available in [Schema Registry](../get-started/schema-registry.md#cloud-sr-config) to use a schema-based
message format (for example, Avro, JSON_SR (JSON Schema), or Protobuf).
* `"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" : "",
```
* `"salesforce.jwt.keystore.password"`: Enter the password used to access the
JWT keystore file.
* `"salesforce.object1"`: Enter the Object1 name to write to.
* `"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.
* `csfle.onFailure`: Configures the connector behavior (`ERROR` or `NONE`) on data decryption failure.
If set to `ERROR`, the connector fails and writes the encrypted data
in the DLQ. If set to `NONE`, the connector writes the encrypted data in the target system without decryption.
When using CSFLE or CSPE with connectors that route failed messages to a Dead Letter Queue (DLQ),
be aware that data sent to the DLQ is written in unencrypted plaintext. This poses
a significant security risk as sensitive data that should be encrypted may be exposed in the DLQ.
Do not use DLQ with CSFLE or CSPE in the current version. If you need error handling for
CSFLE- or CSPE-enabled data, use alternative approaches such as:
* Setting the connector behavior to `ERROR` to throw exceptions instead of routing to DLQ
* Implementing custom error handling in your applications
* Using `NONE` to pass encrypted data through without decryption
For more information on CSFLE or CSPE setup, see [Manage encryption for connectors](csfle.md#connect-csfle).
The following example shows the required connector properties for two SObjects:
```json
{
"connector.class": "SalesforceBulkApiV2Sink",
"input.data.format": "AVRO",
"name": "SalesforceBulkApiV2Sink_0",
"kafka.auth.mode": "KAFKA_API_KEY",
"kafka.api.key": "",
"kafka.api.secret": "",
"salesforce.grant.type": "PASSWORD",
"salesforce.instance": "https://login.salesforce.com",
"salesforce.username": "",
"salesforce.password": "",
"salesforce.password.token": "",
"salesforce.consumer.key": "",
"salesforce.consumer.secret": "",
"salesforce.object.num": "2",
"salesforce.object1": "",
"salesforce.object1.topics": "",
"salesforce.object2": "",
"salesforce.object2.topics": "",
"tasks.max": "1"
}
```
**SMTs**: For details about adding SMTs using the Confluent CLI, see the [Single Message Transformations](single-message-transforms.md#cc-single-message-transforms) documentation.
For all property values and description,
see [Configuration Properties](#cc-salesforce-bulk-api-v2-sink-config-properties). For additional
information, see [Considerations](#cc-salesforce-bulk-api-v2-sink-considerations).
#### 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-bulk-api-v2-sink.json
```
Example output:
```none
Created connector SalesforceBulkApiV2Sink_0 lcc-aj3qr
```
#### 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-aj3qr | SalesforceBulkApiV2Sink_0 | RUNNING | sink
```
#### Step 6: Check Check for records.
Verify that records are being produced at the endpoint. For additional information, see [Considerations](#cc-salesforce-bulk-api-v2-sink-considerations).
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).
### Which topics do you want to get data from?
`topics.regex`
: A regular expression that matches the names of the topics to consume from. This is useful when you want to consume from multiple topics that match a certain pattern without having to list them all individually.
* Type: string
* Importance: low
`topics`
: Identifies the topic name or a comma-separated list of topic names.
* Type: list
* Importance: high
`errors.deadletterqueue.topic.name`
: The name of the topic to be used as the dead letter queue (DLQ) for messages that result in an error when processed by this sink connector, or its transformations or converters. Defaults to ‘dlq-${connector}’ if not set. The DLQ topic will be created automatically if it does not exist. You can provide `${connector}` in the value to use it as a placeholder for the logical cluster ID.
* Type: string
* Default: dlq-${connector}
* Importance: low
`reporter.result.topic.name`
: The name of the topic to produce records to after successfully processing a sink record. Defaults to ‘success-${connector}’ if not set. You can provide `${connector}` in the value to use it as a placeholder for the logical cluster ID.
* Type: string
* Default: success-${connector}
* Importance: low
`reporter.error.topic.name`
: The name of the topic to produce records to after each unsuccessful record sink attempt. Defaults to ‘error-${connector}’ if not set. You can provide `${connector}` in the value to use it as a placeholder for the logical cluster ID.
* Type: string
* Default: error-${connector}
* Importance: low
### 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
### Input messages
`input.data.format`
: Sets the input Kafka record value format. Valid entries are AVRO, JSON_SR and PROTOBUF. Note that you need to have Confluent Cloud Schema Registry configured
* Type: string
* Importance: high
### 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
### 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.password`
: Salesforce account password for connector authentication
* Type: password
* 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.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
`salesforce.object.type`
: Select Salesforce Object type to process. ‘STANDARD_OR_CUSTOM_OBJECT’ for regular SObjects, ‘BIG_OBJECT’ for Big Objects (only supports insert operation).
* Type: string
* Default: STANDARD_OR_CUSTOM_OBJECT
* Importance: high
`salesforce.object.num`
: Number of Salesforce Objects to write to. Must be between 1 and 5.
* Type: int
* Default: 1
* Valid Values: [1,…,5]
* Importance: high
`salesforce.version`
: The version of Salesforce API to use.
* Type: string
* Default: 67.0
* Importance: low
`oauth.refresh.token`
: OAuth 2.0 refresh token. Populated by Confluent Cloud after authorization.
* Type: password
* Importance: high
### Object 1 configuration
`salesforce.object1`
: The Salesforce SObject1 to write to.
* Type: string
* Importance: high
`salesforce.object1.topics`
: Comma separated list of topics associated with Salesforce SObject1
* Type: list
* Default: “”
* Importance: high
`salesforce.object1.override.event.type`
: Override SObject1 EventType(create, update, delete) with the configured sink operation.
* Type: boolean
* Default: false
* Importance: low
`salesforce.object1.sink.object.operation`
: The Salesforce sink operation to perform for SObject1 when override is enabled.
* Type: string
* Default: insert
* Importance: low
`salesforce.object1.ignore.fields`
: Comma separated list of fields to ignore when pushing a record for SObject1.
* Type: list
* Default: “”
* Importance: low
`salesforce.object1.ignore.reference.fields`
: Prevent reference-type fields from being updated or inserted for SObject1.
* Type: boolean
* Default: false
* Importance: low
`salesforce.object1.use.custom.id.field`
: Use a custom external ID field for SObject1 insert/upsert operations.
* Type: boolean
* Default: false
* Importance: low
`salesforce.object1.custom.id.field.name`
: The custom external ID field name for SObject1.
* Type: string
* Default: “”
* Importance: low
`skip.object1.relationship.fields`
: Flag to skip relationship fields in records from being updated or inserted in Salesforce SObjects.
* Type: boolean
* Default: true
* Importance: medium
### Connection details
`behavior.on.api.errors`
: Error handling behavior config for any API errors.
* Type: string
* Default: ignore
* 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
`max.timeout.ms`
: The maximum timeout in milliseconds that the connector will continue waiting for the completion of all batch operations.
* Type: long
* Default: 200000 (200 seconds)
* Importance: low
### Consumer configuration
`max.poll.interval.ms`
: The maximum delay between subsequent consume requests to Kafka. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 300000 milliseconds (5 minutes).
* Type: long
* Default: 300000 (5 minutes)
* Valid Values: [60000,…,1800000] for non-dedicated clusters and [60000,…] for dedicated clusters
* Importance: low
`max.poll.records`
: The maximum number of records to consume from Kafka in a single request. This configuration property may be used to improve the performance of the connector, if the connector cannot send records to the sink system. Defaults to 500 records.
* Type: long
* Default: 500
* Valid Values: [1,…,500] for non-dedicated clusters and [1,…] for dedicated clusters
* Importance: low
### Number of tasks for this connector
`tasks.max`
: Maximum number of tasks for the connector.
* Type: int
* Valid Values: [1,…]
* Importance: high
### Additional Configs
`consumer.override.auto.offset.reset`
: Defines the behavior of the consumer when there is no committed position (which occurs when the group is first initialized) or when an offset is out of range. You can choose either to reset the position to the “earliest” offset (the default) or the “latest” offset. You can also select “none” if you would rather set the initial offset yourself and you are willing to handle out of range errors manually. More details: [https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#auto-offset-reset](https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#auto-offset-reset)
* Type: string
* Importance: low
`consumer.override.isolation.level`
: Controls how to read messages written transactionally. If set to read_committed, consumer.poll() will only return transactional messages which have been committed. If set to read_uncommitted (the default), consumer.poll() will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode. More details: [https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level](https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html#isolation-level)
* Type: string
* Importance: low
`header.converter`
: The converter class for the headers. This is used to serialize and deserialize the headers of the messages.
* Type: string
* Importance: low
`key.converter.use.schema.guid`
: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: string
* Importance: low
`key.converter.use.schema.id`
: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message keys. Only applicable when key.converter.key.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: int
* 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.use.schema.guid`
: The schema GUID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema GUID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: string
* Importance: low
`value.converter.use.schema.id`
: The schema ID to use for deserialization when using ConfigSchemaIdDeserializer. This allows you to specify a fixed schema ID to be used for deserializing message values. Only applicable when value.converter.value.schema.id.deserializer is set to ConfigSchemaIdDeserializer.
* Type: int
* 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: all
* Importance: low
`key.converter.key.schema.id.deserializer`
: The class name of the schema ID deserializer for keys. This is used to deserialize schema IDs from the message headers.
* Type: string
* Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer
* 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.value.schema.id.deserializer`
: The class name of the schema ID deserializer for values. This is used to deserialize schema IDs from the message headers.
* Type: string
* Default: io.confluent.kafka.serializers.schema.id.DualSchemaIdDeserializer
* 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
### Auto-restart policy
`auto.restart.on.user.error`
: Enable connector to automatically restart on user-actionable errors.
* Type: boolean
* Default: true
* Importance: medium
## Considerations
Note the following when using this 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, 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 this `Id` field causes failures when used on different Salesforce
organizations. Inserts always ignore the `Id` field because `Id` fields are
internally fully managed in Salesforce. Upsert operations must be used with the
external ID configuration properties `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. The Kafka
key value is not required.
### 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. 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 Salesforce Bulk API supports `upsert` operations only when used with the external ID configuration properties `salesforce.use.custom.id.field=true` and `salesforce.custom.id.field.name=`.
## Frequently asked questions
Find answers to frequently asked questions about the Salesforce Bulk API 2.0 Sink connector for Confluent Cloud.
### Authentication and connection
#### What authentication methods are supported, and how do I configure `CLIENT_CREDENTIALS` grant type?
The connector supports three authentication grant types:
* **PASSWORD** (username + password + security token): Default authentication method
* **JWT_BEARER** (JSON Web Token): Certificate-based authentication
* **CLIENT_CREDENTIALS** (OAuth 2.0): Machine-to-machine authentication
For `CLIENT_CREDENTIALS` authentication:
1. You must use your Salesforce my domain URL (for example, `https://mycompany.my.salesforce.com`) in the `salesforce.instance` configuration. The default `https://login.salesforce.com` URL will not work.
2. Enable **Client Credentials Flow** in your Salesforce Connected App settings.
3. Configure a **Run As** user (execution user) in the Connected App policies. Without this, authentication fails with an `invalid_grant` error.
4. Ensure the execution user has the necessary permissions and profiles assigned.
For more information, see [Step 4: Enter the connector details](#cc-salesforce-bulk-api-v2-sink-setup-connection).
#### Why am I getting “Connection pool shut down” errors?
This error typically occurs during connector restarts or when the connector is shutting down:
```none
java.lang.IllegalStateException: Connection pool shut down
at io.confluent.connect.salesforce.bulk.v2.SalesforceBulkApiV2SinkTask.tryBatch
```
The error indicates that the connector attempted to execute a Bulk API operation while the HTTP connection
pool was being shut down. This is usually a transient error that occurs during:
* Connector restarts (manual or automatic).
* Configuration updates that require task restarts.
* Platform maintenance windows.
**Resolution:**
* The connector will automatically recover after the restart completes.
* If the error persists after multiple restarts, check connector logs for underlying authentication or network issues.
* Verify that your Salesforce credentials are valid and not expired.
### Data format and schema issues
#### Why am I getting `Bad Kafka SinkRecord. Value is not of type Struct` errors?
This error occurs when the connector encounters tombstone records (null values) or records
that are not in the expected Struct format:
```none
org.apache.kafka.connect.errors.ConnectException: Bad Kafka SinkRecord.
Value is not of type Struct. Kafka topics are not in the format the connector is expecting.
```
The Salesforce Bulk API 2.0 Sink connector requires all records to be in Struct format and does
not handle tombstone records by default.
**Resolution:**
* **Use the Tombstone Handler SMT**: Add the `io.confluent.connect.transforms.TombstoneHandler` Single Message Transform to filter out tombstone records. For more information, see [/platform/current/TombstoneHandler ](/platform/current/TombstoneHandler ).
* **Verify data format**: Ensure your topic data is in Avro, JSON Schema (JSON_SR), or Protobuf format with a valid schema registered in Confluent Cloud Schema Registry.
* **Check Schema Registry**: Confirm that Confluent Cloud Schema Registry is enabled and contains valid schemas for your topics.
Example configuration with Tombstone Handler:
```json
{
"transforms": "TombstoneHandler",
"transforms.TombstoneHandler.type": "io.confluent.connect.transforms.TombstoneHandler"
}
```
#### Why are some fields missing after I updated my Salesforce object definition?
When you add new fields to a Salesforce object, the connector may not immediately recognize them. The connector caches the Salesforce object metadata when it connects.
To ensure the connector processes new fields:
1. **Restart the connector**: This refreshes the cached Salesforce object metadata and ensures the connector recognizes newly added fields.
2. **Verify field visibility**: Ensure the Salesforce user has field-level security permissions to access the new fields.
3. **Check success topic**: Review messages in the success reporter topic to confirm all expected fields are being processed.
#### NOTE
The connector does not automatically detect schema changes. You must manually restart the connector after modifying the Salesforce object definition.
### Batching and performance tuning
#### How do I control batch size to avoid exceeding Salesforce daily batch limits?
Salesforce enforces a strict daily limit of 15,000 batches per 24 hours for Bulk API 2.0 operations. By default, the connector may create many small batches, which can quickly exhaust this limit.
To optimize batching and reduce the number of batches created:
**Configure consumer override properties:**
The following properties control how many records are fetched from Kafka and batched together before sending to Salesforce:
* `consumer.override.max.poll.records`: Maximum number of records fetched in a single poll (default: `500`). Increase this to batch more records together.
* `consumer.override.fetch.min.bytes`: Minimum bytes of data to fetch per request (default: `1 byte`). Increase this to wait for more data before creating a batch.
* `consumer.override.max.poll.interval.ms`: Maximum time between polls (default: `300000 ms` / 5 minutes). Increase if processing large batches takes longer.
Example configuration for large batches:
```json
{
"consumer.override.max.poll.records": "50000",
"consumer.override.fetch.min.bytes": "512000",
"consumer.override.max.poll.interval.ms": "600000"
}
```
**Important considerations:**
* The maximum batch size for Salesforce Bulk API 2.0 is 150 MB (with base64 encoding).
* If your topic contains duplicate external IDs within a batch, the connector splits the batch to comply with Salesforce’s requirement that external IDs must be unique within a batch.
* Larger batches reduce the number of API calls but may increase processing time per batch.
For more information, see [Salesforce Bulk API Limits](https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/asynch_api_concepts_limits.htm).
#### NOTE
The `consumer.override` properties are advanced configurations. Contact [Confluent Support](https://support.confluent.io/) if you need assistance configuring these settings for your environment.
#### Why does the connector create multiple small batches instead of one large batch?
The connector may split records into multiple batches for several reasons:
1. **Duplicate external IDs**: Salesforce Bulk API 2.0 does not allow duplicate external ID values within the same batch. When the connector detects duplicate external IDs in the polled records, it automatically creates separate batches.
2. **Batch size limits**: Salesforce limits each batch to 150 MB. If the polled records exceed this size, the connector splits them into multiple batches.
3. **Operation type differences**: Records with different operation types (insert, update, upsert, delete) are batched separately.
4. **Object type differences**: When using multiple SObjects, records for different objects are batched separately.
**To minimize batch creation:**
* Ensure records in your Kafka topic have unique external ID values when using upsert operations.
* Increase `consumer.override.max.poll.records` and `consumer.override.fetch.min.bytes` to fetch more records per poll.
* Consider using a single operation type per connector instance.
#### How can I monitor Salesforce API usage to avoid hitting batch limits?
To monitor your Salesforce API and batch usage:
1. **Salesforce Setup Console**: Navigate to **Setup > System Overview > API Usage** in your Salesforce org to view current API and Bulk API usage.
2. **Success and Error Topics**: Monitor the connector’s success and error reporter topics to track how many batches are being created and their success rate.
3. **Connector Metrics**: Use Confluent Cloud metrics to monitor connector throughput, lag, and error rates.
4. **Salesforce Event Monitoring**: Enable Event Monitoring in Salesforce (Enterprise Edition and above) to track API usage patterns.
If you consistently approach the 15,000 batch limit:
* Optimize batching configuration as described in the batching FAQ above.
* Distribute workload across multiple Salesforce organizations if possible.
* Consider using multiple Salesforce user accounts to distribute API usage.
### SObject operations and configuration
#### Can I configure multiple SObjects in one connector?
Yes, the connector supports configuring up to **five SObjects** in a single connector instance.
To configure multiple SObjects:
1. Set **Salesforce SObject Number** (`salesforce.object.num`) to the number of SObjects (1-5).
2. For each SObject, provide:
* The SObject name (`salesforce.object`).
* A comma-separated list of associated topics (`salesforce.object.topics`).
* Optional: SObject-specific configurations like `salesforce.object.sink.operation`.
3. Each topic must be mapped to exactly one SObject. A topic cannot contain records for multiple SObjects.
Example configuration for two SObjects:
```json
{
"salesforce.object.num": "2",
"salesforce.object1": "Account",
"salesforce.object1.topics": "account-topic",
"salesforce.object1.sink.operation": "upsert",
"salesforce.object2": "Contact",
"salesforce.object2.topics": "contact-topic",
"salesforce.object2.sink.operation": "insert"
}
```
This feature allows you to reduce operational overhead and costs by consolidating multiple low-throughput SObjects into a single connector instance.
#### Does the connector support Salesforce Big Objects?
Yes, the Salesforce Bulk API 2.0 Sink connector supports writing to Salesforce
Big Objects (custom objects with names ending in `__b`) in addition to
standard and custom SObjects.
Big Objects have different API requirements and limitations compared to
standard SObjects:
* They use a different indexing mechanism (composite indexes instead of standard IDs).
* For this sink connector, Big Objects support insert operations only; update
and delete operations are not applicable.
* They do not support standard Salesforce IDs.
Keep the following in mind when configuring the connector for Big Objects:
* Because Big Objects do not support update or delete operations, records
with an `_EventType` of `updated` or `deleted` are routed to the error
topic. As a best practice, omit `_EventType` from Big Object schemas.
with an `_EventType` of `updated` or `deleted` are routed to the error
topic. As a best practice, omit `_EventType` from Big Object schemas.
* Non-indexed Big Object fields require explicit field-level security (FLS)
grants on the connector user’s profile or permission set.
* Relationship fields are not supported for Big Objects.
For more information, see
[Salesforce Big Objects](https://developer.salesforce.com/docs/atlas.en-us.bigobjects.meta/bigobjects/big_object.htm).
#### How do I configure the connector to handle relationship fields?
The Salesforce Bulk API 2.0 Sink connector supports relationship fields (lookup and master-detail relationships) using the `skip.objectN.relationship.fields` configuration property.
**Default behavior (skip.objectN.relationship.fields=true):**
Relationship fields are filtered out and not sent to Salesforce. This is the default behavior for backward compatibility.
**Enable relationship fields (skip.objectN.relationship.fields=false):**
Set `sskip.objectN.relationship.fields=false` to process relationship fields in your records. Relationship fields must follow Salesforce’s external ID format (for example, `ParentObject__r.ExternalId__c`).
Example configuration:
```json
{
"skip.objectN.relationship.fields": "false"
}
```
**Important considerations:**
* Relationship fields must reference the external ID of the related object (for example, `Account__r.AccountNumber__c`), not the Salesforce internal ID.
* The related object must exist in Salesforce before the relationship can be established.
* The connector currently does not support the format `SomeObject__r.ExternalId__c` for referential fields in all scenarios. Contact [Confluent Support](https://support.confluent.io/) if you encounter issues.
For more information about Salesforce relationship fields, see [Salesforce Relationship Fields documentation](https://developer.salesforce.com/docs/atlas.en-us.object_reference.meta/object_reference/relationships_among_objects.htm).
### Error handling and troubleshooting
#### Why is the success reporter topic not receiving messages?
The connector writes success and error records to reporter topics. If the success topic is not receiving messages, several issues may be occurring:
**Common causes:**
* **CSV record mapping failure**: The connector failed to map Salesforce API responses back to the original Kafka records. This can happen when special characters or formatting in CSV data causes mismatches.
You may see this log message:
```none
failed to map API response to sink record. We will miss reporting record to success topic.
```
* **CSV data normalization**: Salesforce Bulk API may normalize certain values (for example, `#N/A` literals) which prevents the connector from matching responses to original records.
* **Schema mismatches**: If the Salesforce object schema changed, the connector may fail to properly process success responses.
**Resolution:**
1. **Check connector logs**: Look for warnings about mapping failures or CSV processing errors.
2. **Verify CSV formatting**: Ensure your Kafka records do not contain special CSV characters that could cause parsing issues.
3. **Restart the connector**: A connector restart may resolve transient mapping issues.
4. **Review field values**: Avoid using special literals like `#N/A` in field values that could be normalized by Salesforce.
If the issue persists, contact [Confluent Support](https://support.confluent.io/) with your connector ID and sample records that are failing to map.
#### What should I do if the connector fails with `Index out of bounds` errors?
Index out of bounds errors can occur due to internal processing issues with CSV batch handling:
```none
java.lang.IndexOutOfBoundsException: Index: X, Size: Y
```
This is typically a transient error that may occur during:
* Processing of malformed CSV data.
* Handling of Salesforce API responses with unexpected formats.
* Internal state inconsistencies during batch processing.
**Resolution:**
1. **Restart the connector**: The connector should recover automatically after restart.
2. **Check for schema changes**: Verify that the Salesforce object schema matches the connector’s expectations.
3. **Review recent data**: Check if recently produced records have any unusual formatting or values.
4. **Contact support**: If the error persists after restart, contact [Confluent Support](https://support.confluent.io/) with the connector ID and error logs.
### API limits and Salesforce quotas
#### What are the Salesforce API limits for Bulk API 2.0?
Salesforce enforces several limits on Bulk API 2.0 operations that can affect the connector:
* **Daily batch limit**: Maximum of 15,000 batches per 24-hour rolling window per Salesforce organization.
* **Job data size**: Maximum of 100 MB per batch (with base64 encoding).
* **Concurrent jobs**: Salesforce limits the number of concurrent Bulk API jobs that can run simultaneously.
* **API request limits**: General API request limits apply to authentication and metadata operations.
To monitor your Salesforce API usage:
1. Navigate to **Setup > System Overview > API Usage** in your Salesforce organization.
2. Review current API consumption and remaining daily limits.
3. Set up alerts in Salesforce to notify you when approaching limits.
For detailed information, see [Salesforce Bulk API Limits](https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/asynch_api_concepts_limits.htm) and [Salesforce Bulk API 2.0 Sink Connector](limits.md#cc-salesforce-bulk-api-v2-sink-limits).
#### What happens when I exceed Salesforce API limits?
If you exceed Salesforce API limits:
1. **Connector failure**: The connector fails with an error indicating which limit was exceeded.
2. **Automatic retry**: The connector uses `ExponentialBackOff` for retry with `request.max.retries.time.ms` configuration. After the retry time expires, the task is marked as failed.
3. **Data backlog**: Records accumulate in the Kafka topic, increasing consumer lag.
**To prevent exceeding limits:**
* Optimize batching configuration to reduce the number of batches created.
* Monitor API usage regularly in the Salesforce Setup console.
* Distribute workload across multiple Salesforce organizations if possible.
* Use multiple Salesforce user accounts to distribute API quota.
## 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.
[](https://docs.confluent.io/platform/current/tutorials/examples/cloud-etl/docs/index.html)