1. Home
  2. Kafka Connectors
  3. Salesforce Bulk API Source Connector for Confluent Platform

Salesforce Bulk API Source Connector for Confluent Platform

Note

If you are using Confluent Cloud, see Salesforce Bulk API Source Connector for Confluent Cloud for the cloud Quick Start.

The Salesforce Bulk API Source connector integrates Salesforce.com with Apache Kafka®.

The Salesforce Bulk API Source connector provides the capability to pull records and capture changes from Salesforce.com via Salesforce Bulk Query API.

Salesforce Objects are standard salesforce objects. The SalesforceBulkApiSourceConnector can be used to pull objects/capture changes and write them to Kafka. You can use this connector with either standalone or distributed Connect workers.

Features

The Salesforce Bulk API Source connector includes the following features:

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.

Supports one task

The Salesforce Bulk API Source connector supports running only one task.

Configuration Properties

For a complete list of configuration properties for this connector, see Salesforce Bulk API Source Connector Configuration Properties.

Considerations

Note the following when using the Salesforce Bulk API source connector.

Restarting

When the connector operates, it periodically records the last query time execution in the Connect offset topic. When the connector is restarted, it fetches Salesforce objects with a LastModifiedDate that is later than last queried time.

API Limits

The Salesforce Bulk API connector is limited by non compound fields. For Example, Bulk Query doesn’t support address, location fields. The connector will discard the address and geolocation fields.

SObjects Limitation

The following Salesforce object (SObject) error message may be displayed when you are using the Salesforce Bulk API Source connector:

Entity 'Order' is not supported to use PKChunking.

Set the configuration property batch.enable=false for these SObjects. This property is available for Salesforce Bulk API Source connector version 1.7.0 (or later).

Salesforce Object Support

The following objects from Salesforce are supported in this version of Kafka Connect Salesforce Bulk API Source connector:

  • Account
  • Campaign
  • CampaignMember
  • Case
  • Contact
  • Contract
  • Event
  • Group
  • Lead
  • Opportunity
  • OpportunityContactRole
  • OpportunityLineItem
  • Period
  • PricebookEntry
  • Product2
  • Task
  • TaskFeed
  • TaskRelation
  • User
  • UserRole

The Kafka Connect Salesforce Bulk API Source connector also supports custom objects with non-compound fields.

The following objects are not supported by Salesforce Bulk API:

  • Feed (e.g. AccountFeed, AssetFeed, …)
  • Share (e.g. AccountBrandShare, ChannelProgramLevelShare, …)
  • History (e.g. AccountHistory, ActivityHistory, …)
  • EventRelation (e.g. AcceptedEventRelation, DeclinedEventRelation, …)
  • AggregateResult
  • AttachedContentDocument
  • CaseStatus
  • CaseTeamMember
  • CaseTeamRole
  • CaseTeamTemplate
  • CaseTeamTemplateMember
  • CaseTeamTemplateRecord
  • CombinedAttachment
  • ContentFolderItem
  • ContractStatus
  • EventWhoRelation
  • FolderedContentDocument
  • KnowledgeArticleViewStat
  • KnowledgeArticleVoteStat
  • LookedUpFromActivity
  • Name
  • NoteAndAttachment
  • OpenActivity
  • OwnedContentDocument
  • PartnerRole
  • RecentlyViewed
  • ServiceAppointmentStatus
  • SolutionStatus
  • TaskPriority
  • TaskStatus
  • TaskWhoRelation
  • UserRecordAccess
  • WorkOrderLineItemStatus
  • WorkOrderStatus

Quick Start

In this quick start, the Salesforce Bulk API source connector is used to import data from Salesforce to Kafka.

  1. Create Salesforce developer account using this link if you don’t have it.
  2. Add records to the objects by clicking on App Launcher and selecting the required Salesforce object.

Install the connector through the Confluent Hub Client.

# run from your CP installation directory
confluent-hub install confluentinc/kafka-connect-salesforce-bulk-api:latest

Tip

By default, the connector will install the plugin into the share/confluent-hub-components directory and add the directory to the plugin path. For the plugin path change to take effect, you must restart the Connect worker.

Start the services using the Confluent CLI.

confluent local start

Every service starts in order, printing a message with its status.

Starting Zookeeper
Zookeeper is [UP]
Starting Kafka
Kafka is [UP]
Starting Schema Registry
Schema Registry is [UP]
Starting Kafka REST
Kafka REST is [UP]
Starting Connect
Connect is [UP]
Starting KSQL Server
KSQL Server is [UP]
Starting Control Center
Control Center is [UP]

