GIP Debezium — MySQL and Mongo bootstrap

Scope

This page documents GIP-specific Debezium / Kafka Connect flows implemented in apps/debezium-connector/ in the zwing-connect monorepo:

• GIP MySQL — one bootstrap MySQL connector built from DIP outbound events + vendor database metadata, deployed via PUT to Kafka Connect.
• GIP Mongo — one MongoDB source connector per requested database, deployed via POST to Kafka Connect.

Other tasks in the same app (for example CreateDebeziumConnectorTask and the OutboundConnector debezium-connectors/outbound path) build non-GIP “ZWING” bulk outbound connectors with different topic routing (zwing.connect.outbound.$2). Those are out of scope here except where noted for name comparison.

Shared MySQL connector config for GIP is centralized in @libs/dip-v2: GipMysqlDebeziumConfigService (libs/dip-v2/src/services/gip-mysql-debezium-config.ts).

HTTP entry points and auth

Base path: controllers are registered on the app; the run commands live in apps/debezium-connector/src/http/controllers/run-command.ts.

Authentication: ApiGuard requires header x-api-secret to equal services.apiSecret from config. Missing or wrong secret returns a GenericException.

Kafka Connect base URL: services.debezium.host (used to build new URL('connectors', dbzHost) and similar). GIP MySQL bootstrap uses PUT .../connectors/{name}/config inside GipConnectorTask’s putConnectorConfig; GIP Mongo uses DebeziumService.create which POSTs to {dbzHost}/connectors (trailing path depends on dbzHost value — see implementation).

GIP MySQL: createGipMysqlConnector

File: apps/debezium-connector/src/services/gip.ts.

1. Discover outbound events

dipV2Service.eventsRepo.findMappedOutboundEvents() returns outbound events that:

• Have a non-empty listenerConfig.kafkaTopicName, and
• Appear in at least one connector–event mapping ($lookup on connector_event_mappings).

Convention: for MySQL GIP, kafkaTopicName is treated as a MySQL table name in the vendor’s database (not the literal Kafka topic name the outbound listener might use in other layers).

2. Resolve vendors (master MySQL)

GipConnectorTask.fetchVendorsByEnterpriseIds loads vendor rows: db_name, client_id, db_config_id keyed by sso_enterprise_id. Enterprises with events but no vendor are skipped (logged as errors in gip.ts).

3. Build include lists

• For each qualifying event, resolve vendor.dbName and add {dbName}.{tableName} to the table set where tableName === kafkaTopicName (after trimming / Mongo check below).
• isMongoDBCollection: if kafkaTopicName (case-insensitive) is customers, that “table” is excluded from the MySQL GIP connector (CRM Mongo is handled separately; see GIP Mongo).

If no table entries remain after resolution, the API throws: no MySQL GIP connector can be created.

4. Anchor db_config and createBootstrapConnector

The code picks an anchor vendor (first resolvable) and loads credentials via fetchDbConfig from the db_config table. One bootstrap connector is created that includes all resolved databases and tables in the merged lists. That implies a shared host/credentials model: every included tenant’s DB is expected to be reachable on that same MySQL endpoint (as stored for the anchor), which is a deployment constraint operators must validate.

GipConnectorTask.createBootstrapConnector (gip-connector-task.ts):

• Generates a unique connector name gip-bootstrap-{shortUlid}.
• Random MySQL database.server.id, a databaseServerName, and a unique schema.history.internal.kafka.topic (schema history topic per bootstrap).
• Merges GipMysqlDebeziumConfigService.buildBaseConfig with:
  • database.include.list (comma-separated DB names)
  • table.include.list (comma-separated db.table pairs)
  • topic.prefix = app.env
• Sends configuration with PUT to Kafka Connect: connectors/{connectorName}/config.

5. Base Debezium MySQL config (GipMysqlDebeziumConfigService)

File: libs/dip-v2/src/services/gip-mysql-debezium-config.ts

Notable keys in buildBaseConfig (excerpted conceptually; full map is in the source file):

• connector.class: io.debezium.connector.mysql.MySqlConnector
• snapshot.mode: schema_only — no full data snapshot, schema capture only
• include.schema.changes: false
• transforms: routeToDbTopic,filter with a Groovy Filter SMT and RegexRouter routeToDbTopic — topic rewrite zw.out.$2 (DB segment dropped, table name in position $2 of the regex)
• Schema history: schema.history.internal.kafka.bootstrap.servers from KAFKA_BROKERS, topic from identity or default

Topic naming difference: the older OutboundConnector payload in debezium-connectors/outbound.ts uses transforms.routeToDbTopic.replacement zwing.connect.outbound.$2. GIP bootstrap uses zw.out.$2. When debugging Kafka topics, use the connector that actually ran (GIP bootstrap name prefix gip-bootstrap- vs ZWING_CONNECT_OUTBOUND_CONNECTOR_).

6. Failure cases

createGipMysqlConnector throws (400/500) when:

• No mapped outbound events with kafkaTopicName exist
• No vendor can be resolved for any enterprise
• No MySQL tables can be built after filtering
• services.debezium.host is missing (in GipConnectorTask before PUT)

GIP Mongo: createGipMongoConnectors

File: apps/debezium-connector/src/services/gip.ts

• Requires mongo.crm.uri (or env alias documented in your deployment; code reads mongo.crm.uri and mentions CRM_MONGO_URI in the error message if unset).
• For each database name in the command suffix:
  • Connector name: gip-mongo-{slug}-{ulid} (slug from sanitized dbName).
  • connector.class com.mongodb.kafka.connect.MongoSourceConnector
  • buildConfig: connection.uri, database, collection: customers, topic.prefix: crm.out, output.format.value/key: json, change stream / full document settings, and a RegexRouter that normalizes to crm.out.$1 for topics.

Creation uses DebeziumService.create (HTTP POST .../connectors with body { name, config }).

Incremental MySQL GIP table patch (ACS, different connector name)

The debezium-connector app is not the only place GIP MySQL is updated. The App Connector Service task DebeziumConnectorPatchTask (apps/app-connector-service/src/tasks/debezium-connector-patch.ts):

• Targets a fixed connector name GIP-connector-v001
• Uses the same GipMysqlDebeziumConfigService mergeTableIntoConfig and needsRefresh to append a new db.table when a new mapped table is added for a tenant
• Skips the customers “table” for the same reason as the bootstrap (Mongo)

Operators should be aware: bootstrap may create gip-bootstrap-*, while runtime patches from ACS may update GIP-connector-v001. Environments can differ; confirm in Kafka Connect which connector is authoritative.

Environment and config checklist

Flow (mermaid)

flowchart LR
  subgraph dip [DipV2]
    Ev[outbound event kafkaTopicName]
    Map[connector event mapping]
  end
  subgraph gip [GipService in debezium-connector]
    V[Vendor and db_config from MySQL]
    T[Build database and table include lists]
    C[GipConnectorTask createBootstrapConnector]
  end
  subgraph kc [Kafka Connect]
    MyC[Debezium MySQL gip-bootstrap]
    MoC[Mongo source gip-mongo]
  end
  Ev --> T
  Map --> T
  V --> T
  T --> C
  C -->|PUT config| MyC
  Mgo[createGipMongoConnectors] -->|POST| MoC
  MyC --> KTopics[zw.out topics]
  MoC --> Crm[crm.out topics]

Related

• Data model reference — listenerConfig, events
• Kafka — environments and topic naming
• GIP blackout window — separate concern from CDC provisioning