Kafka Connect and Debezium

In Kafka Connect, a connector is created from a configuration JSON sent to the Connect REST API.

In Pyahu CLI v1, you don’t need to keep that JSON as a file. You declare the connector in pyahu.yaml; the CLI generates or receives the JSON, writes it to a Kubernetes Secret, and creates a Job that applies the configuration to Kafka Connect.

What v1 supports

The CLI’s declarative surface covers two cases:

• Debezium source for PostgreSQL, with defaults generated by Pyahu.
• Custom source or sink, using free-form config in the Kafka Connect format.

Debezium/PostgreSQL source

For Postgres CDC, use type: source and kind: debezium.postgres:

services:
  kafkaConnect:
    enabled: true
    connectors:
      - name: app-cdc
        type: source
        kind: debezium.postgres
        database: app
        topicPrefix: app-cdc
        snapshotMode: initial
        tables:
          include:
            - public.orders
            - public.customers
        config:
          decimal.handling.mode: string
          tombstones.on.delete: "false"

Fields like slot, publication, topicPrefix, database, and snapshotMode have defaults. In practice, the minimal example can be:

services:
  kafkaConnect:
    enabled: true
    connectors:
      - name: app-cdc
        type: source
        tables:
          include:
            - public.orders

With this, pyahu up creates or updates the connector.

Generated JSON

For the app-cdc connector, the CLI generates a JSON similar to this:

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "tasks.max": "1",
  "database.hostname": "postgres.pyahu-local-dev.svc.cluster.local",
  "database.port": "5432",
  "database.user": "pyahu",
  "database.password": "pyahu_local",
  "database.dbname": "app",
  "topic.prefix": "app-cdc",
  "plugin.name": "pgoutput",
  "slot.name": "app_cdc_slot",
  "publication.name": "app_cdc_publication",
  "publication.autocreate.mode": "filtered",
  "snapshot.mode": "initial",
  "table.include.list": "public.orders,public.customers",
  "decimal.handling.mode": "string",
  "tombstones.on.delete": "false"
}

This is the body sent to:

PUT /connectors/app-cdc/config

Plugins

The default quay.io/debezium/connect image ships only Debezium. For any other connector — Redis Streams, JDBC, S3 — or for your own SMT, declare the artifacts under services.kafkaConnect.plugins. The CLI installs each one into a plugin directory before the worker starts; there is no image to build and publish.

services:
  kafkaConnect:
    enabled: true
    plugins:
      - name: redis-kafka-connect
        url: https://github.com/redis-field-engineering/redis-kafka-connect/releases/download/v1.1.0/redis-redis-kafka-connect-1.1.0.zip
        sha256: 7e4249ca356f702220cf09e9e150c8336e24824d7a72fce05d7999bf2a7e03df
      - name: redis-kafka-connect        # same name = same directory
        file: connect-plugins/my-outbox-router.jar

Field	Rule
name	DNS label; it is the plugin directory name. Repeating a name is allowed and meaningful.
url + sha256	Downloads and verifies the hash. sha256 is required with url: an unverified download is not accepted.
file	Path relative to the pyahu.yaml directory, ≤ 1 MiB, stored in a Secret. Meant for a small jar that has no public URL yet.

Exactly one of url or file per entry. The artifact must be a .zip, .tar.gz, .tgz, or .jar.

Why two artifacts can share a name

Kafka Connect isolates each plugin directory in its own classloader. A transform that needs the connector’s classes (to read the type of a record field, say) has to sit in the same directory as the connector — a loose jar in another directory does not work, and the failure shows up as a ClassNotFoundException only when the task runs. Repeating the name is how you say “these artifacts belong together”.

How installation works

An install-plugins initContainer (alpine:3.21) runs before the worker and, for each plugin: fetches or copies the artifact, verifies sha256, extracts .zip/.tar.gz, and flattens every *.jar it finds into /plugins/<name>/ — release archives nest jars under lib/, and Connect only treats the immediate children of a plugin.path entry as a plugin. Any failure fails the pod with a message that names the plugin.

