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

Worker configuration

A minimal connect-distributed.properties for an unauthenticated, unencrypted Kafka view:

bootstrap.servers=<vip1>:9092,<vip2>:9092,<vip3>: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.

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). 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="<vast-user>" password="<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="<vast-user>" password="<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="<vast-user>" password="<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="<vast-user>" password="<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="<vast-user>" password="<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.

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.

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.

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.

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://<connect-host>:8083/connectors/docs-kafka-connect-cdc/config -d '{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "plugin.name": "pgoutput",
  "database.hostname": "<postgres-host>",
  "database.port": "5432",
  "database.user": "<replication-user>",
  "database.password": "<password>",
  "database.dbname": "<database>",
  "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:

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:

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.

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."<bucket>/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 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 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. Then:

curl -s -X DELETE http://<connect-host>: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.
• Instead of a parsing job that reads JSON events and writes columns, use blob expansion. It writes typed rows in the same transaction as each event.

See also

• Integrations: the checklist for any Kafka tool, and the status of each.
• Kafka compatibility and topic configuration
• Security: authentication modes and identity-policy actions.
• Blob expansion and Query topics as tables
• Upstream: Kafka Connect user guide, Kafka Connect configs, Debezium PostgreSQL connector