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.
In Confluent Cloud, create a service account for Embrasure 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. Embrasure needs no consumer group, cluster or write access. It keeps its own offsets and never commits to a consumer group, so it cannot move a group your applications use.
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.
pkc-abc12.us-east-1.aws.confluent.cloud:9092, and the cluster API key and secret.https://psrc-….confluent.cloud) with its API key and secret.Embrasure 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.
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.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.
is_tombstone = true, not as deletes. Messages compacted away before they were read do not appear.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.id. It does not remove rows.decode_error says which; the rest of the row is kept.null) are not decoded; those messages keep their raw bytes.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.