Most enterprises treat Salesforce as a system of record for the business, and a data platform (a Lakehouse, a warehouse, a set of streaming jobs) as the place where that business data gets joined, enriched, and analyzed. The gap between the two is deceptively simple to state — "get Salesforce changes into Kafka in real time" — and surprisingly easy to get wrong. This post walks through the architecture of doing it properly: subscribing to Salesforce Change Data Capture (CDC), landing every change event in Kafka without loss, and feeding a medallion-style lakehouse downstream. It is design-focused; the code is illustrative, not a copy-paste tutorial. The problem: there is no cable you can just plug-in Salesforce publishes change events through its Pub/Sub API. This is not a REST endpoint you poll and not a Kafka broker you can point a consumer at. It is a gRPC service over HTTP/2, streaming Avro-encoded binary payloads, at api.pubsub.salesforce.com:7443. Its three RPCs matter to us: Subscribe — a bidirectional stream. You send FetchRequest messages to request N events (flow control), and the server streams events back. GetSchema — returns the Avro schema for a given schemaId. GetTopic — metadata about a channel. Kafka, on the other side, speaks its own binary protocol and knows nothing about gRPC or Salesforce Avro. Nothing in a standard data platform bridges these two natively. REST-based Salesforce connectors exist, but they poll objects — they are not a CDC event subscription and will miss the ordering, deletes, and low latency that CDC gives you. So the first architectural truth is: Between the Salesforce Pub/Sub API and Kafka you must run a small bridge component that subscribes over gRPC, decodes Avro, and produces to Kafka. The good news: Salesforce ships an official Pub/Sub API client library, so the bridge is closer to a wrapper than a from-scratch protocol implementation. What a change event actually looks like Before choosing where to run the bridge, it helps to see the payload. Every CDC event carries a stable envelope — the ChangeEventHeader — plus the record fields. A CREATE on an Account-like object decodes to roughly this: { "ChangeEventHeader": { "entityName": "Account", "recordIds": ["001XX000003ABCDEAG"], "changeType": "CREATE", "commitNumber": 1785576707554279428, "commitTimestamp": 1785576707000, "sequenceNumber": 1, "transactionKey": "000047dd-d8ed-e5e5-1b48-498b18084963", "changedFields": [] }, "Name": "Acme Corp", "Industry": null, "OwnerId": "005XX000001AbcdYAR" } Two things are worth internalizing: The envelope is platform-wide and fixed. changeType (CREATE / UPDATE /  DELETE / UNDELETE), recordIds, commitNumber, commitTimestamp, sequenceNumber, transactionKey, and the changedFields bitmap are the same for every object in every org. Your downstream logic can be written once against this contract. The payload fields are per-object. Which fields appear — and which custom  fields (*__c) show up — depends on which objects the org enabled CDC for. This is the one part you cannot know from documentation; it comes from the specific org's configuration. That split (fixed envelope, variable payload) is what makes a generic, reusable pipeline possible. The resume mechanism: replayId Every event carries a replayId — an opaque byte bookmark marking the event's position in the stream. It is conceptually identical to a Kafka offset or a database CDC checkpoint (an SCN-style ordering clock, if you come from the Oracle world). When you subscribe you choose a ReplayPreset: LATEST — only events from now on. EARLIEST — from the start of the retention window. CUSTOM — resume after a specific replayId. This is the one that prevents data loss. The operational scenario is the whole ballgame. Your bridge stops — a deploy, a crash, a network blip — and Salesforce keeps producing events the whole time. If you restart with LATEST, everything produced during the outage is gone. If instead you persisted the replayId of the last event you successfully wrote to Kafka, you restart with CUSTOM from that bookmark and backfill the entire gap. Two properties fall out of this, and you should state both plainly to stakeholders: No loss. Resume-from-bookmark guarantees outage events are recovered. At-least-once, not exactly-once. The Pub/Sub API delivers at least once. If the bridge writes to Kafka but dies before persisting the replayId, that event replays on restart. Duplicates at the boundary are normal and expected. You make the pipeline idempotent downstream by deduplicating on (recordId, commitNumber). The goal is "no loss," never "zero duplicates." Where should the bridge run? Comparing the options There are several defensible places to run the subscribe-decode-produce logic. Condensed, the realistic choices are: OptionShapeVerdict A Bridge writes straight to the serving store (no Kafka) Loses replay/buffer; couples ingestion to storage B Bridge → Kafka, then Flink downstream Good, but "bridge" is undefined ops-wise C Bridge → Kafka → routing tool → store Extra hop, little gain D Kafka Connect Source connector → Kafka → Flink Recommended E Standalone microservice → Kafka → Flink Works, but you own a service lifecycle Option D — implement the bridge as a Kafka Connect Source connector — wins for one decisive reason: the framework already solves the hard part. In the Kafka Connect model, a SourceTask.poll() returns records each carrying a sourceOffset. Connect persists that offset automatically to an internal offsets topic and hands it back on restart. So the replayId ↔ offset bookkeeping — the thing you would otherwise write, test, and get subtly wrong — becomes map replayId onto the Connect source offset and let the framework do it. You also inherit the rest of the Connect operational surface for free: Horizontal scale via tasks. If you must subscribe to many orgs or many channels, Connect distributes tasks across workers. This matters enormously at fan-out (think hundreds of subscriptions). Managed lifecycle, config, and monitoring through the Connect REST API and whatever streaming-management UI your platform ships. Dead-letter handling for poison records without killing the task. A standalone microservice (Option E) can do all of this too — but then you are building offset persistence, scaling, config management, and health checks. On a platform that already runs Kafka Connect (for example, the Streams Messaging stack in Cloudera Data Platform, which ships an Apache-based Kafka Connect managed by SMM), Option D is strictly less code you own. Here is the architecture:  Enable the CDC feature in Salesforce Click the [setup] icon in SFDC menu. Search Change Data Capture / 変更データキャプチャ Then configure what entity to sync. (PoC) Use Python to subscribe the events and then save in Kafka # pubsub_cdc_poc.py — Salesforce Pub/Sub API で CDC を購読する最小 PoC (CDP 非依存) import io, queue import avro.schema, avro.io import grpc, requests import pubsub_api_pb2 as pb2 import pubsub_api_pb2_grpc as pb2_grpc # --- 1. OAuth ログイン (PoC=username-password / 生産は JWT bearer 推奨) --- def sfdc_login(login_url, cid, csecret, user, pw): r = requests.post(f"{login_url}/services/oauth2/token", data={ "grant_type": "password", "client_id": cid, "client_secret": csecret, "username": user, "password": pw}) # pw = パスワード + security token r.raise_for_status(); j = r.json() org_id = j["id"].split("/")[-2] # id=".../<orgId>/<userId>" → orgId=tenantId return j["access_token"], j["instance_url"], org_id TOKEN, INSTANCE_URL, TENANT_ID = sfdc_login("https://login.salesforce.com", ...) # --- 2. gRPC チャネル (443, TLS/HTTP2) + 認証メタデータ --- channel = grpc.secure_channel("api.pubsub.salesforce.com:443", grpc.ssl_channel_credentials()) stub = pb2_grpc.PubSubStub(channel) AUTH = (("accesstoken", TOKEN), ("instanceurl", INSTANCE_URL), ("tenantid", TENANT_ID)) TOPIC = "/data/活動対象__ChangeEvent" # ← 自定义对象の CDC 通道 # --- 3. Avro スキーマ取得 (schema_id ごとにキャッシュ) --- _cache = {} def avro_schema(schema_id): if schema_id not in _cache: info = stub.GetSchema(pb2.SchemaRequest(schema_id=schema_id), metadata=AUTH) _cache[schema_id] = avro.schema.parse(info.schema_json) return _cache[schema_id] def decode(schema, payload): return avro.io.DatumReader(schema).read(avro.io.BinaryDecoder(io.BytesIO(payload))) # --- 4. Subscribe (双方向ストリーミング + フロー制御) --- def run(): reqs = queue.Queue() # 初回: LATEST から。再開時は replay_preset=CUSTOM + replay_id=<保存した replayId> reqs.put(pb2.FetchRequest(topic_name=TOPIC, replay_preset=pb2.ReplayPreset.LATEST, num_requested=10)) def req_iter(): while True: yield reqs.get() for resp in stub.Subscribe(req_iter(), metadata=AUTH): for ev in resp.events: rec = decode(avro_schema(ev.event.schema_id), ev.event.payload) h = rec["ChangeEventHeader"] print(h["changeType"], h["entityName"], h["recordIds"], h["commitTimestamp"]) save_replay_id(ev.replay_id) # ★これを永続化 → 断点续接 if resp.pending_num_requested == 0: # 残枠が尽きたら補充 (フロー制御) reqs.put(pb2.FetchRequest(topic_name=TOPIC, num_requested=10)) # 継続時は preset 不要 run() I ran this on my MBP, it works. (Production) Use Kafka Connect to subscribe the events and then save in Kafka In a Product environment, we can create a Kafka Connect Source Connector Source code example: // SalesforcePubSubSourceConnector.java — 外殻。topic(entity)/org 単位で task 分割 public class SalesforcePubSubSourceConnector extends SourceConnector { private Map<String,String> props; @Override public void start(Map<String,String> p) { this.props = p; } @Override public Class<? extends Task> taskClass() { return SalesforcePubSubSourceTask.class; } @Override public List<Map<String,String>> taskConfigs(int maxTasks) { List<String> topics = Arrays.asList(props.get("salesforce.topics").split(",")); // 購読トピック(=entity×org)を task へ配分。230 org はここで水平分割。 List<Map<String,String>> cfgs = new ArrayList<>(); for (List<String> g : ConnectorUtils.groupPartitions(topics, Math.min(maxTasks, topics.size()))) { Map<String,String> c = new HashMap<>(props); c.put("salesforce.topics", String.join(",", g)); cfgs.add(c); } return cfgs; } @Override public void stop() {} @Override public ConfigDef config() { return CONFIG_DEF; } @Override public String version() { return "0.1.0"; } } // SalesforcePubSubSourceTask.java — 内核は developerforce/pub-sub-api の Java stub をラップ public class SalesforcePubSubSourceTask extends SourceTask { private PubSubClient client; // gRPC channel(443)+OAuth(JWT) のラッパー private String kafkaTopic; private final BlockingQueue<SourceRecord> buffer = new LinkedBlockingQueue<>(); @Override public void start(Map<String,String> props) { kafkaTopic = props.get("kafka.topic"); // 例: sfdc.toyotec.katsudo client = new PubSubClient(props); for (String t : props.get("salesforce.topics").split(",")) { // partition に org を入れておくと 230 org でも offset が衝突しない Map<String,Object> partition = Map.of("sfdcTopic", t, "org", props.get("org.id")); // ★ 再開: 前回 replayId を offset storage から復元 → CUSTOM で続き Map<String,Object> off = context.offsetStorageReader().offset(partition); byte[] replayId = off == null ? null : ((ByteBuffer) off.get("replayId")).array(); ReplayPreset preset = replayId == null ? ReplayPreset.LATEST : ReplayPreset.CUSTOM; client.subscribe(t, preset, replayId, this::onEvent); // 非同期ストリーム } } // gRPC で届いた変更イベント → SourceRecord にして buffer へ private void onEvent(String sfdcTopic, String org, byte[] replayId, GenericRecord ev) { Map<String,Object> partition = Map.of("sfdcTopic", sfdcTopic, "org", org); Map<String,Object> offset = Map.of("replayId", ByteBuffer.wrap(replayId)); // ★肝 buffer.offer(new SourceRecord(partition, offset, kafkaTopic, Schema.STRING_SCHEMA, recordId(ev), // key = org+PK 推奨(保序) /* value schema */ null, ev.toString())); // 実際は Avro/Struct 変換 } @Override public List<SourceRecord> poll() throws InterruptedException { List<SourceRecord> batch = new ArrayList<>(); SourceRecord first = buffer.poll(1, TimeUnit.SECONDS); if (first != null) { batch.add(first); buffer.drainTo(batch, 500); } return batch; // ← framework が offset(replayId) を自動 commit } @Override public void stop() { if (client != null) client.close(); } @Override public String version() { return "0.1.0"; } } Downstream: one CDC paradigm, reused Once change events are in Kafka, the downstream is not Salesforce-specific anymore — it is just CDC, the same shape you already handle for database sources. A Flink job reads the topic, inspects ChangeEventHeader.changeType, and routes: Silver (current state) — upsert/delete by recordId into a mutable store such as Kudu. This is the "what does the record look like now" table. Bronze (history) — append every event to Iceberg for full audit, replay, and time travel. Gold — modeled, business-ready tables built from Silver/Bronze. The strategic payoff: if you already run an Oracle/DB CDC pipeline into the same medallion layout, Salesforce CDC folds into the identical Flink logic rather than becoming a second, parallel implementation. You unify on one CDC engine and avoid re-implementing history, logical deletes, and metadata handling twice. The failure modes worth designing for up front An honest architecture names what can go wrong. Three things dominate: Schema evolution. The envelope is stable, but payloads gain and lose fields as admins change objects. Register schemas and enforce backward- compatible evolution so added fields are absorbed harmlessly. This is the most common cause of "the job suddenly can't parse an event." Poison records. Route un-parseable events to a dead-letter topic instead of crashing the task; fix and replay later. Because the raw event is already durably in Kafka (and optionally landed raw in Bronze), a transform you can't handle today is never data loss — you reprocess from the offset once the logic is fixed. Kafka + Bronze is your replayable source of truth. Authentication for unattended runs. Interactive OAuth is fine for a laptop proof-of-concept, but a long-running connector needs the JWT Bearer flow (a connected app with a certificate) so it can mint and refresh tokens headlessly. Decide this before implementation — it has a lead time (certificates, admin setup). And one environmental prerequisite that stalls more projects than any code bug: network egress. Your platform typically lives in a private network; it needs an allowed outbound TLS path to api.pubsub.salesforce.com:7443. Confirm it early. Proving it before you build it You do not need the full connector to de-risk the design. A minimal bridge — a ~200-line script that authenticates, calls Subscribe, decodes Avro with the schema fetched via GetSchema, produces JSON to a local single-node Kafka, and writes the last replayId to a state file — is enough to validate the two things that actually carry risk: No loss under steady state: create a known batch of records, confirm every recordId lands and the changeType counts match. Resume correctness: start the bridge, kill it, make several changes while it is down, restart, and confirm the downtime events all arrive via CUSTOM replay — with boundary duplicates converging idempotently on (recordId, commitNumber). That local, self-contained test reproduces the production design's core behavior (what Kafka Connect will later automate) and turns "we think this works" into "we watched it work." It is also the cheapest possible artifact to show a skeptical stakeholder. Takeaways The Salesforce Pub/Sub API is gRPC + Avro; you need a bridge to reach Kafka. There is no native connector, and REST polling is not CDC. Implement the bridge as a Kafka Connect Source connector so the framework handles offset persistence, scaling, and lifecycle — map replayId onto the Connect source offset. Design for at-least-once: guarantee no loss via replay, and make the downstream idempotent on (recordId, commitNumber). Keep the raw event in Kafka (and Bronze) so any transform failure is reprocessable, never lost. Settle schema evolution, dead-lettering, JWT auth, and network egress before you write the connector — those, not the gRPC plumbing, are where projects actually stall. The plumbing is interesting, but the durable lesson is the same one that applies to every CDC integration: make ingestion lossless and replayable, push idempotency downstream, and reuse one change-data paradigm across all your sources. Salesforce just happens to speak gRPC on the way in. ... View more