Kafka topics

In short

Institutions with their own streaming platform can publish to a Kafka topic instead of calling the API. SpeculaPro’s connector reads the topic and puts every record through exactly the same checks, currency conversion, encryption and duplicate protection as the ingest API.

• Each record is one item: one transaction, customer, account, account event or FX rate, as plain JSON, exactly as it would appear in the API’s items array.
• Each topic carries one kind of item. You map it under Settings → Data sources. See Data sources.
• A record that fails the checks goes to a dead-letter topic, unchanged, with the reasons in its headers.
• Nothing is lost on a restart: SpeculaPro marks a record as read only once it is accepted or dead-lettered.

On a self-hosted installation, the connector role must be running. See Connector.

Which cluster?

The connector reads the Kafka cluster your operator configures (SP_CONNECTOR_KAFKA_BROKERS). Left empty, that is SpeculaPro’s own cluster, the usual set-up for a self-hosted installation: you publish to your topics on the same cluster SpeculaPro uses internally. Workspaces only choose which topics to read; they can’t point SpeculaPro at another cluster.

The records

The record’s value is one JSON object, with the same fields, allowed values and limits as an item sent to the API:

{"external_id":"TX-20261009-000001","occurred_at":"2026-10-09T08:15:00Z","customer_external_id":"C-1001","direction":"out","type":"transfer","channel":"momo","amount_minor":150000,"currency":"GHS"}

Topic carries	Fields as in
Transactions	Transactions
Customers	Customers
Accounts	Accounts
Account events	Account events
FX rates	FX rates

• Don’t wrap it in {"items": [...]}: one record, one item.
• Fields SpeculaPro doesn’t know are refused, as by the API, with unexpected property.
• The record key isn’t read, but it decides the partition and so the order. Key records by the item’s external_id (or by customer or account) so that updates to the same thing keep their order.

How reading works

• New topics are read from the beginning. When you map a topic, SpeculaPro starts at its oldest record. Records it has seen before are recognised by their external_id and reported as duplicates, so mapping a topic with history is safe.
• At least once, in effect exactly once. The connector marks records as read (commits offsets) only after every record it fetched together has been accepted, or written to the dead-letter topic. If SpeculaPro’s pipeline is unavailable, it waits and tries again, from a quarter of a second up to 30 seconds apart, without marking anything read. After a restart, unmarked records are read again and recognised as duplicates.
• Changes to mappings apply within 30 seconds. The connector re-reads the mappings every SP_CONNECTOR_KAFKA_RELOAD_INTERVAL.
• It slows down rather than overwhelming monitoring. If the processes that store and monitor transactions fall more than SP_CONNECTOR_KAFKA_PAUSE_LAG records behind (500,000 by default), the connector pauses reading every topic, and resumes when they are back below half that. Your records simply wait on the topic.

The dead-letter topic

A record is rejected when it isn’t JSON, isn’t an object, fails the API’s checks, or is rejected by ingest (for example an external_id reused with different content). It is then written, unchanged (same key, same value, same original headers), to the mapping’s dead-letter topic: <topic>.dlq unless you choose another. SpeculaPro tries to create the dead-letter topic if it doesn’t exist; if your cluster doesn’t allow that, create it yourself.

SpeculaPro adds these headers before the original ones:

Header	Value
sp-source-topic	The topic the record came from
sp-source-partition	Its partition
sp-source-offset	Its offset
sp-rejected-at	When it was rejected (UTC, RFC 3339)
sp-data-source	The ID of the data source that read it, when there was one
sp-error	One header per reason, each up to 1,000 characters

Fix the record and publish it again to the source topic. Because the original headers are kept, a replay tool can strip the sp- headers and republish as is.

Watch the dead-letter topic

Records on it are not monitored. Alert on anything arriving there, as you would on rejected records from the API.

Who may read which topic

A topic is how SpeculaPro knows which workspace a record belongs to. The rules depend on how your operator has set up the installation:

Set-up	How it works	Who it suits
One institution (the default)	A workspace can map any topic name. Each topic belongs to one workspace: once one maps it, no other can	A self-hosted installation for one institution
A namespace per workspace (SP_CONNECTOR_KAFKA_TOPIC_NAMESPACE set, for example inst.)	A workspace can map only topics named <namespace><workspace slug>.<name>, such as inst.kweku-pay.transactions, and its dead-letter topic must follow the same pattern. The broker’s access rules must let each institution’s credentials write only to its own namespace: the topic name is then what proves which workspace a record comes from	Several institutions on one cluster, such as SaaS
Shared topics (SP_CONNECTOR_KAFKA_SHARED_TOPICS)	Several workspaces can map the same topic. Each record names its workspace in an sp-workspace header (the workspace’s ID or slug), and is read only if that workspace has mapped the topic. A record without the header, or naming a workspace that hasn’t mapped the topic, is dead-lettered	A bridge run by your operator that fans in records for several workspaces

Shared topics trust whoever writes to them

Anyone who can write to a shared topic can put records into any workspace that maps it, just by naming it in the header. Only producers your operator controls may write to shared topics; never give an institution write access to one. Each institution writes to its own namespaced topic instead.

SpeculaPro’s own topics can never be mapped, and can never be a dead-letter topic either. They are recognised by their prefix (sp. by default) and by name, so they stay protected even if an operator sets an empty topic prefix.

A dead-letter topic also can’t be a shared topic, and with a namespace it must be inside the workspace’s own namespace, whatever the mode. This matters because rejected records are copied there unchanged: if the dead-letter topic were one that SpeculaPro or another source reads as trusted input, a record crafted to fail the checks would reach it without being validated, encrypted or tied to the right workspace.

How to read a topic

1. Agree with your operator which topic name to use. If your installation uses a namespace, the form under Settings → Data sources shows the prefix your topics must start with.
2. Under Settings → Data sources, select Add data source, choose Kafka topic, and pick what the topic carries.
3. Enter the Topic and, if you want, a Dead-letter topic. Select Save.
4. Open the data source and select Test connection. It checks the topic exists and that SpeculaPro can read it, and says how many partitions and records it has.
5. Publish a few records and watch Rows loaded and Rejected on the data source. A data source reading a topic also shows how many records are Waiting.

Related

• Data sources
• File drops (SFTP and S3)
• Responses and statuses
• Idempotency and retries
• Connector
• Kafka (self-hosting)