Creating a Debezium Connector

Step-by-step guide to adding a new Debezium MySQL connector in the zwing-connect service, covering class structure, module registration, transforms, and deployment.

Connectors live in apps/debezium-connector/src/debezium-connectors/. Each connector is a NestJS injectable class that extends BaseConnectorClass and is decorated with @DebeziumConnector.

How It Works

@DebeziumConnector('connector-id') marks the class as injectable and stores a metadata key on it.
At startup, DebeziumConnectorExplorer scans all providers for that metadata and registers each instance in DebeziumMetadata.
DebeziumService / ConnectorManagerService read from DebeziumMetadata to build, list, validate, and deploy connectors via the Kafka Connect REST API.

Step 1 — Create the connector file

Create apps/debezium-connector/src/debezium-connectors/<name>.ts.

import { InjectKnexConnection } from "@ginesys-one/data-schemas";
import { BaseConnectorClass } from "@libs/debezium/connector-class";
import {
  ConnectorGroup,
  EventTable,
  TenantDBList,
} from "@libs/debezium/interface";
import { DebeziumConnector } from "@libs/debezium/decorators";
import { ConfigService } from "@nestjs/config";
import type { Knex } from "knex";
import { ALLOWED_HOSTS } from "../const";

// Each connector type must use a unique SERVER_ID_OFFSET to avoid MySQL replication ID conflicts.
// Existing offsets: outbound = 1_000_000 | item-master = 2_000_000 | offline = 3_000_000
const SERVER_ID_OFFSET = 4_000_000;

@DebeziumConnector("my-connector") // must be globally unique
export class MyConnector extends BaseConnectorClass {
  readonly allowedHost = ALLOWED_HOSTS;
  // indexOffset disambiguates connector names when multiple host-groups exist.
  // Pick a value that does not collide with other connectors' ranges.
  readonly indexOffset = 30;

  constructor(
    @InjectKnexConnection("master") protected readonly knex: Knex,
    private readonly configService: ConfigService,
  ) {
    super(knex);
  }

  async payload(
    connectorName: string,
    group: ConnectorGroup,
    databases: string[],
    tables: string[],
  ): Promise<Record<string, any>> {
    return {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "database.hostname": group.config.host,
      "database.port": group.config.port,
      "database.user": group.config.username,
      "database.password": group.config.password,
      "database.server.id": (
        SERVER_ID_OFFSET + (group.config.id ?? 0)
      ).toString(),
      "database.server.name": group.config.name,
      "database.include.list": databases.join(","),
      "table.include.list": tables.join(","),
      "database.allowPublicKeyRetrieval": "true",

      "schema.history.internal.kafka.bootstrap.servers":
        this.configService.get("KAFKA_BROKERS"),
      "schema.history.internal.kafka.topic": `schemahistory_${connectorName}`,
      "schema.history.internal.skip.unparseable.ddl": "true",
      "schema.history.internal.store.only.captured.tables.ddl": "true",

      "snapshot.mode": "schema_only",
      "snapshot.locking.mode": "none",
      "include.schema.changes": "false",

      "errors.log.enable": "true",
      "errors.log.include.messages": "true",
      "errors.retry.timeout": "300000",
      "errors.retry.delay.max.ms": "60000",
      "errors.tolerance": "all",

      transforms: "routeToDbTopic",
      "transforms.routeToDbTopic.type":
        "org.apache.kafka.connect.transforms.RegexRouter",
      "transforms.routeToDbTopic.regex": "([^.]+)\\.([^.]+)\\..*",
      "transforms.routeToDbTopic.replacement": "zwing.connect.my-connector.$2",

      "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",

      "topic.prefix": this.configService.get("app.env"),
      "decimal.handling.mode": "double",

      "schema.history.internal.kafka.recovery.poll.interval.ms": "30000",
      "schema.history.internal.kafka.recovery.attempts": "10",
    };
  }

  name(group: ConnectorGroup, index: number): string {
    return `ZWING_CONNECT_MY_CONNECTOR_${index}`;
  }

  async databases(group: ConnectorGroup): Promise<string[]> {
    return group.databases;
  }

  async tables(group: ConnectorGroup): Promise<string[]> {
    return group.databases.flatMap((db) =>
      this.getEventTableList().map((t) => `${db}.${t.tblName}`),
    );
  }

  protected getEventTableList(): EventTable[] {
    return [{ tblName: "my_table", key: ["id"] }];
  }

  async tenants(): Promise<TenantDBList> {
    return this.getAllTenants();
  }
}

Step 2 — Register in the module

Add the class to providers in apps/debezium-connector/src/module.ts:

import { MyConnector } from './debezium-connectors/my-connector';

@Module({
  ...
  providers: [
    ...,
    MyConnector,  // add here
  ],
})
export class DebeziumConnectorModule {}

That's all that's needed — DebeziumConnectorExplorer auto-discovers it at startup.

Abstract methods reference

Method	What to return
`payload(...)`	Full Kafka Connect connector config object
`name(group, index)`	Unique connector name string (e.g. `ZWING_CONNECT_X_${index}`)
`databases(group)`	List of database names for this group
`tables(group)`	Fully qualified table names: `db.table_name`
`getEventTableList()`	`EventTable[]` — the tables this connector captures
`tenants()`	Call `this.getAllTenants()` unless you need a custom query

Class properties

Property	Purpose
`allowedHost`	Only host-groups whose host is in this list are captured. Use `ALLOWED_HOSTS` from `../const`.
`indexOffset`	Added to the loop index to produce the connector name suffix. Must not overlap with other connectors' ranges.

Adding a Groovy filter

If events need filtering before hitting Kafka, add a filter() method and wire it into transforms:

// in payload():
transforms: 'routeToDbTopic,filter',
'transforms.filter.type': 'io.debezium.transforms.Filter',
'transforms.filter.language': 'jsr223.groovy',
'transforms.filter.condition': await this.filter(),

// method:
async filter(): Promise<string> {
  return `try {
    def op = value.get('op');
    def tableName = value.get('source').get('table');
    // return true to forward the event, false to drop it
    return op == 'c' || op == 'u';
  } catch (Exception e) { return false; }`;
}

The filter script receives the Debezium envelope. Use value.get('op') for the operation (c/u/d/r) and value.get('after') for the new row state.

Unwrap transform

Add unwrap to transforms if consumers expect a flat row rather than the full Debezium envelope:

transforms: 'routeToDbTopic,unwrap,filter',
'transforms.unwrap.type': 'io.debezium.transforms.ExtractNewRecordState',
'transforms.unwrap.drop.tombstones': 'true',
'transforms.unwrap.delete.handling.mode': 'rewrite',

See item-master.ts for a working example with all three transforms chained.

Unique values checklist

When adding a new connector, make sure these are unique across all connectors:

@DebeziumConnector('...') — the connector ID string
SERVER_ID_OFFSET — MySQL replication server ID base (existing: 1M outbound, 2M item-master, 3M offline)
indexOffset — the name index offset (existing: 0 offline, 15 item-master, 19 outbound)
schema.history.internal.kafka.topic — the schema history Kafka topic
name() return value pattern — the connector name prefix

Deployment

Connectors are deployed via the ConnectorManagerService, which calls DebeziumService.deployIfChanged. It compares the live config against the built payload on these keys:

table.include.list
database.include.list
transforms.filter.condition
database.hostname
database.port

A deploy is skipped if none of those changed. Use update() to force-push regardless.

The service always uses PUT /connectors/:name/config (upsert) rather than POST, so a redeploy never drops and recreates the connector — preserving CDC continuity.

Creating a Debezium Connector

On this page