Confluent Cloud

Sync Kafka topics from a Confluent Cloud cluster into your warehouse, one row per message, with Schema Registry decoding.

Confluent Cloud is in beta. Each topic the API key can describe becomes a warehouse table, and each message becomes one row.

Before you connect

In Confluent Cloud, create a service account for Contrast and a cluster API key owned by it. Grant the service account READ and DESCRIBE on the topics to sync. A prefixed ACL, such as every topic starting with orders., works. Contrast needs no cluster or write access. It keeps its own offsets, so it cannot move a consumer group your applications use.

To see each sync's consumer lag in Confluent Cloud, also grant the service account READ on consumer groups with the prefix contrast-. This is optional. See Consumer lag.

To decode Avro, Protobuf or JSON Schema messages, also create a Schema Registry API key for the environment with read access to the topics' subjects (DeveloperRead in RBAC). Cluster keys and Schema Registry keys are separate.

The cluster must accept connections over the public internet. Clusters reachable only through private networking (PrivateLink, VPC peering or Transit Gateway) cannot be connected.

Connect

You need to be a workspace administrator to start a sync. Editors can connect the cluster.

Open the Confluent Cloud setup link.
Paste the bootstrap server from Cluster settings → Endpoints, such as pkc-abc12.us-east-1.aws.confluent.cloud:9092, and the cluster API key and secret.
To decode schema-encoded messages, choose Decode with Schema Registry and paste the Schema Registry endpoint (https://psrc-….confluent.cloud) with its API key and secret.
Choose where each topic's history begins: All retained messages, Last 7 days (the week before you create the sync) or New messages only (messages produced after you create the sync). The starting point is fixed when the sync is created, so a full resync later rereads the same range.
Select topics and start the sync.

Contrast checks the key against the cluster and Schema Registry before saving. The source is then bound to that cluster's ID: a key rotated later must belong to the same cluster. Credentials are stored encrypted.

Tables

Topics appear in the confluent schema. Internal topics, such as _schemas and __consumer_offsets, are not listed.

Column	Contents
`id`	`<topic id>:<partition>:<offset>`, the row key
`kafka_topic_id`, `kafka_partition`, `kafka_offset`	Where the message is in Kafka
`kafka_timestamp`, `kafka_timestamp_type`	The message timestamp, and whether the producer (`create_time`) or broker (`log_append_time`) set it
`key`, `key_format`, `key_schema_id`	The key as text: its exact text for plain keys, JSON for schema-encoded keys, or base64 when `key_format` is `binary`
`value`, `value_format`, `value_schema_id`	The decoded value as JSON
`value_base64`	The raw value, when it could not be decoded
`value_size_bytes`	The value's size in Kafka
`headers`	Headers in order, as `[{"key": ..., "value": ...}]`; non-UTF-8 header values use `value_base64`
`is_tombstone`	`true` for a message with a null value
`decode_error`	Why part of the message could not be decoded or stored

value_format records how each value was read:

avro, protobuf and json_schema use the message's Schema Registry schema, including schema references. Avro decimals and Protobuf 64-bit integers become strings, bytes become base64, and Avro timestamps become ISO 8601 text.
json for plain JSON, string for other UTF-8 text, and binary for anything else, including serialized numbers such as a Java Long (kept in value_base64). Plain-text keys keep their exact text.
schema_registry for a schema-encoded message that could not be decoded, and omitted for a value over the size limit.

Sync behavior

Each scheduled sync reads every selected partition up to its last stable offset (the end of committed transactions) at the time the sync starts. Messages produced while it runs are read by the next sync. Rows become queryable as each batch finishes. A retry resumes from the last committed batch and merges by id, so retried messages are not duplicated.

Each batch holds up to 50,000 messages or about 16 MiB of rows. In our tests a topic with a million 350-byte JSON messages copied at about 2,000 messages per second, taking about eight minutes the first time.

Transactions. Only committed messages are read. Aborted messages and transaction markers never become rows. A transaction still open when a sync starts ends that partition's read early. Its messages are read after it commits.
New partitions. Partitions added to a topic are read from their first message.
Retention. Messages that retention removes before Contrast reads them are skipped and cannot be recovered, so keep each topic's retention well above the sync interval. Rows already in the warehouse stay when Kafka deletes their messages.
Compaction and tombstones. Tombstones are kept as rows with is_tombstone = true, not as deletes. Messages compacted away before they were read do not appear.
Recreated topics. If a topic is deleted and recreated with the same name, the new topic is read from its start. Its rows have a new topic ID, so rows from the old topic are not overwritten.
Undecodable messages. A message that cannot be decoded keeps its raw bytes in value_base64 and its reason in decode_error. It never stops a sync. If you add Schema Registry credentials later, run Actions → Full resync to decode earlier messages that Kafka still retains, from the sync's starting point onward.
Full resync. A full resync rereads each topic from the starting point chosen when the sync was created and merges by id. It does not remove rows.

Consumer lag

After each batch is saved, Contrast records the sync's position as the offsets of the consumer group contrast-<sync ID>. The sync's page shows the group name under Consumer group. In Confluent Cloud, open the cluster's Clients page to see the group's lag: the messages not yet in your warehouse. Lag grows as messages arrive between syncs and returns to zero when a sync catches up. The group has no active members, because Contrast never joins it.

Without READ on contrast- groups, syncs run normally and the group does not appear.
The group only displays progress. Contrast never reads it back, so resetting its offsets changes nothing. To reread messages, run Actions → Full resync.
After you delete a sync, its group stays until Kafka expires its offsets, after 7 days by default.

Limits

One value can be up to 8 MiB (raw or decoded), and a key or its headers up to 1 MiB each. Larger parts are not stored and decode_error says which; the rest of the row is kept.
Messages are decompressed and decoded in an isolated reader with a 768 MiB memory ceiling. A batch or message that exceeds it, for example a highly compressed batch that expands to gigabytes, stops that table's sync with an error naming the topic and, when known, the partition and offset. Remove or replace the message in Kafka, then retry.
Avro arrays whose items take no space (such as an array of null) are not decoded; those messages keep their raw bytes.
Authentication uses cluster API keys (SASL/PLAIN over TLS). OAuth and mTLS are not supported.
A rejected or deleted API key, a missing READ ACL or a deleted topic stops the table's sync with an error that names the fix. These are not retried until you correct them. Network failures and broker outages are retried from the last committed batch.

See Confluent's documentation on API keys, ACLs and Schema Registry wire format.