Source Connector Workflow
Source Connector Workflow
This page describes the common configuration steps used by GIFROST source connectors. Database-specific pages list the exact parameters for PostgreSQL, Oracle, MySQL, SQL Server, Db2, Informix, and MongoDB.
Wizard Steps
Basic
The required field in this step is name. It must be unique in the selected Kafka Connect cluster and is used by GIFROST to derive the Debezium topic prefix automatically.
By convention, the generated topic.prefix is:
<connector-name>-server
Kafka topics created by the connector follow this structure:
<topic.prefix>.<schema>.<table>
For example, a connector named orders-pg that replicates public.customer writes to:
orders-pg-server.public.customer
tasks.max is kept at 1 for relational CDC source connectors. Reading transactional logs is ordered and effectively single-threaded, so increasing the task count does not increase log-reading parallelism and can create misleading expectations. The tag, config.action.reload, and task enforcement fields are operational GIFROST fields used for grouping and connector actions.
Converters
GIFROST uses Avro by default for key and value serialization. The Avro converter writes records with schemas registered in Schema Registry, which makes downstream consumers work with versioned schemas instead of unstructured JSON.
| Parameter | Default / Example | Description |
|---|---|---|
key.converter | io.confluent.connect.avro.AvroConverter | Serializes Kafka record keys as Avro. |
value.converter | io.confluent.connect.avro.AvroConverter | Serializes Kafka record values as Avro. |
key.converter.schema.registry.url | http://kafka-schema-registry:8081 | Schema Registry endpoint for key schemas. |
value.converter.schema.registry.url | http://kafka-schema-registry:8081 | Schema Registry endpoint for value schemas. |
value.converter.ignore.default.for.nullables | true | Preserves real NULL values for nullable fields instead of replacing them with Avro defaults. This is important for CDC because a database NULL must remain distinguishable from a default value. |
value.converter.enhanced.avro.schema.support | true | Keeps additional Avro schema information such as namespace and richer schema metadata. This improves compatibility for consumers that rely on precise schema structure. |
Database Connection
The Database Connection step uses entries stored in Database Store. GIFROST filters the list by connector type: a PostgreSQL connector shows PostgreSQL connections, an Oracle connector shows Oracle connections, and so on.
After selecting a connection, click Validation to confirm that the host, port, database name, username, and password can be used by the selected Kafka Connect cluster.
The Raw tab shows the concrete connector parameters generated from the selected Database Store entry, such as database.hostname, database.port, database.dbname, database.user, and password-related fields. Passwords should not be displayed in plain text.
Raw parameters can be edited, but this is not recommended. Other GIFROST features rely on the Database Store entry as the source of truth. If the connection data is wrong, create or update the Database Store credential first, and then use it in the connector wizard.
Database Settings
Database Settings contains connector-specific Debezium parameters that control the replication mechanism. For most standard pipelines, the defaults should be left unchanged.
For PostgreSQL, GIFROST uses these conventions:
| Parameter | Default / Convention | Meaning |
|---|---|---|
slot.name | debezium_<connector-name> | Logical replication slot used by Debezium to read WAL changes. |
publication.name | dbz_publication | PostgreSQL publication used by pgoutput. |
publication.autocreate.mode | all_tables | Debezium can create a publication for all tables available to the replication user. |
slot.drop.on.stop | true / enabled | Drops the replication slot on a graceful stop. This avoids unused WAL retention, but a restart cannot resume from the old slot position. |
Database Wizard
The Database Wizard defines the replication scope: schemas, tables, and columns. GIFROST intentionally starts from a safe state where nothing is selected for replication. Selecting All Schemas expands the scope to the entire visible database.
The wizard writes Debezium filters under the hood:
schema.include.list
schema.exclude.list
table.include.list
table.exclude.list
column.include.list
column.exclude.list
The filters are regular expressions. To replicate one specific table safely, select the schema first and then select the table. If only a table pattern is entered manually without narrowing the schema, the final regex can match more objects than expected.
Message Keys
message.key.columns is configured through Columns PK Mapping. It is critical for tables without a database primary key, because Debezium still needs a stable Kafka message key to represent updates and deletes correctly.
The Generate Message Key Columns action generates message.key.columns entries for selected tables that require a custom key.
Topic Creation
Topic creation defines how Kafka topics are created for replicated tables.
| Parameter | Default / Example | Description |
|---|---|---|
topic.creation.groups | default | List of topic creation groups. |
topic.creation.default.replication.factor | 1 | Kafka replication factor for created topics. |
topic.creation.default.partitions | 1 | Number of partitions. Keep 1 for relational CDC when event order matters. |
topic.creation.default.cleanup.policy | delete | Cleanup behavior. compact can be used when the topic should keep the latest value per key. |
topic.creation.default.compression.type | producer | Compression setting inherited from producers. |
topic.creation.default.delete.retention.ms | 604800000 | Seven days. This delayed cleanup window keeps delete markers/tombstones available before Kafka removes them. |
The default group applies to every replicated table unless a custom group matches the table. Custom groups can tune large and small tables differently, for example two days of retention for large tables and seven days for smaller tables.
Snapshot
Snapshot settings decide whether the connector loads existing data before streaming changes from transaction logs.
snapshot.mode | Meaning |
|---|---|
initial | Default. Runs an initial snapshot only when no previous offsets exist, then streams changes. |
always | Runs a snapshot every time the connector starts, then streams changes. |
initial_only | Runs the snapshot and stops before streaming changes. |
no_data | Skips data snapshot and starts streaming from available log position. |
never | Legacy or connector-specific equivalent of skipping snapshots; use only if supported by the selected connector version. |
when_needed | Runs a snapshot only when offsets are missing or no longer usable. |
configuration_based | Uses additional snapshot.mode.configuration.based.* properties to decide behavior. |
custom | Uses a custom snapshotter implementation named by snapshot.mode.custom.name. |
Use these formats for filtered snapshots:
snapshot.include.collection.list=<schema>.<table>,<schema>.<table>
snapshot.select.statement.overrides=<schema>.<table>
snapshot.select.statement.overrides.<schema>.<table>=SELECT * FROM <schema>.<table> WHERE <condition>
Signal, Notification, and Heartbeat
Signals let operators trigger Debezium actions after connector creation, for example incremental snapshots. A signal table is usually referenced as:
signal.data.collection=<schemaName>.<tableName>
Debezium notifications report the state of long-running operations, including snapshot progress. GIFROST uses a notification topic named:
<connector-name>-notifications
Incremental snapshots use watermarking to mark snapshot windows. The connector reads data in chunks while also streaming log changes; watermarks help separate chunk boundaries from concurrent CDC events.
Connector
The Connector step contains runtime and data-handling settings.
| Area | Parameters | Purpose |
|---|---|---|
| Data handling | decimal.handling.mode, interval.handling.mode, time.precision.mode, binary.handling.mode, JSON-related fields, unavailable.value.placeholder | Controls how source database values are represented in Kafka records. |
| Performance | poll.interval.ms, max.batch.size, max.queue.size, skipped.operations | Controls polling, batching, queueing, and operation types that may be skipped. |
| Failure handling | event.processing.failure.handling.mode, event.deserialization.failure.handling.mode | Controls how connector-specific event processing failures are handled. |
| Connector monitoring | internal.advanced.metrics.enable, custom.metric.tags | Enables extended GIFROST metrics and dashboard tags. |
GIFROST recommends enabling connector monitoring. The default metric tag is:
connectorName=<connector-name>
Additional tags can be added to enrich dashboards and metrics for create, update, delete, and filtered operations.
Exactly Once Support
exactly.once.support=requested asks Kafka Connect and the connector to use exactly-once semantics when supported by the environment. transaction.boundary=poll uses the poll cycle as the transaction boundary, and transaction.boundary.interval.ms can limit transaction duration.
For CDC pipelines, change ordering remains the most important operational property, so these settings should be changed only with a clear reason.
Events
The Events step configures event shape and post-processing.
| Parameter | Purpose |
|---|---|
converters | List of custom value converters. |
post.processors | Change event post-processors. |
topic.naming.strategy | Class that builds topic names, for example io.debezium.schema.SchemaTopicNamingStrategy. |
transaction.metadata.factory | Factory for transaction metadata structures. |
table.ignore.builtin | Ignores database system tables when supported. |
column.mask.hash.<N>.with.salt.<M> | Masks sensitive columns with salted hashes. |
column.mask.with.<N>.chars | Replaces column values with a fixed number of masking characters. |
column.truncate.to.<N>.chars | Truncates long values before they are emitted. |
include.schema.comments | Includes table and column comments in schemas. |
column.propagate.source.type | Propagates source column type metadata. |
Transforms
Transforms are Kafka Connect Single Message Transforms applied after Debezium creates a record and before it reaches Kafka.
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
predicates and predicates.config let operators apply transforms conditionally.
Error Handling
| Parameter | Purpose |
|---|---|
errors.retry.timeout | How long Kafka Connect retries failed operations. |
errors.retry.delay.max.ms | Maximum delay between retries. |
errors.tolerance | none stops on errors; all continues and should be used carefully. |
errors.log.enable | Enables error logging. |
errors.log.include.messages | Adds record payloads to logs; useful for debugging but may expose sensitive data. |
Summary
The final step shows the JSON that will be submitted to Kafka Connect. Before starting the connector, verify name, generated topic.prefix, selected connection, replication scope, snapshot mode, topic creation groups, transforms, and error handling.