Salesforce CDC

​

Overview

• Native CDC (default) — Streamkap subscribes directly to the Salesforce Change Data Capture stream using OAuth credentials. No Apex code required.
• Apex-trigger webhook — Apex triggers in your org POST change events to a Streamkap-generated webhook URL. Use this for standard objects without CDC support, custom objects, or when you need fine-grained control.

​

Prerequisites

• A Salesforce org (Developer, Professional, Enterprise, or Unlimited edition) with System Administrator access.
• For native CDC: the Change Data Capture feature enabled in your org and selected for the objects you want to capture (Setup → Change Data Capture).
• OAuth Consumer Key and Consumer Secret from a Salesforce Connected App or External Client App (see Salesforce Setup below).
• A Streamkap workspace with permission to create source connectors.

​

How It Works

1. OAuth authentication — Streamkap exchanges your Consumer Key and Secret for a Salesforce access token.
2. CDC subscription — When Native CDC is enabled, Streamkap subscribes to one channel per selected object (/data/AccountChangeEvent, /data/ContactChangeEvent, …) and streams events as they happen.
3. Webhook ingestion (optional) — If you also publish from Apex triggers, those POSTs hit the same connector at a generated webhook URL and flow through the same router.
4. Routing — Each event carries an entity name (from the CDC ChangeEventHeader or your Apex payload). The router maps it to a topic of the form {Object}_events — Account_events, Contact_events, etc.
5. Key extraction — The Salesforce record Id becomes the Kafka message key, enabling upsert-style consumption.
6. Delete handling — DELETE and GAP_DELETE change types set __deleted: true on the record so downstream sinks can issue tombstones.

​

Streamkap Setup

​

1. Create the Source

1. Navigate to Sources and choose Salesforce CDC.
2. Give the source a memorable Name (for example, salesforce-prod).

​

2. Connection Settings (Auth tab)

​

3. Schema (Salesforce Objects)

Account, Contact, Lead, Opportunity, Case, Task, Event, User, Campaign,
CampaignMember, Contract, Order, OrderItem, Product2, Pricebook2,
PricebookEntry, Asset, Note, Attachment, ContentDocument, ContentVersion,
FeedItem, Solution, OpportunityLineItem, OpportunityContactRole,
AccountContactRole, Partner, Quote, QuoteLineItem

​

4. Settings

​

5. Save

​

Salesforce Setup

​

OAuth Connected App (required for both modes)

1. In Salesforce, go to Setup → App Manager → New External Client App (or New Connected App on older orgs).
2. Configure:
   • App name — Streamkap CDC.
   • Contact Email — your email.
   • Distribution State — Local.
   • API (Enable OAuth Settings) — checked.
   • Callback URL — https://login.salesforce.com/services/oauth2/callback (any valid URL works; the connector does not use it).
   • OAuth scopes — add:
     • Manage user data via APIs (api)
     • Full access (full)
     • Perform requests at any time (refresh_token, offline_access)
   • Uncheck Require Proof Key for Code Exchange (PKCE).
   • If you want password-less auth, enable Client Credentials Flow and select a Run-As user with the necessary object permissions.
3. Save and wait 2–10 minutes for Salesforce to provision the app.
4. Open the app and reveal the Consumer Key and Consumer Secret (you may need to click Manage Consumer Details to reveal the secret). Paste these into the Streamkap source as Client ID and Client Secret.
5. Confirm the user (either the Run-As user for client-credentials flow, or the Username/Password user) has API Enabled, View All Data (or the relevant object permissions), and is allowed to log in from Streamkap’s IP range if you have IP restrictions.

​

Username / password fallback

• Username — the Salesforce user’s login email. We recommend a dedicated integration user with API Enabled, the relevant read permissions, and no MFA requirement.
• Password + Token — concatenate the user’s password and security token directly with no separator. To get a security token, log in as that user and go to Settings → My Personal Information → Reset My Security Token — Salesforce emails a new token.

​

Native CDC Setup

1. In Salesforce, go to Setup → Change Data Capture.
2. Move the objects you want to stream from Available Entities to Selected Entities.
3. Save. Salesforce begins publishing change events to the per-object channels (/data/AccountChangeEvent, etc.).
4. In Streamkap, ensure Enable Native CDC is on and the same objects are listed under Salesforce Objects.

​

Apex-trigger Webhook (optional)

​

1. Create the External Credential

1. Setup → Named Credentials → External Credentials → New.
2. Fill in:
   • Label — Streamkap Webhook.
   • Name — Streamkap_Webhook.
   • Authentication Protocol — Custom.
3. Save.
4. Under Principals, click New:
   • Parameter Name — Default.
   • Identity Type — Named Principal (read-only).
   • Sequence Number — 1.
5. Save.
6. Open the Default principal and under Authentication Parameters click Add:
   • Parameter 1 Name — x-api-key.
   • Parameter 1 Value — paste the API Key from the Streamkap source.
7. Save.

​

2. Create the Named Credential

1. Click the Named Credentials tab → New.
2. Fill in:
   • Label — Streamkap.
   • Name — Streamkap.
   • URL — paste the Streamkap Webhook URL, with no path appended.
   • External Credential — select Streamkap Webhook.
3. Save.

​

3. Grant Access via Permission Set

1. Setup → Permission Sets → New:
   • Label — Streamkap Webhook Access.
   • Save.
2. Inside the permission set, open External Credential Principal Access.
3. Click Edit, move Streamkap_Webhook - Default into Enabled, save.
4. Click Manage Assignments → Add Assignment, select the user(s) that will execute the triggers, and assign.
5. Have those users log out and back in to refresh their session.

​

4. Create the Apex Sender Class

