How to Load & Connect Kafka to Salesforce: Step-by-Step Explanation

Moving data from Kafka to Salesforce can really boost your CRM with real-time updates. The problem? Kafka’s high-throughput, distributed messaging system doesn’t natively sync with Salesforce’s structured, object-based environment, often leaving teams wrestling with complex middleware or manual processes to bridge the gap. This can lead to delays, data inconsistencies, and frustrated users.

So, in this blog, I’ll show you two easy ways to make it work. First, we’ll look at a Confluent-based Salesforce Kafka Connector that’s super straightforward. Then, we’ll try a custom API method for those of you who like to tweak things yourself.

Let’s get started!

What Is Kafka?

Kafka is a distributed streaming platform that allows you to publish, subscribe to, store, and process streams of records in real-time. It’s designed to handle large volumes of data, making it ideal for applications that require high throughput, low latency, and fault tolerance. Whether you’re building real-time analytics, monitoring systems, or event-driven applications, Kafka helps you manage and process data streams efficiently.

Key Features

• Scalability: Kafka can scale horizontally by adding more brokers to your cluster, allowing it to handle increasing data loads.
• High Throughput: Kafka’s architecture is optimized for high write and read throughput, making it suitable for large-scale data streaming.
• Durability: Data in Kafka is stored in a fault-tolerant manner, ensuring it is preserved even in case of hardware failures.
• Fault Tolerance: Kafka replicates data across multiple nodes, ensuring no data is lost if a broker goes down.
• Real-time Processing: Kafka enables you to process data in real-time, making it ideal for time-sensitive applications.

Introduction to Salesforce

Salesforce is a leading cloud-based Customer Relationship Management (CRM) platform that helps businesses manage their relationships with customers, streamline processes, and improve overall productivity. It provides a comprehensive suite of tools to automate tasks, track customer interactions, and enhance data-driven decision-making. It also offers cloud-based tools such as data analytics, and IoT products.

Key Features of Salesforce

• Sales Cloud: Manages sales processes, tracks opportunities, and automates workflows.
• Service Cloud: Enhances customer support with case management and self-service portals.
• Marketing Cloud: Personalizes marketing campaigns and provides data-driven insights.
• Commerce Cloud: Powers e-commerce operations and integrates sales, marketing, and support.
• Community Cloud: Builds online communities for collaboration and knowledge sharing.
• Analytics Cloud (Tableau CRM): Advanced data analytics and visualization tools for better decision-making.

Why Send Data from Apache Kafka to Salesforce?

Sending data from Kafka to Salesforce enables real-time customer insights, automation, and analytics. It helps businesses trigger instant sales & marketing workflows, centralize customer data, and improve support with proactive issue resolution. Kafka’s scalability ensures seamless integration without performance bottlenecks.

Methods to Move Data From Kafka to Salesforce

Method 1: Using Salesforce Kafka Connector

The Salesforce Kafka Connector, specifically Confluent’s sink connector, enables seamless integration of Kafka with Salesforce. This method pushes data from a Kafka topic into Salesforce as sObjects (e.g., updating or creating Contact records). Confluent’s platform is required to use this connector effectively.

Prerequisites

• Working knowledge of Salesforce, Kafka, Apex, and databases.
• A Kafka topic with data (e.g., JSON records like {“Id”: “003…”, “Name”: “John Doe”})

Step 1: Prepare Salesforce

• Ensure a writable Salesforce object exists (e.g., Contact). Identify its API name (Contact) and fields (e.g., Id, Name). No additional setup like PushTopics is needed.

Step 2: Install the Salesforce Kafka Connector

• This method depends on Confluent’s Kafka Connect framework. Start your Kafka server and Confluent’s Kafka Connect.
• Install the Salesforce Kafka Connector via Confluent Hub:

<code>confluent-hub install confluentinc/kafka-connect-salesforce:latest

Step 3: Configure the Connector

• Create a config file (e.g., salesforce-sink.json):

{
  "name": "salesforce-sink",
  "connector.class": "io.confluent.connect.salesforce.SalesforceSinkConnector",
  "tasks.max": "1",
  "topics": "sf-contact-updates",
  "salesforce.instance": "https://login.salesforce.com",
  "salesforce.username": "your-username",
  "salesforce.password": "your-password-with-security-token",
  "salesforce.sobject": "Contact",
  "salesforce.version": "60.0",
  "key.converter": "org.apache.kafka.connect.json.JsonConverter",
  "value.converter": "org.apache.kafka.connect.json.JsonConverter",
  "key.converter.schemas.enable": "false",
  "value.converter.schemas.enable": "false"
}

• Deploy it to Confluent’s Kafka Connect:

curl -X POST -H "Content-Type: application/json" --data @salesforce-sink.json http://localhost:8083/connectors

