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

# Salesforce Platform Events Source Connector for Confluent Platform

The Kafka Connect Salesforce [Platform Events](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro_emp.htm) Source connector captures events and writes them to Apache Kafka®. This connector can be used with either standalone or distributed Connect workers.

#### NOTE
* The connector’s `salesforce.platform.event.name` property indicates the
  name of the event, and must end with `__e`. For example,
  `salesforce.platform.event.name=UserLogout__e`. These platform events must
  exist in Salesforce before running the connector. For more information, see [Salesforce
  Developer Guide](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_define.htm).
* The connector supports Salesforce up to API version 65.0.

## Features

The Salesforce Platform Events Source Connector includes the following features:

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

<a id="platform-events-source-at-least-once-delivery"></a>

### At least once delivery

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

<a id="platform-events-source-supports-one-task"></a>

### Supports one task

The Salesforce Platform Events Source Connector supports running only one task.

<a id="platform-events-source-csfle-sm"></a>

### Client-side encryption

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

<a id="salesforce-platformevent-source-prerequisites"></a>

## Prerequisites

Before you configure the connector, ensure the Salesforce user account has the
following permissions:

* **API Enabled** permission.
* **Read** access to the specific Platform Event.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Salesforce Platform Events Source Connector Configuration Properties](salesforce_platformevent_source_connector_config.md#salesforce-platformevent-source-connector-config).

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

## Considerations

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

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

### Replaying Messages

High-volume platform event messages are [stored for 72 hours (3 days)](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro_emp.htm),
and standard-volume platform event messages are [stored for 24 hours (1 day)](https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_intro_emp.htm).
When you create a connector, you can control how stored events are treated when the connector starts for the first time.

1. Consume only new events that arrive in the PushTopic after the connector starts up by setting `salesforce.initial.start` to `latest`. This is the default behavior, due largely to [API limits](../pushtopics/overview.md#salesforce-pushtopic-limits).
2. Consume all of the events in the PushTopic by setting `salesforce.initial.start` to `all`.

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

### Restarting

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

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

### Retries

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

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

### Retries Example

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

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

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

## Limitations

The following limitations apply to this Salesforce connector:

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

#### NOTE
To enable OAuth JWT bearer token support, set the following connector
configuration properties:

- `salesforce.username`
- `salesforce.consumer.key`
- `salesforce.jwt.keystore.path`
- `salesforce.jwt.keystore.password`

<a id="salesforce-platformevent-source-connector-examples"></a>

## Examples

#### NOTE
`salesforce.consumer.key` and `salesforce.consumer.secret` are required properties used for OAuth2 secure authentication by Salesforce.com. Additional information and tutorials are available at [Salesforce.com](https://developer.salesforce.com/docs/atlas.en-us.api_streaming.meta/api_streaming/code_sample_auth_oauth.htm/).

<a id="salesforce-platformevent-source-connector-property-examples"></a>

### Property based example

This configuration is used typically along with [standalone workers](/platform/current/connect/concepts.html#standalone-workers).

```properties
 name=SFDCPlatformEvents1
 connector.class=io.confluent.salesforce.SalesforcePlatformEventSourceConnector
 tasks.max=1
 kafka.topic=< Required Configuration >
 salesforce.consumer.key=< Required Configuration >
 salesforce.consumer.secret=< Required Configuration >
 salesforce.password=< Required Configuration >
 salesforce.password.token=< Required Configuration >
 salesforce.platform.event.name=< Required Configuration >
 salesforce.username=< Required Configuration >
 salesforce.initial.start=all
 confluent.topic.bootstrap.servers=localhost:9092
 confluent.topic.replication.factor=1
 confluent.license=
```

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

<a id="salesforce-platformevent-source-connector-rest-examples"></a>

### REST-based example

This configuration is used typically along with [distributed workers](/platform/current/connect/concepts.html#distributed-workers).
Write the following JSON to `connector.json`, configure all of the required values, and use the command below to
post the configuration to one of the distributed connect worker(s).
See Kafka Connect [REST API](/platform/current/connect/references/restapi.html) for more information.

**Connect Distributed REST example with Platform Event:**

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

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

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

**Create a new connector:**

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

**Update an existing connector:**

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

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

## Sample data formats

The following examples show the JSON document structure of a Salesforce Platform
Event as it received by the Salesforce Connector, converted to a Kafka record,
and then stored in a topic. The Platform Event has four custom fields:
`NumberField` , `TextField`, `BooleanField`, and `TextArea`. The `__c`
suffix in those fields is appended by Salesforce when the Platform Event is
defined. Platform Events do not generate topic keys in Kafka.

**Raw Platform Event received from Salesforce:**

```json
{
  "data": {
    "schema": "VrdbV-4eqHB3hllFiCjIrw",
    "payload": {
      "NumberField__c": 33,
      "CreatedById": "0051U000004EE5wQAG",
      "CreatedDate": "2019-03-25T19:43:11.336Z",
      "TextField__c": "This is a Text field",
      "BooleanField__c": true,
      "TextArea__c": "This is a TextArea field"
    },
    "event": {
      "replayId": 44479
    }
  },
  "channel": "/event/ConfluentEvent__e"
}
```

**Kafka record value for the Salesforce Platform Event:**

```json
{
  "ReplayId": {
    "string": "44479"
  },
  "CreatedDate": {
    "long": 1553542991336
  },
  "CreatedById": {
    "string": "0051U000004EE5wQAG"
  },
  "TextField__c": {
    "string": "This is a Text field"
  },
  "NumberField__c": {
    "double": 33
  },
  "BooleanField__c": {
    "boolean": true
  },
  "TextArea__c": {
    "string": "This is a TextArea field"
  },
  "_EventType": {
    "string": "VrdbV-4eqHB3hllFiCjIrw"
  }
}
```

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

## Troubleshooting Connector and Task Failures

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

**Authorization Failures**

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

**Salesforce Error Codes**

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

**Repeated Connection Failures**

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

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

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

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

## Enabling Debug Logging

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

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

**On-Premises Installation**

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

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

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

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

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

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

**Docker**

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

```bash
CONNECT_LOG4J2_ROOT_LOGLEVEL=DEBUG
```

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

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

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

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