1. Setup → Apex Classes → New, paste the code below, and save:

public class WebhookSender {

    @future(callout=true)
    public static void sendAsync(String recordId, String changeType, String objectName) {
        Map<String, Schema.SObjectField> fieldMap = Schema.getGlobalDescribe()
            .get(objectName).getDescribe().fields.getMap();

        List<String> fieldNames = new List<String>();
        for (String fieldName : fieldMap.keySet()) {
            Schema.DescribeFieldResult fieldDesc = fieldMap.get(fieldName).getDescribe();
            if (fieldDesc.isAccessible()) {
                fieldNames.add(fieldDesc.getName());
            }
        }

        String soql = 'SELECT ' + String.join(fieldNames, ', ')
                     + ' FROM ' + objectName
                     + ' WHERE Id = \'' + String.escapeSingleQuotes(recordId) + '\'';

        List<SObject> records = Database.query(soql);
        if (records.isEmpty()) {
            sendPayload(recordId, objectName, changeType, new Map<String, Object>{'Id' => recordId});
            return;
        }

        Map<String, Object> recordMap = new Map<String, Object>();
        SObject record = records[0];
        for (String fieldName : fieldNames) {
            recordMap.put(fieldName, record.get(fieldName));
        }

        sendPayload(recordId, objectName, changeType, recordMap);
    }

    @future(callout=true)
    public static void sendDeleteAsync(String recordId, String objectName, String recordJson) {
        Map<String, Object> recordMap = (Map<String, Object>) JSON.deserializeUntyped(recordJson);
        sendPayload(recordId, objectName, 'DELETE', recordMap);
    }

    private static void sendPayload(String recordId, String objectName,
                                     String changeType, Map<String, Object> recordFields) {
        Map<String, Object> header = new Map<String, Object>{
            'entityName' => objectName,
            'recordIds' => new List<String>{recordId},
            'changeType' => changeType,
            'commitTimestamp' => System.currentTimeMillis()
        };

        Map<String, Object> payload = new Map<String, Object>();
        payload.put('ChangeEventHeader', header);
        payload.putAll(recordFields);

        Map<String, Object> body = new Map<String, Object>{
            'data' => new Map<String, Object>{
                'payload' => payload,
                'event' => new Map<String, Object>{'replayId' => 0}
            },
            'channel' => '/data/' + objectName + 'ChangeEvent'
        };

        HttpRequest req = new HttpRequest();
        req.setEndpoint('callout:Streamkap');
        req.setMethod('POST');
        req.setHeader('Content-Type', 'application/json');
        req.setBody(JSON.serialize(body));
        req.setTimeout(30000);

        Http http = new Http();
        try {
            HttpResponse res = http.send(req);
            if (res.getStatusCode() != 200) {
                System.debug(LoggingLevel.ERROR, 'Webhook failed: ' + res.getStatusCode() + ' ' + res.getBody());
            }
        } catch (Exception e) {
            System.debug(LoggingLevel.ERROR, 'Webhook error: ' + e.getMessage());
        }
    }
}

​

5. Create One Trigger per Object

trigger AccountToKafka on Account (after insert, after update, after delete, after undelete) {
    List<SObject> records;
    String changeType;

    if (Trigger.isInsert) {
        changeType = 'CREATE';
        records = Trigger.new;
    } else if (Trigger.isUpdate) {
        changeType = 'UPDATE';
        records = Trigger.new;
    } else if (Trigger.isDelete) {
        changeType = 'DELETE';
        records = Trigger.old;
    } else if (Trigger.isUndelete) {
        changeType = 'UNDELETE';
        records = Trigger.new;
    }

    for (SObject rec : records) {
        if (changeType == 'DELETE') {
            WebhookSender.sendDeleteAsync(rec.Id, 'Account', JSON.serialize(rec));
        } else {
            WebhookSender.sendAsync(rec.Id, changeType, 'Account');
        }
    }
}

​

Event Routing Reference

​

Change Data Capture (CDC)

​

PushTopic events

​

Platform events

​

Message keys

​

Example Payload

{
  "data": {
    "payload": {
      "ChangeEventHeader": {
        "entityName": "Account",
        "recordIds": ["001Hp00002abcDE"],
        "changeType": "UPDATE",
        "commitTimestamp": 1764000000000
      },
      "Id": "001Hp00002abcDE",
      "Name": "Acme Corp",
      "Phone": "555-0101",
      "BillingCity": "San Francisco"
    },
    "event": { "replayId": 12345 }
  },
  "channel": "/data/AccountChangeEvent"
}

{
  "Id": "001Hp00002abcDE",
  "Name": "Acme Corp",
  "Phone": "555-0101",
  "BillingCity": "San Francisco",
  "__deleted": false
}

​

Common Patterns

​

State table (upsert) ingestion

• Flatten Record Fields — true
• Flatten Prefix — (empty)
• Include Change Metadata — false

​

Audit log ingestion

• Flatten Record Fields — false (or true, depending on your destination)
• Include Change Metadata — true

​

Troubleshooting

​

Limitations

• The Salesforce CDC source is currently Beta.
• Native CDC requires that each object be enabled for Change Data Capture in your Salesforce org. Objects without CDC support must use Apex-trigger mode.
• Maximum payload size is 50 MB per request and maximum header size is 64 KB.
• Each request is treated as a single Kafka record; the connector does not batch multiple events into one Kafka message.
• The connector runs as a single task; horizontal scaling requires multiple source instances.
• Initial backfill of historical records is not performed automatically — only changes from the time the source starts are streamed.

​

See Also

• Webhook Source — generic webhook source for any HTTP-capable producer.
• Zendesk Webhook — Zendesk-specific webhook source.