--- title: "Kafka Connect" slug: "dev-event-broker-integrations-kafka-connect" description: "Learn how to configure Kafka Connect with VAST Event Broker, using Debezium for CDC, worker settings, and security modes for efficient data integration." updated: 2026-10-30T18:19:01Z published: 2026-10-30T18:19:01Z canonical: "docs.vastdata.com/dev-event-broker-integrations-kafka-connect" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Kafka Connect *Applies to VAST Cluster 5.5.* Kafka Connect runs connectors that move data between Kafka topics and other systems: source connectors write into topics, sink connectors read from them. With VAST Event Broker, the useful case is mostly the source side, for example capturing database changes with Debezium into a topic, which you then query as a VAST DataBase table. This page gives the worker settings that Event Broker needs in distributed mode, the security settings for each authentication mode, what won't work, and a worked example with the Debezium PostgreSQL connector. The examples use Kafka Connect 4.3.1 (Debezium 3.7.0), one worker and unencrypted connections: the worker settings and internal topics on this page, and the Debezium PostgreSQL example end to end. SASL/PLAIN and mTLS, several workers rebalancing, sink connectors, and the failure modes of idempotent and exactly-once settings are described from the Apache Kafka and Debezium documentation and from what Event Broker supports. ## Requirements | Setting | Value | Why | | --- | --- | --- | | `enable.idempotence` (worker) and `producer.enable.idempotence` | `false` | Idempotent producing isn't supported: the broker doesn't offer `InitProducerId`. The Apache Kafka Java producer, which Connect uses, turns idempotence on by default. Set it for the worker's own producers and, with the `producer.` prefix, for source tasks. See [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility#unsupported-features). | | `compression.type` (worker) and `producer.compression.type` | `none` | The broker rejects compressed batches. `none` is the Java producer's default; set it explicitly so a connector default or override can't change it. | | `exactly.once.source.support` | `disabled` (the default) | Exactly-once source support uses Kafka transactions, which Event Broker doesn't support. Don't set `exactly.once.support=required` on connectors either. | | `config.storage.topic`, `offset.storage.topic`, `status.storage.topic` | Topics you create first, with `cleanup.policy=compact` | Connect stores its state in these compacted topics. Event Broker supports `cleanup.policy=compact`. The config topic must have a single partition. | | `*.storage.replication.factor` | Any value, for example `1` | The broker accepts the replication factor and ignores it. Durability comes from VAST storage. | | `consumer.partition.assignment.strategy` | `org.apache.kafka.clients.consumer.RangeAssignor` | Cooperative rebalancing isn't supported. The Java consumer's default list already uses the range assignor; setting it explicitly keeps sink tasks on an eager assignor. | | `bootstrap.servers` | Three VIPs of the Kafka view's pool, on 9092 or 9093 | Three are recommended. Every worker must reach every VIP in the pool, because the broker advertises each VIP as a broker. | ## Worker configuration A minimal `connect-distributed.properties` for an unauthenticated, unencrypted Kafka view: ``` bootstrap.servers=:9092,:9092,:9092 group.id=docs-kafka-connect # Internal topics: create them first (below) config.storage.topic=docs-kafka-connect-configs offset.storage.topic=docs-kafka-connect-offsets status.storage.topic=docs-kafka-connect-status config.storage.replication.factor=1 offset.storage.replication.factor=1 status.storage.replication.factor=1 # Event Broker: no idempotence, no compression, no exactly-once enable.idempotence=false compression.type=none producer.enable.idempotence=false producer.compression.type=none exactly.once.source.support=disabled # Sink tasks: eager assignor consumer.partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor # Plain JSON values, without the Connect schema wrapper key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable=false value.converter.schemas.enable=false plugin.path=/opt/connectors ``` Settings without a prefix apply to the clients the worker uses for its own internal topics. Settings with `producer.` apply to source tasks, and `consumer.` to sink tasks. The admin clients that create topics for source connectors and dead letter queue topics for sinks take the `admin.` prefix. ### Internal topics The Apache Kafka docs recommend creating the three internal topics yourself. The config topic should have one partition; all three should be compacted. Create them before the first worker starts, for example with the Kafka command-line tools: ``` for t in configs:1 offsets:25 status:5; do kafka-topics.sh --bootstrap-server "$KAFKA_BOOTSTRAP" --create \ --topic "docs-kafka-connect-${t%%:*}" --partitions "${t##*:}" \ --replication-factor 1 --config cleanup.policy=compact done ``` The partition counts are the Connect defaults (`offset.storage.partitions=25`, `status.storage.partitions=5`). The partition count of a topic is fixed when it's created, so choose it now. Add `--command-config` with the security settings below when the view requires authentication. Compaction on Event Broker runs in the background, so the topics can hold several records per key for a while. That's also true in Apache Kafka, and Connect reads the latest record per key. See [compaction](/develop/docs/dev-event-broker-guides-topics#compaction). ### Rebalancing Connect workers form a group through the same classic `JoinGroup` and `SyncGroup` requests that consumers use, but they run their own rebalance protocol on the client side. `connect.protocol` defaults to `sessioned`, which, like `compatible`, uses incremental cooperative rebalancing of connectors and tasks (KIP-415, KIP-507). The `eager` value turns that off. Cooperative rebalancing isn't supported for consumer groups, but the broker doesn't inspect the assignment protocol a group uses. Whether `sessioned` works across several workers is part of the test plan. If workers keep rebalancing or tasks aren't reassigned after a worker leaves, set `connect.protocol=eager` on every worker. ### Client config overrides A connector can override its own clients' settings with `producer.override.`, `consumer.override.` and `admin.override.`. Which overrides are allowed is set on the worker by `connector.client.config.override.policy`, whose default in Apache Kafka 4.3 is `All`. With `All`, a connector can set `producer.override.enable.idempotence=true` or a compression type and break itself. To prevent that, set the policy to `None` and keep the Event Broker settings on the worker, or review every connector's overrides. ## Security settings Use the authentication mode your Kafka view requires (see [Security](/develop/docs/dev-event-broker-operations-security)). Each block below goes into the worker configuration three or four times: once without a prefix (the worker's own clients), and again with `producer.`, `consumer.` and `admin.` (the connectors' clients). SASL/PLAIN over TLS, port 9093. The user is a VAST local user: ``` security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; ssl.truststore.type=PEM ssl.truststore.location=/etc/kafka/kafka-ca.pem ``` SASL/PLAIN on an unencrypted connection, port 9092: `security.protocol=SASL_PLAINTEXT`, the same `sasl.*` lines, and no `ssl.*` lines. The password then crosses the network in clear text. TLS without client authentication, port 9093: ``` security.protocol=SSL ssl.truststore.type=PEM ssl.truststore.location=/etc/kafka/kafka-ca.pem ``` mTLS (5.5), port 9093. The certificate's common name (CN) is the VAST user the worker acts as. mTLS needs TLS 1.3, which the Java client enables by default (`ssl.enabled.protocols=TLSv1.2,TLSv1.3`): ``` security.protocol=SSL ssl.truststore.type=PEM ssl.truststore.location=/etc/kafka/kafka-ca.pem ssl.keystore.type=PEM # One PEM file with the private key (PKCS#8) and the certificate chain ssl.keystore.location=/etc/kafka/connect-client.pem ``` Repeated with prefixes, the SASL/PLAIN over TLS case looks like this: ``` security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; ssl.truststore.type=PEM ssl.truststore.location=/etc/kafka/kafka-ca.pem producer.security.protocol=SASL_SSL producer.sasl.mechanism=PLAIN producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; producer.ssl.truststore.type=PEM producer.ssl.truststore.location=/etc/kafka/kafka-ca.pem consumer.security.protocol=SASL_SSL consumer.sasl.mechanism=PLAIN consumer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; consumer.ssl.truststore.type=PEM consumer.ssl.truststore.location=/etc/kafka/kafka-ca.pem admin.security.protocol=SASL_SSL admin.sasl.mechanism=PLAIN admin.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; admin.ssl.truststore.type=PEM admin.ssl.truststore.location=/etc/kafka/kafka-ca.pem ``` The broker's certificate lists the VIPs as IP addresses in its Subject Alternative Names. The Java client checks the host name by default (`ssl.endpoint.identification.algorithm=https`). When identity-policy authorization is on for the view, the worker's user needs, on its internal topics, `KafkaRead`, `KafkaWrite` and `KafkaDescribe`; `KafkaRead` on the worker group; and `KafkaCreate` if the worker or a connector creates topics. Source connectors need `KafkaWrite` on their topics, and sink connectors `KafkaRead` on their topics and group. See [Security](/develop/docs/dev-event-broker-operations-security#actions). ## Source and sink connectors **Source connectors** write into Event Broker topics. Each one needs its topics to exist: there's no automatic topic creation on the broker. Either create the topics first, or let Kafka Connect create them by setting at least `topic.creation.default.replication.factor` and `topic.creation.default.partitions` on the connector (Connect's `topic.creation.enable` is on by default). Leave `topic.creation.default.compression.type` unset: topic-level compression has no effect on Event Broker and doesn't let the producer compress. Any other topic config you add must be one the broker accepts, or topic creation fails with `INVALID_CONFIG`. See [topic configuration](/develop/docs/dev-event-broker-reference-topic-configuration). **Sink connectors** read from Event Broker topics and write to another system. Before you deploy one that lands events in a database or a data lake, consider whether you need it: each topic is already a VAST DataBase table that SQL engines can query. See [Alternatives on VAST](/develop/docs/dev-event-broker-integrations-kafka-connect#alternatives-on-vast). ## What won't work - **Exactly-once source connectors** (`exactly.once.source.support=enabled` with `exactly.once.support=required`). They need transactions. Delivery is at-least-once, so consumers and queries can see duplicates after a task restart. - **Connectors or SMTs that add partitions** to an existing topic, or that change topic configs with `IncrementalAlterConfigs`. - **Static membership** for sink consumers (`consumer.group.instance.id`): accepted but ignored. - **Kafka ACL management** from Connect-based tooling. Access is controlled by identity policies. ## Worked example: Debezium PostgreSQL CDC into a topic The Debezium PostgreSQL connector reads row changes from PostgreSQL's write-ahead log and writes one event per change to a topic per table. In this example, it captures changes to a `public.customers` table into Event Broker. You then query the change events as a VAST DataBase table, and optionally expand their JSON into typed columns with blob expansion. ### Prepare PostgreSQL Debezium needs logical decoding. Per the Debezium docs: - Set `wal_level=logical` on the PostgreSQL server. - The connector's database user needs the `REPLICATION` privilege, and permission to create the publication if the connector creates it. - For update and delete events to carry the full previous row in `before`, set the table's replica identity to `FULL`. With the default, `before` holds only the primary key. ``` psql -c "ALTER TABLE public.customers REPLICA IDENTITY FULL;" ``` ### Create the topic Debezium names topics `<topic.prefix>.<schema>.<table>`. Create the change-event topic as a compacted topic, so it keeps the latest change for each row's key: ``` kafka-topics.sh --bootstrap-server "$KAFKA_BOOTSTRAP" --create \ --topic docs-kafka-connect-cdc.public.customers --partitions 1 \ --replication-factor 1 --config cleanup.policy=compact ``` Leave out `cleanup.policy=compact` if you want the topic to keep every change instead. Retention then removes old changes by age. See [Topics](/develop/docs/dev-event-broker-guides-topics). If you turn on heartbeats, **create the heartbeat topic before you register the connector**. Its name is `<topic.heartbeat.prefix>.<topic.prefix>`, by default `__debezium-heartbeat.docs-kafka-connect-cdc`. Event Broker doesn't create topics automatically, and without the heartbeat topic the connector's producer keeps waiting for it: no change events reach the change topic at all, and the worker logs `UNKNOWN_TOPIC_OR_PARTITION` for the heartbeat topic every second. The name's leading `__` is accepted. Alternatively, set the `topic.creation.default.*` properties on the connector and let Connect create both. ### Register the connector Send the connector configuration to the Connect REST API: ``` curl -s -X PUT -H "Content-Type: application/json" \ http://:8083/connectors/docs-kafka-connect-cdc/config -d '{ "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "plugin.name": "pgoutput", "database.hostname": "", "database.port": "5432", "database.user": "", "database.password": "", "database.dbname": "", "topic.prefix": "docs-kafka-connect-cdc", "table.include.list": "public.customers", "slot.name": "docs_kafka_connect_cdc", "publication.name": "docs_kafka_connect_cdc", "publication.autocreate.mode": "filtered", "tombstones.on.delete": "true", "decimal.handling.mode": "string", "heartbeat.interval.ms": "10000", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.enable": "false", "value.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter.schemas.enable": "false" }' ``` Notes on the settings: | Setting | Why | | --- | --- | | `plugin.name=pgoutput` | The logical decoding plug-in built into PostgreSQL 10 and later. The Debezium default is `decoderbufs`, which must be installed on the server. | | `topic.prefix` | Required. The namespace for all the connector's topic names. The Debezium docs say not to change it later. | | `publication.autocreate.mode=filtered` | Creates the publication for the included tables only. The default, `all_tables`, publishes every table. | | `tombstones.on.delete=true` | The default. A delete produces a delete event followed by a tombstone, so compaction can remove the row's key. | | `decimal.handling.mode=string` | Decimals arrive as strings in the JSON. The default, `precise`, encodes them as binary, which is harder to read in SQL and in blob expansion. | | `heartbeat.interval.ms` | Off by default (`0`). Heartbeats show that the connector is alive while the table is quiet. On a low-traffic database, Debezium also documents `heartbeat.action.query` to stop the write-ahead log growing. | | `schemas.enable=false` | The JSON converter then writes plain JSON values, without the Connect schema wrapper, so the value is the Debezium change event itself. | The PostgreSQL connector has no schema history topic: logical decoding doesn't report DDL changes. Debezium connectors for other databases, such as MySQL, do keep one, set with `schema.history.internal.kafka.topic` and `schema.history.internal.kafka.bootstrap.servers`. That history uses its own producer and consumer, configured with the `schema.history.internal.producer.` and `schema.history.internal.consumer.` prefixes, so repeat the security settings and `enable.idempotence=false` there, and create the history topic first. The worker settings for idempotence and compression apply to the connector's producer. Don't add `producer.override.*` settings that turn them back on. ### What a change event looks like With `schemas.enable=false`, each event's value is a JSON object with Debezium's change-event envelope: | Field | Holds | | --- | --- | | `before` | The row before the change. Null for inserts. With replica identity `FULL`, the whole row; otherwise the key columns only. | | `after` | The row after the change. Null for deletes. | | `source` | Where the change came from: connector, database, schema, table, log position and so on. | | `op` | `c` (insert), `u` (update), `d` (delete), `r` (read, during the initial snapshot), `t` (truncate) or `m` (message). | | `ts_ms` | When the connector processed the change (Debezium 3.x also writes `ts_us` and `ts_ns`). | The event key is the row's primary key, as JSON. A delete is two events with the same key: a change event with `op` `d` and `after` null, then a tombstone, an event with a null value. On a compacted topic, compaction keeps the latest event for each key. So once compaction has run, the topic holds the current state of each row, and a deleted row's tombstone is removed after `delete.retention.ms`. Compaction runs in the background: until it does, the topic can hold several changes for the same row. It can run within seconds: after an insert, an update and a delete, the topic can soon hold only the update and the tombstone. See [compaction](/develop/docs/dev-event-broker-guides-topics#compaction). ### Query the change events as a table The topic is a table in the `kafka_topics` schema of the Kafka view's bucket. Its `key` and `value` columns are binary, so decode them. For example, in Trino, the latest change per customer: ``` SELECT key_json, op, after_json, timestamp_broker FROM ( SELECT from_utf8(key) AS key_json, json_extract_scalar(from_utf8(value), '$.op') AS op, json_format(json_extract(from_utf8(value), '$.after')) AS after_json, timestamp_broker, row_number() OVER (PARTITION BY key ORDER BY timestamp_broker DESC) AS rn FROM vast."/kafka_topics"."docs-kafka-connect-cdc.public.customers" WHERE value IS NOT NULL ) WHERE rn = 1; ``` The `WHERE value IS NOT NULL` skips tombstones. Ordering by `timestamp_broker` stands in for the offset, which isn't a column of the topic table. See [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics) for the SDK and the Query Engine. ### Expand the change events with blob expansion To get typed columns without parsing JSON in every query, attach a [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion) to the topic. Create it before the connector starts, because events already in the topic aren't expanded. The Debezium envelope nests the row under `before` and `after`, so declare them as structs: ``` import pyarrow as pa from vastdb.table import BlobExpansionConfig, ExpansionFormat customer = pa.struct([("id", pa.int64()), ("email", pa.string()), ("balance", pa.string())]) expansion_schema = pa.schema([ ("op", pa.string()), ("ts_ms", pa.int64()), ("before", customer), ("after", customer), ]) # Create the target with the flattened columns: SDK 2.1.0 can't create a table with no columns. flat = pa.schema([("op", pa.string()), ("ts_ms", pa.int64())] + [(f"{side}__{f.name}", f.type) for side in ("before", "after") for f in customer]) with session.transaction() as tx: tx.bucket(EVENTS_BUCKET).create_schema("docs_kafka_connect", fail_if_exists=False) \ .create_table("customers_changes", flat, fail_if_exists=False) with session.transaction() as tx: topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table("docs-kafka-connect-cdc.public.customers") topic.create_blob_expansion( expansion_schema=expansion_schema, target_table_name="customers_changes", target_table_schema="docs_kafka_connect", config=BlobExpansionConfig(expansion_format=ExpansionFormat.JSON, flatten_path=True), ) ``` `balance` is a string because the connector sends decimals as strings (`decimal.handling.mode=string`). Match each field's type to how Debezium encodes that PostgreSQL column type. With `flatten_path=True`, the struct fields become top-level columns such as `after__id` and `after__email`. Without it, `before` and `after` stay nested struct columns. Things to expect in the target table: - `source` isn't in the expansion schema, so every event counts excess fields (along with `ts_us` and `ts_ns`). Add `source` as a struct if you want its fields, or ignore the counter. - Inserts have the `before__*` fields null, and deletes the `after__*` fields. A JSON `null` isn't counted as missing. - A tombstone has no JSON at all. Its row has every expanded field null, with the parse-error flag `0`. Filter on `op IS NOT NULL` to leave tombstones out. - The target table is an ordinary table: compaction of the topic doesn't remove rows from it. It keeps every change, while the compacted topic keeps the latest per key (the create, update, delete and tombstone rows stay in the target after compaction removes them from the topic). So the expansion gives you the full change history, and the compacted topic the current state. Debezium's event flattening transformation (`io.debezium.transforms.ExtractNewRecordState`) is the other way to get flat rows: it replaces the envelope with the `after` row before the event reaches the topic. Its `delete.tombstone.handling.mode` decides what happens to deletes (for example `rewrite` adds a `__deleted` field). Then the expansion schema is the row's columns, without `before` and `after`. ### Clean up Remove the connector, then drop the blob expansion, if you created one, before you delete the topic. A topic that still has a blob expansion can't be deleted: the delete fails with `Unknown broker error`. Drop the expansion and its target table with the SDK, as in [Blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion#remove-the-expansion). Then: ``` curl -s -X DELETE http://:8083/connectors/docs-kafka-connect-cdc kafka-topics.sh --bootstrap-server "$KAFKA_BOOTSTRAP" --delete --topic docs-kafka-connect-cdc.public.customers psql -c "SELECT pg_drop_replication_slot('docs_kafka_connect_cdc');" -c "DROP PUBLICATION IF EXISTS docs_kafka_connect_cdc;" ``` Delete the internal topics and the heartbeat topic when you retire the Connect cluster. ## Alternatives on VAST - **Instead of a sink connector into a database or data lake,** query the topic table directly with the VAST DataBase SDK, the Query Engine, Trino or Spark. The events are already stored as a VAST DataBase table. See [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics). - **Instead of a parsing job** that reads JSON events and writes columns, use [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion). It writes typed rows in the same transaction as each event. ## See also - [Integrations](/develop/docs/dev-event-broker-integrations-index): the checklist for any Kafka tool, and the status of each. - [Kafka compatibility](/develop/docs/dev-event-broker-reference-kafka-compatibility) and [topic configuration](/develop/docs/dev-event-broker-reference-topic-configuration) - [Security](/develop/docs/dev-event-broker-operations-security): authentication modes and identity-policy actions. - [Blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion) and [Query topics as tables](/develop/docs/dev-event-broker-guides-querying-topics) - Upstream: [Kafka Connect user guide](https://kafka.apache.org/43/kafka-connect/user-guide/), [Kafka Connect configs](https://kafka.apache.org/43/configuration/kafka-connect-configs/), [Debezium PostgreSQL connector](https://debezium.io/documentation/reference/stable/connectors/postgresql.html)