This command will download and install the Kafka connector for Salesforce on your system.

Step 4: Produce Data to Kafka

• Send JSON data to the sf-contact-updates topic using the Salesforce Kafka Connector’s expected format:

echo '{"Id": "003...", "Name": "Jane Doe"}' | kafka-console-producer --broker-list localhost:9092 --topic sf-contact-updates

• The connector upserts this data into Salesforce’s Contact object based on the Id.

Step 5: Verify in Salesforce

• Log in to Salesforce and check the Contact object (e.g., via Setup > Object Manager > Contact or SOQL: SELECT Id, Name FROM Contact).
• Confirm the record (e.g., Jane Doe) is updated or inserted.

Please Note:

• Data Format Expectations: Kafka topic data must be JSON with Salesforce-compatible field names (e.g., Name, case-sensitive). The Id field dictates update or insert behavior.
• Confluent Dependency: The Salesforce Kafka Connector requires Confluent’s Kafka Connect platform. Ensure Confluent is installed and running.

Method 2: Using Custom API Integration

This approach involves building a lightweight application (e.g., in Python) that consumes data from Kafka and pushes it to Salesforce via the Salesforce REST API. Unlike Kafka Connect, this gives you full control over the integration logic.

Prerequisites

• Salesforce credentials (username, password, security token) and a Connected App for OAuth.
• Working knowledge of Salesforce, Kafka, Python (or your preferred language), and REST APIs.
• A Kafka topic with data (e.g., JSON records like {“Id”: “003…”, “Name”: “John Doe”}).

Step 1: Set Up Salesforce Authentication

• Create a Connected App in Salesforce (Setup > Apps > App Manager > New Connected App) to get a Client ID and Client Secret.
• Note your Salesforce username, password, and security token.

Step 2: Install Required Tools

• Install Python libraries:

pip install kafka-python simple-salesforce requests

• Ensure Kafka is running with a topic (e.g., sf-contact-updates).

Step 3: Write the Custom Integration Script

• Create a Python script (e.g., kafka_to_salesforce.py):

from kafka import KafkaConsumer
from simple_salesforce import Salesforce
import json

# Salesforce authentication
sf = Salesforce(
    username='your-username',
    password='your-password',
    security_token='your-security-token',
    domain='login'  # Use 'test' for sandbox
)

# Kafka consumer setup
consumer = KafkaConsumer(
    'sf-contact-updates',
    bootstrap_servers=['localhost:9092'],
    auto_offset_reset='earliest',
    value_deserializer=lambda x: json.loads(x.decode('utf-8'))
)

# Consume Kafka messages and send to Salesforce
for message in consumer:
    data = message.value
    print(f"Processing: {data}")
    # Upsert Contact based on Id (update if exists, insert if not)
    sf.Contact.upsert(f"{data['Id']}", {'Id': data['Id'], 'Name': data['Name']})
    print(f"Upserted Contact: {data['Id']}")

• This script:
  • Authenticates with Salesforce using simple-salesforce.
  • Consumes JSON messages from the Kafka topic sf-contact-updates.
  • Upserts data into the Contact object using the Id as the external key.

Step 4: Run the Script

• Start the script:

<code>python kafka_to_salesforce.py

• Produce test data to Kafka:

echo '{"Id": "003...", "Name": "Jane Doe"}' | kafka-console-producer --broker-list localhost:9092 --topic sf-contact-updates

Step 5: Verify in Salesforce

• Check the Contact object in Salesforce (e.g., via Setup > Object Manager > Contact or SOQL: SELECT Id, Name FROM Contact).
• Confirm the record (e.g., Jane Doe) is updated or inserted.

Things to Note:

• Data Flow Direction: This moves data from Kafka to Salesforce. For Salesforce to Kafka, you’d need to poll Salesforce (e.g., via API) and produce to Kafka.
• Flexibility: This custom API method allows tailored logic (e.g., transformations) but requires manual coding and maintenance.
• Data Format: Kafka messages must be JSON with fields matching Salesforce API names (e.g., Name, case-sensitive). The Id field determines upsert behavior.
• Error Handling: Add try-except blocks in production to manage API or Kafka errors.

Conclusion

In conclusion, integrating Kafka with Salesforce is a powerful way to enhance your data-driven decision-making and improve overall efficiency. By seamlessly transferring real-time data between the two platforms, you can ensure that your teams have the most up-to-date information at their fingertips. 

Whether you’re automating workflows or improving customer experiences, this integration helps you unlock new levels of productivity. If you’re ready to take your data management to the next level, Kafka and Salesforce make the perfect pair! Sign up for Hevo’s 14-day free trial and connect Kafka and Salesforce efficiently. 

FAQ on Kafka to Salesforce