How to Load & Connect Kafka to Salesforce?

Integrating Kafka with Salesforce can be a game-changer for your data workflows. As businesses look to streamline their processes, syncing real-time data from Kafka to Salesforce opens up endless possibilities for automation, analytics, and customer insights.

In this blog, I’ll walk you through how this integration works, why it’s beneficial, and the steps you need to get started. Whether you’re looking to enhance your CRM or improve data flow, this guide will make the process smooth and straightforward. Let’s dive in!

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.

Prerequisites

• Working knowledge of Salesforce.
• Working knowledge of Kafka.
• A general idea about Apex programming language.
• A general idea about databases and their operations.

Methods to Move Data From Kafka to Salesforce

1. Salesforce Streaming APIs allow Kafka to capture the real-time events from Salesforce via an HTTP connection. T
2. he data is available in the form of a Salesforce Object (sObject).
3. Users can easily modify the data by performing various operations such as deleting, updating, and creating new events. You can use a query written in Salesforce Object Query Language (SOQL) to retrieve information about various data events as per the need.

This method can be implemented using the following steps:

• Step 1: Creating a PushTopic record in Salesforce
• Step 2: Installing the Kafka connector for Salesforce
• Step 3: Configuring the Salesforce Streaming Events
• Step 4: Configuring the ETL Pipeline for connecting Kafka to Salesforce

Step 1: Creating a PushTopic record in Salesforce

The easiest way to create a new PushTopic record is by using the Salesforce Developer Console, an integrated development environment that you can use to create, debug, and test applications in Salesforce.
From the debug menu, select open execute anonymous window in the Developers Console and paste the following Apex code to create a PushTopic record: Sign Up for a 14-day free trial and experience the feature-rich Hevo suite firsthand.

PushTopic pushTopic = new PushTopic();
pushTopic.Name = 'ContactUpdates';
pushTopic.Query = 'SELECT Id, Name FROM Contact';
pushTopic.ApiVersion = 36.0;
pushTopic.NotifyForOperationCreate = true;
pushTopic.NotifyForOperationUpdate = true;
pushTopic.NotifyForOperationUndelete = true;
pushTopic.NotifyForOperationDelete = true;
pushTopic.NotifyForFields = 'Referenced';
insert pushTopic;

Using the above Apex code, you will be able to create a PushTopic called ContantUpadates, that will create a new event every time a user, creates or modifies or deletes a contact. The PushTopic will send the Id and name of the contact the user is trying to access.

Step 2: Installing the Kafka connector for Salesforce

To install the Kafka connector for Salesforce, start your Kafka server and use the following command:

sudo npm install -g salesforce-kafka-connect

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

Step 3: Configuring the Salesforce Streaming Events

To configure the Kafka connector, begin by using the following command:

sudo nano /usr/lib/node_modules/salesforce-kafka-connect/config/default.js

The configuration file will now open up as follows. Replace the username and password fields with your credentials to allows Kafka to access the data associated with your account.

"use strict";
const path = require("path");
const config = {
    kafka: {
        zkConStr: "localhost:2181/",
        logger: null,
        groupId: "kc-salesforce-group",
        clientName: "kc-salesforce-client",
        workerPerPartition: 1,
        options: {
            sessionTimeout: 8000,
            protocol: ["roundrobin"],
            fromOffset: "earliest", //latest
            fetchMaxBytes: 1024 * 100,
            fetchMinBytes: 1,
            fetchMaxWaitMs: 10,
            heartbeatInterval: 250,
            retryMinTimeout: 250,
            autoCommit: true,
            autoCommitIntervalMs: 1000,
            requireAcks: 1,
            //ackTimeoutMs: 100,
            //partitionerType: 3
        }
    },
    topic: "sf-test-topic",
    partitions: 1,
    maxTasks: 1,
    maxPollCount: 5,
    pollInterval: 250,
    produceKeyed: true,
    produceCompressionType: 0,
    connector: {
        username: "user",
        password: "password",
        loginUrl: "https://user.salesforce.com",
        streamingSource: {
            batchSize: 5,
            topic: "StreamingTopic",
            kafkaTopic: "sf-test-topic",
            idProperty: "id"
        },
        restSink: {
            sObject: "sobject",
            idProperty: "id"
            batchSize: 500
        }
    },
    http: {
        port: 3149,
        middlewares: []
    },
    enableMetrics: true
};
module.exports = config;

Perform the same operation and replace the username and password parameters with your credentials in test-config and source-config files. You can access the files by using the following file paths:

/usr/lib/node_modules/salesforce-kafka-connect/test/sink-config.js

/usr/lib/node_modules/salesforce-kafka-connect/test/source-config.js

Once you have configured your Salesforce data source to create events whenever the Salesforce Streaming APIs produces a request, you now need to set up a sink, that will retrieve the data from Salesforce and transfer it to Kafka. You can make use of the Salesforce-Kafka-Connect package to implement this. You can open the index.js using the following file path:

/usr/lib/node_modules/salesforce-kafka-connect/index.js

The streaming of Salesforce data into Kafka is handled by the runSourceConnector function.

Once you have called the function, you need to transfer data from Salesforce, use the following command in your command-line interface: Sign Up for a 14-day free trial and experience the feature-rich Hevo suite firsthand.

nkc-salesforce-source --help

Step 4: Configuring the ETL Pipeline for connecting Kafka to Salesforce

To fetch data from the Salesforce Streaming APIs and store it into Kafka and then transform the data into an outgoing Salesforce PoducerRecord, you need to set up the Kafka sink. The runSinkConnector function can help you performs this operation.

You can use the following command in your command-line interface to call this function:

nkc-salesforce-sink --help

Once you have called the function, you now need to transfer your Salesforce data into the Kafka sink. To do this, you can create a function called messageToProducerRecord as follows:

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