DatabaseChangeTracking Configuration

Defines database change tracking to monitor and capture data changes in real-time using Apache Camel with Debezium connectors. Changes are automatically published to NATS streams for event-driven integration patterns.

Schema properties#

YAML examples#

PostgreSQL change tracking#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: postgres-production
  description: Track changes on production database
spec:
  type: postgres
  parameters:
    databaseHostname: db.example.com
    databasePort: "5432"
    databaseDbname: northwind
    databaseUser: weikio_user
    databasePassword: ${db-password}
    schemaIncludeList: public
    tableIncludeList: public.customers

PostgreSQL with variable substitution#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: postgres-secure
spec:
  type: postgres
  parameters:
    databaseHostname: "{{dbHost}}"
    databasePort: "5432"
    databaseDbname: "{{dbName}}"
    databaseUser: "{{dbUser}}"
    databasePassword: "{{dbPassword}}"
    tableIncludeList: public.orders
  variables:
    - dbHost: "{{sys:DB_HOST}}"
    - dbName: "{{sys:DB_NAME}}"
    - dbUser: "{{sys:DB_USERNAME}}"
    - dbPassword: "{{sys:DB_PASSWORD}}"

MySQL change tracking#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: mysql-inventory
spec:
  type: mysql
  parameters:
    databaseHostname: mysql.example.com
    databasePort: "3306"
    databaseUser: weikio
    databasePassword: ${mysql-password}
    databaseServerName: mysql-prod
    databaseIncludeList: inventory
    tableIncludeList: inventory.products

SQL Server change tracking#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: sqlserver-orders
spec:
  type: sqlserver
  parameters:
    databaseHostname: sqlserver.example.com
    databasePort: "1433"
    databaseUser: sa
    databasePassword: ${sqlserver-password}
    databaseDbname: OrderManagement
    tableIncludeList: dbo.Orders

MongoDB change tracking#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: mongodb-events
spec:
  type: mongodb
  parameters:
    mongodbConnectionString: "{{connectionString}}"
    collectionIncludeList: mydb.events
  variables:
    - connectionString: "{{sys:MONGODB_URI}}"

Oracle change tracking#

apiVersion: weik.io/v1alpha1
kind: DatabaseChangeTracking
metadata:
  name: oracle-crm
spec:
  type: oracle
  parameters:
    databaseHostname: oracle.example.com
    databasePort: "1521"
    databaseUser: weikio
    databasePassword: ${oracle-password}
    databaseDbname: ORCL
    schemaIncludeList: MYSCHEMA
    tableIncludeList: MYSCHEMA.CUSTOMERS

Database types#

Weik.io uses Debezium connectors for change data capture. Each database type has specific parameters defined by Debezium.

PostgreSQL#

type: postgres
parameters:
  databaseHostname: localhost
  databasePort: "5432"
  databaseDbname: mydb
  databaseUser: user
  databasePassword: pass
  schemaIncludeList: public
  tableIncludeList: public.tablename

Common Parameters:

• databaseHostname: PostgreSQL server hostname
• databasePort: PostgreSQL port (default: 5432)
• databaseDbname: Database name
• databaseUser: Database user
• databasePassword: Database password
• schemaIncludeList: Schemas to monitor (comma-separated)
• tableIncludeList: Tables to monitor (comma-separated, format: schema.table)
• pluginName: Logical decoding plugin (automatically set to pgoutput)

Reference: Debezium PostgreSQL Connector

MySQL#

type: mysql
parameters:
  databaseHostname: localhost
  databasePort: "3306"
  databaseUser: user
  databasePassword: pass
  databaseServerName: mysql-server
  databaseIncludeList: mydb
  tableIncludeList: mydb.tablename

Common Parameters:

• databaseHostname: MySQL server hostname
• databasePort: MySQL port (default: 3306)
• databaseUser: Database user
• databasePassword: Database password
• databaseServerName: Logical name for the MySQL server
• databaseIncludeList: Databases to monitor (comma-separated)
• tableIncludeList: Tables to monitor (comma-separated, format: db.table)

Reference: Debezium MySQL Connector

SQL Server#

type: sqlserver
parameters:
  databaseHostname: localhost
  databasePort: "1433"
  databaseUser: user
  databasePassword: pass
  databaseDbname: mydb
  tableIncludeList: dbo.tablename

Common Parameters:

• databaseHostname: SQL Server hostname
• databasePort: SQL Server port (default: 1433)
• databaseUser: Database user
• databasePassword: Database password
• databaseDbname: Database name
• tableIncludeList: Tables to monitor (comma-separated, format: schema.table)

Reference: Debezium SQL Server Connector

MongoDB#

type: mongodb
parameters:
  mongodbConnectionString: mongodb://localhost:27017
  collectionIncludeList: mydb.mycollection

Common Parameters:

• mongodbConnectionString: MongoDB connection URI
• collectionIncludeList: Collections to monitor (comma-separated, format: db.collection)
• mongodbName: MongoDB replica set or cluster name

Reference: Debezium MongoDB Connector

Oracle#