Note

The SalesforceBulkApiSourceConnector supports a single task only.

Property-based example

Create a configuration file, salesforce-bulk-api.properties.This configuration is used typically along with standalone workers.

name=SalesforceBulkApiSourceConnector
tasks.max=1
connector.class=io.confluent.connect.salesforce.SalesforceBulkApiSourceConnector
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081
salesforce.username=< Required Configuration >
salesforce.password=< Required Configuration >
salesforce.password.token=< Required Configuration >
salesforce.object=< Required Configuration >
salesforce.since=< Required Configuration >
kafka.topic=< Required Configuration >
salesforce.instance=< Required Configuration >
confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1
confluent.license=Omit to enable trial mode

Before starting the connector, make sure that the configurations in salesforce-bulk-api.properties are properly set.

Then start the Salesforce Bulk API source connector by loading its configuration with the following command.

Caution

You must include a double dash (--) between the connector name and your flag. For more information, see this post.

confluent local load salesforce-bulk-api-source -- -d salesforce-bulk-api.properties
{
   "name" : "SalesforceBulkApiSourceConnector",
   "config" : {
     "connector.class", "io.confluent.connect.salesforce.SalesforceBulkApiSourceConnector",
     "tasks.max" : "1",
     "key.converter": "org.apache.kafka.connect.storage.StringConverter",
     "value.converter": "io.confluent.connect.avro.AvroConverter",
     "value.converter.schema.registry.url": "http://localhost:8081",
     "kafka.topic" : "< Required Configuration >",
     "salesforce.password" : "< Required Configuration >",
     "salesforce.password.token" : "< Required Configuration >",
     "salesforce.object" : "< Required Configuration >",
     "salesforce.username" : "< Required Configuration >",
     "salesforce.since" : "< Required Configuration >",
     "confluent.topic.bootstrap.servers": "localhost:9092",
     "confluent.topic.replication.factor": "1",
     "confluent.license": ""
   },
  "tasks": []
 }

Check that the connector started successfully. Review the Connect worker’s log by entering the following:

confluent local log connect

Confirm that the connector is in a RUNNING state.

confluent local status SalesforceBulkApiSourceConnector

Confirm that the messages are being sent to Kafka.

kafka-avro-console-consumer \
    --bootstrap-server localhost:9092 \
    --property schema.registry.url=http://localhost:8081 \
    --topic <topic-name> \
    --from-beginning | jq '.'

REST-based example

This configuration is used typically along with 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 Kafka Connect REST Interface for more information.

Connect Distributed REST example:

{
  "name" : "SalesforceBulkApiSourceConnector",
  "config" : {
    "connector.class": "io.confluent.connect.salesforce.SalesforceBulkApiSourceConnector",
    "tasks.max" : "1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081",
    "kafka.topic" : "< Required Configuration >",
    "salesforce.password" : "< Required Configuration >",
    "salesforce.password.token" : "< Required Configuration >",
    "salesforce.object" : "< Required Configuration >",
    "salesforce.username" : "< Required Configuration >",
    "salesforce.since" : "< Required Configuration >",
    "confluent.topic.bootstrap.servers": "localhost:9092",
    "confluent.topic.replication.factor": "1",
    "confluent.license": " Omit to enable trial mode "
  }
}

Note

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

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

Create a new connector:

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

Update an existing connector:

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

Sample data formats

The following examples show the JSON document structure of a Salesforce Bulk Query results (for Contact object) as it received by the Saleforce Connector, converted to a Kafka record, and then stored in a topic.

Raw JSON received from Salesforce Bulk Query:

[
{
  "attributes" : {
      "type" : "Contact",
      "url" : "/services/data/v47.0/sobjects/Contact/0032v00002qXTBlAAO"
   },
  "Id" : "0032v00002qXTBlAAO",
  "IsDeleted" : false,
  "MasterRecordId" : null,
  "AccountId" : "0012v00002RkgUVAAZ",
  "LastName" : "Gonzalez",
  "FirstName" : "Rose",
  "Salutation" : "Ms.",
  "Name" : "Rose Gonzalez",
  "OtherStreet" : null,
  "OtherCity" : null,
  "OtherState" : null,
  "OtherPostalCode" : null,
  "OtherCountry" : null,
  "OtherLatitude" : null,
  "OtherLongitude" : null,
  "OtherGeocodeAccuracy" : null,
  "MailingStreet" : "313 Constitution Place\nAustin, TX 78767\nUSA",
  "MailingCity" : null,
  "MailingState" : null,
  "MailingPostalCode" : null,
  "MailingCountry" : null,
  "MailingLatitude" : null,
  "MailingLongitude" : null,
  "MailingGeocodeAccuracy" : null,
  "Phone" : "(512) 757-6000",
  "Fax" : "(512) 757-9000",
  "MobilePhone" : "(512) 757-9340",
  "HomePhone" : null,
  "OtherPhone" : null,
  "AssistantPhone" : null,
  "ReportsToId" : null,
  "Email" : "rose@edge.com",
  "Title" : "SVP, Procurement",
  "Department" : "Procurement",
  "AssistantName" : null,
  "LeadSource" : "Trade Show",
  "Birthdate" : "1967-07-14",
  "Description" : null,
  "OwnerId" : "0052v00000ajtG3AAI",
  "CreatedDate" : 1564636138000,
  "CreatedById" : "0052v00000ajtG3AAI",
  "LastModifiedDate" : 1564636138000,
  "LastModifiedById" : "0052v00000ajtG3AAI",
  "SystemModstamp" : 1564636138000,
  "LastActivityDate" : null,
  "LastCURequestDate" : null,
  "LastCUUpdateDate" : null,
  "LastViewedDate" : 1573528066000,
  "LastReferencedDate" : 1573528066000,
  "EmailBouncedReason" : null,
  "EmailBouncedDate" : null,
  "IsEmailBounced" : false,
  "PhotoUrl" : "/services/images/photo/0032v00002qXTBlAAO",
  "Jigsaw" : null,
  "JigsawContactId" : null,
  "CleanStatus" : "Pending",
  "IndividualId" : null,
  "Level__c" : "Primary",
  "Languages__c" : "English"
}
]

Kafka record value:

{
  "Id":"0032v00002qXTBlAAO",
  "IsDeleted":{
    "boolean":false
  },
  "MasterRecordId":null,
  "AccountId":{
    "string":"0012v00002RkgUVAAZ"
  },
  "LastName":{
    "string":"Gonzalez"
  },
  "FirstName":{
    "string":"Rose"
  },
  "Salutation":{
    "string":"Ms."
  },
  "Name":{
    "string":"Rose Gonzalez"
  },
  "OtherStreet":null,
  "OtherCity":null,
  "OtherState":null,
  "OtherPostalCode":null,
  "OtherCountry":null,
  "OtherLatitude":null,
  "OtherLongitude":null,
  "OtherGeocodeAccuracy":null,
  "MailingStreet":{
    "string":"313 Constitution Place\nAustin, TX 78767\nUSA"
  },
  "MailingCity":null,
  "MailingState":null,
  "MailingPostalCode":null,
  "MailingCountry":null,
  "MailingLatitude":null,
  "MailingLongitude":null,
  "MailingGeocodeAccuracy":null,
  "Phone":{
    "string":"(512) 757-6000"
  },
  "Fax":{
    "string":"(512) 757-9000"
  },
  "MobilePhone":{
    "string":"(512) 757-9340"
  },
  "HomePhone":null,
  "OtherPhone":null,
  "AssistantPhone":null,
  "ReportsToId":null,
  "Email":{
    "string":"rose@edge.com"
  },
  "Title":{
    "string":"SVP, Procurement"
  },
  "Department":{
    "string":"Procurement"
  },
  "AssistantName":null,
  "LeadSource":{
    "string":"Trade Show"
  },
  "Birthdate":{
    "int":-903
  },
  "Description":null,
  "OwnerId":{
    "string":"0052v00000ajtG3AAI"
  },
  "CreatedDate":{
    "long":1564636138000
  },
  "CreatedById":{
    "string":"0052v00000ajtG3AAI"
  },
  "LastModifiedDate":{
    "long":1564636138000
  },
  "LastModifiedById":{
    "string":"0052v00000ajtG3AAI"
  },
  "SystemModstamp":{
    "long":1564636138000
  },
  "LastActivityDate":null,
  "LastCURequestDate":null,
  "LastCUUpdateDate":null,
  "LastViewedDate":{
    "long":1573722558000
  },
  "LastReferencedDate":{
    "long":1573722558000
  },
  "EmailBouncedReason":null,
  "EmailBouncedDate":null,
  "IsEmailBounced":{
    "boolean":false
  },
  "PhotoUrl":{
    "string":"/services/images/photo/0032v00002qXTBlAAO"
  },
  "Jigsaw":null,
  "JigsawContactId":null,
  "CleanStatus":{
    "string":"Pending"
  },
  "IndividualId":null,
  "Level__c":{
    "string":"Primary"
  },
  "Languages__c":{
    "string":"English"
  }
}