The volume is mounted at /kafka/connect-extra on the worker, which gets CONNECT_PLUGIN_PATH=/kafka/connect,/kafka/connect-extra — the image’s own plugins still apply.

Change the plugin list and the pod is recreated: artifacts are installed at pod start, so an untouched pod would keep serving the previous set.

Checking what the worker loaded

curl -s http://localhost:8083/connector-plugins | jq -r '.[].class'
# transforms and converters only show up with connectorsOnly=false
curl -s 'http://localhost:8083/connector-plugins?connectorsOnly=false' | jq -r '.[].class'

pyahu describe kafka-connect      # declared (plugins) and loaded (pluginsLoaded)

The Job that registers a connector waits for that class to appear in /connector-plugins before issuing the PUT. That separates “the plugin was never installed” from “the connector started and failed”.

Custom sink

Sink connectors are also Kafka Connect JSON. The difference is that the sink plugin must exist in the image used by services.kafkaConnect.image.

Example with a JDBC sink in a custom image that already contains the plugin:

services:
  kafkaConnect:
    enabled: true
    image: ghcr.io/acme/connect-with-jdbc-sink
    version: 1.0.0
    connectors:
      - name: orders-jdbc-sink
        type: sink
        kind: custom
        config:
          connector.class: io.confluent.connect.jdbc.JdbcSinkConnector
          tasks.max: "1"
          topics: app-cdc.public.orders
          connection.url: jdbc:postgresql://warehouse.local:5432/orders
          connection.user: warehouse
          connection.password: warehouse_local
          insert.mode: upsert
          pk.mode: record_key
          auto.create: "true"

For custom connectors, Pyahu does not try to interpret the payload. The contents of config become the JSON sent to Kafka Connect:

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "tasks.max": "1",
  "topics": "app-cdc.public.orders",
  "connection.url": "jdbc:postgresql://warehouse.local:5432/orders",
  "connection.user": "warehouse",
  "connection.password": "warehouse_local",
  "insert.mode": "upsert",
  "pk.mode": "record_key",
  "auto.create": "true"
}

The same model works for a custom source:

services:
  kafkaConnect:
    enabled: true
    connectors:
      - name: external-source
        type: source
        kind: custom
        config:
          connector.class: com.example.SourceConnector
          tasks.max: "1"
          topic: external.events

Equivalent REST API

If you were creating it manually, the command would be:

curl -X PUT \
  -H 'Content-Type: application/json' \
  --data-binary @connector.json \
  http://localhost:8083/connectors/app-cdc/config

Pyahu does this inside the cluster, using the Connect internal URL:

http://kafka-connect.pyahu-local-dev.svc.cluster.local:8083

How the CLI applies it

During pyahu up, the CLI:

1. Brings up the Kafka Connect Deployment using the quay.io/debezium/connect image.
2. Creates the internal compacted config, offset, and status topics.
3. Renders or reads the connector JSON from services.kafkaConnect.connectors.
4. Writes that JSON into a Secret as connector.json.
5. Creates a Job that waits for Connect to respond and runs PUT /connectors/<name>/config.
6. Waits for every connector task to become RUNNING.

When you change the configuration, the Secret and the Job receive a name with a hash of the payload. This makes the apply idempotent and lets you reapply changes with another pyahu up.

Removing a managed connector

After a successful pyahu up, Pyahu stores the names of the connectors managed by this stack. On a later run, it deletes a registration that is in that inventory but no longer appears in pyahu.yaml.

Connectors created manually through the REST API or another tool are not added to the inventory and are left alone. The inventory remains in the cluster when Kafka Connect is disabled, so registrations restored from Kafka’s internal topics can still be reconciled if the service is enabled again.

Removing a connector registration does not delete the Kafka topics or other service data it used.

Inspecting the connector

curl http://localhost:8083/connectors
curl http://localhost:8083/connectors/app-cdc/status

You can also view it through the Kafka UI when it is enabled:

https://kafka-ui.localhost

v1 rules

• kind: debezium.postgres is always type: source.
• kind: custom can be type: source or type: sink.
• Custom connectors require config.connector.class.
• Plugins that do not ship with the image go in services.kafkaConnect.plugins; alternatively, use an image that already contains the plugin.