type: oracle
parameters:
  databaseHostname: localhost
  databasePort: "1521"
  databaseUser: user
  databasePassword: pass
  databaseDbname: ORCL
  schemaIncludeList: MYSCHEMA
  tableIncludeList: MYSCHEMA.TABLENAME

Common Parameters:

• databaseHostname: Oracle server hostname
• databasePort: Oracle port (default: 1521)
• databaseUser: Database user
• databasePassword: Database password
• databaseDbname: Database SID or service name
• schemaIncludeList: Schemas to monitor (comma-separated)
• tableIncludeList: Tables to monitor (comma-separated, format: SCHEMA.TABLE)

Reference: Debezium Oracle Connector

Variables#

Variables provide substitution for connection parameters, credentials, and other values. Variables are not for column mapping.

Variable substitution#

Variables use double-brace syntax {{variableName}} in parameters and can reference system environment variables with the sys: prefix:

spec:
  parameters:
    databaseHostname: "{{dbHost}}"
    databasePassword: "{{dbPassword}}"
  variables:
    - dbHost: db.production.com
    - dbPassword: "{{sys:DB_PASSWORD}}"

System environment variables#

Access environment variables with {{sys:VARIABLE_NAME}}:

variables:
  - dbHost: "{{sys:DATABASE_HOST}}"
  - dbUser: "{{sys:DATABASE_USER}}"
  - dbPassword: "{{sys:DATABASE_PASSWORD}}"

Security with variables#

Store sensitive credentials as environment variables and reference them:

export DB_PASSWORD="secure_password"
export DB_HOST="secure-db.example.com"

spec:
  parameters:
    databaseHostname: "{{host}}"
    databasePassword: "{{password}}"
  variables:
    - host: "{{sys:DB_HOST}}"
    - password: "{{sys:DB_PASSWORD}}"

NATS integration#

DatabaseChangeTracking automatically creates NATS JetStream resources:

1. NATS Stream: Created as dbchangetracking-{name}-stream
2. NATS Subject: Changes published to dbchangetracking.{name}
3. Republish Subject: Republished to dbchangetrackingprocessing.{name}

Changes are automatically captured and published to NATS. No additional configuration required.

Stream configuration#

Automatically created streams have:

• Storage: File-based persistence
• Max messages: 512
• Retention: Limits-based (oldest messages discarded)
• Discard policy: Old

Consuming changes#

Subscribe to the NATS subject to consume change events:

apiVersion: weik.io/v1alpha1
kind: IntegrationFlow
metadata:
  name: process-db-changes
spec:
  Description: Process database changes from NATS
  # Subscribe to: dbchangetracking.{tracking-name}

Change event structure#

Change events contain:

• Operation type: INSERT, UPDATE, DELETE
• Before state (for UPDATE and DELETE)
• After state (for INSERT and UPDATE)
• Timestamp
• Transaction metadata
• Table/collection name
• Schema information

Events are serialized as JSON and published to NATS.

Usage notes#

Debezium connectors#

Weik.io uses Apache Camel with Debezium connectors for change data capture. All parameters in the parameters section are passed directly to the Debezium connector.

For complete parameter lists and advanced configuration options, refer to the Debezium documentation for your database type.

Connection parameters#

Parameters use Debezium’s camelCase naming convention:

• databaseHostname (not Host)
• databaseDbname (not Database)
• databaseUser (not Username)
• databasePassword (not Password)
• tableIncludeList (not Table)

Change detection methods#

Debezium uses database-specific methods:

• PostgreSQL: Logical replication with pgoutput plugin
• MySQL: Binary log (binlog) parsing
• SQL Server: Change Data Capture (CDC)
• MongoDB: Change streams
• Oracle: LogMiner or Xstream

Table and schema filtering#

Use include lists to specify which tables to monitor:

parameters:
  schemaIncludeList: public,customer_data
  tableIncludeList: public.orders,public.customers

Patterns follow Debezium’s filtering syntax. See connector documentation for advanced filtering options.

Monitoring multiple tables#

Create separate DatabaseChangeTracking configurations for each table or use Debezium’s include list patterns:

# Single configuration for multiple tables
parameters:
  tableIncludeList: public.orders,public.customers,public.products

Each configuration creates its own NATS stream and integration flow.

Best practices#

• Use variables for credentials and connection strings
• Store secrets as environment variables
• Use minimal database permissions for tracking user
• Test in development before production deployment
• Monitor NATS stream sizes and message rates
• Consider database performance impact
• Use specific table include lists rather than monitoring entire databases
• Enable SSL/TLS for database connections when supported
• Review Debezium documentation for database-specific requirements

Security considerations#

• Store database credentials using variables with environment variable references
• Use database users with read-only permissions when possible
• Enable SSL/TLS connections in Debezium parameters
• Audit access to tracked tables
• Monitor who subscribes to change event streams
• Rotate credentials regularly

Use cases#

• Real-time data synchronization
• Event-driven architectures
• Audit logging and compliance
• Cache invalidation
• Search index updates
• Data warehouse updates
• Microservices data coordination
• Real-time analytics pipelines

Related#

• Integration Flows: Define flows that consume change events
• Variables: Manage credentials and configuration values
• Using Variables: Variable substitution examples