Configuration Reference for Debezium MySQL Source Connector for Confluent Platform ¶

<fully-qualified_tableName>:_<keyColumn>_,<keyColumn>
To base a table key on multiple column names, insert commas between the column names.
Each fully-qualified table name is a regular expression in the following format:
<databaseName>.<tableName>
The property can include entries for multiple tables. Use a semicolon to
separate table entries in the list. The following example sets the message key
for the tables inventory.customers and purchase.orders:
inventory.customers:pk1,pk2;(.*).purchase.orders:pk3,pk4
For the table inventory.customer, the columns pk1 and pk2 are specified as the
message key. For the purchase.orders tables in any database, the columns
pk3 and pk4 server as the message key.
There is no limit to the number of columns that you use to create custom message
keys. However, it’s best to use the minimum number that are required to specify
a unique key.
Type: list
Default: No default
binary.handling.mode
Specifies how binary columns (for example, blob, binary, varbinary) should be
represented in change events.
Type: bytes or string
Importance: low
Default: Bytes
schema.name.adjustment.mode
Specifies how schema names should be adjusted for compatibility with the
message converter used by the connector.
Type: string
Importance: low
Valid values:

avro: Replaces the characters that can’t be used in Avro type name with _.
none: Doesn’t apply any adjustment.
avro_unicode: Replaces the underscore or characters that cannot be
used in the Avro type name with corresponding unicode like _uyyyy. Note
that _ is an escape sequence like a backlash in Java.
Default: avro
field.name.adjustment.mode
Specifies how field names should be adjusted for compatibility with the
message converter used by the connector.
Type: string
Importance: low
Valid values:

none: Doesn’t apply any adjustment.
avro: Replaces the characters that can’t be used in Avro type name with _.
avro_include: Replaces the underscore or characters that cannot be used in the Avro type name with corresponding unicode like _uxxxx.
Default: none
Advanced Parameters¶
connect.keep.alive




    
A Boolean value that specifies whether a separate thread should be used to
ensure that the connection to the MySQL server/cluster is kept alive.
Type: boolean
Default: true
converters
Enumerates a comma-separated list of symbolic names of the custom converter
instance that the connector can use. For each converter that you configure for
a connector, you must also add .type property, which specifies the
fully-qualified name of the class that implements the converter interface. For
an example and more details, see the Debezium documentation.
Type: list
Default: No default
table.ignore.builtin
A Boolean value that specifies whether built-in system tables should be
ignored. This applies regardless of the table include and exclude lists. By
default, system tables are excluded from having their changes captured, and no
events are generated when changes are made to any system tables.
Type: boolean
Default: true
database.ssl.mode
Specifies whether to use an encrypted connection.
Type: string
Valid values:

disabled: Specifies the use of an unencrypted connection.
preferred: Establishes an encrypted connection if the server supports secure connections. If the server does not support secure connections, falls back to an unencrypted connection.
required: Establishes an encrypted connection or fails if one can’t be made for any reason.
verify_ca: Behaves like required but additionally it verifies the server TLS certificate against the configured Certificate Authority (CA) certificates and fails if the server TLS certificate does not match any valid CA certificates.
verify_identity: Behaves like verify_ca but additionally verifies the server certificate matches the host of the remote connection.
Default: disabled
binlog.buffer.size
The size of a look-ahead buffer used by the binlog reader. The default setting
is 0,  which disables buffering.
Type: int
Default: 0
snapshot.mode
Specifies the criteria for running a snapshot when the connector starts.
Type: string
Valid values:

initial: Runs a snapshot only when no offsets have been recorded for the logical server name
initial_only: Runs a snapshot only when no offsets have been recorded for the logical server name and then stop
when_needed: Runs a snapshot only upon startup whenever the connector deems it necessary. Note that if the connector cannot find the binlog file mentioned in the offsets, it will take another snapshot and that may lead to duplicate data.
never: Never uses snapshots
schema_only: Runs a snapshot of schemas and not the data
schema_only_recovery: For a connector that has already been capturing change, enables recovery of a corrupted or lost database history topic that has been growing unexpectedly
Default: initial
snapshot.locking.mode
Controls whether and for how long the connector holds the global MySQL read
lock, which prevents any updates to the database while the connector is
performing a snapshot.
Type: string
Valid values:

minimal: Hold the global read lock for only the initial portion of the snapshot during which the connector reads the database schemas and other metadata
minimal_percona: Hold the global backup lock https://www.percona.com/doc/percona-server/5.7/management/backup_locks.html for only the initial portion of the snapshot during which the connector reads the database schemas and other metadata.
extended: Block all writes for the duration of the snapshot. Use if there are clients that are submitting operations that MySQL excludes from REPEATABLE READ semantics
none: Prevent the connector from acquiring any table locks during a snapshot. Best to use if and only if no schema changes are happening while the snapshot is running
Default: minimal
snapshot.include.collection.list
An optional, comma-separated list of regular expression that match the
fully-qualified names (<databaseName>.<tableName>) of the tables to
include in a snapshot. The specified items must be named in the connector’s
table.include.list property.
Type: list
Default: All tables specified in table.include.list
snapshot.select.statement.overrides
Specifies the table rows to include in a snapshot.  Use this property if you
want a snapshot to include only a subset of rows in a table. This property
affects snapshots only. It doesn’t apply to events that the connector reads
from the log. This property contains a comma-separated list of fully-qualified
table names in the form <databaseName>.<tableName>. For an example and
more details, see the Debezium
documentation.
Type: list
Default: No default
min.row.count.to.stream.results
Specifies the minimum number of rows a table must contain before the connector
streams results. To skip all table-size checks and always stream all results
during a snapshot, set this property to 0.
Type: int
Default: 1000
heartbeat.interval.ms
Controls how frequently the connector sends heartbeat messages to a Kafka
topic. The default behavior is the connector does not send heartbeat messages.
Type: int
Default: 0
heartbeat.action.query
Specifies a query that the connector executes on the source database when the
the connector sends a heartbeat message.
Type: string
Default: No default
database.initial.statements
A semicolon-separated list of SQL statements to be executed when a JDBC
connection, not the connection that is reading the transaction log, to the
database is established. To specify a semicolon as a character in a SQL
statement and not as a delimiter, use two semicolons, ;;.
Type: list
Default: No default
snapshot.delay.ms
An interval in milliseconds that the connector should wait before performing a
snapshot when the connector starts. If starting several connectors in a
cluster, this property is useful for avoiding snapshot interruptions, which
might cause re-balancing of connectors.
Type: int
Default: No default
snapshot.fetch.size
During a snapshot, the connector reads table content in batches of rows. This
property specifies the maximum number of rows in a batch.
Type: int
Default: No default
snapshot.lock.timeout.ms
Specifies the maximum amount of time (in milliseconds) to wait to obtain table
locks when performing a snapshot. If the connector does not obtain table locks
in the specified time interval, the snapshot fails. For more details, see how
MySQL connectors perform database snapshots.
Type: int
Default: 10000
enable.time.adjuster
Specify whether or not the connector converts a 2-digit year specification to
four digits. Set this to false if you want the conversion to be fully
delegated to the database.
With MySQL, you can insert year values with either 2 or 4 digits. For
2-digit values, the value gets mapped to a year in the range 1970 - 2069. The
default behavior is that the connector does the conversion.
Type: boolean
Default: true
source.struct.version
Schema version for the source block in Debezium events. By setting this
parameter to v1, the structures used in earlier versions can be produced. This setting is not recommended.
Type: string
Default: v2
skipped.operations
A comma-separated list of operation types that will be skipped during
streaming.  Possible values include: c for inserts/create, u for
updates, d for deletes. By default, no operations are skipped.
Type: list
Default: No default
signal.data.collection
Fully-qualified name of the data collection that is used to send signals to
the connector. Use the following format to specify the collection name:
<databaseName>.<tableName>.
Type: string
Default: No default
signal.enabled.channels
A list of the signaling channel names that are enabled for the connector. By
default, the following channels are available: source, kafka,
file, and jmx (optionally, you can also implement a custom signal
channel).
Type: list of strings
Default: source
notification.enabled.channels
A list of the notification channel names that are enabled for the connector.
By default, the following channels are available: sink, log, and
jmx (optionally, you can also implement a custom signal channel).
Type: string
Default: No default
incremental.snapshot.allow.schema.changes
Set to true to allow schema changes during an incremental snapshot.
Type: string
Default: false
incremental.snapshot.chunk.size
The maximum number of rows the connector fetches and reads into memory during
an incremental snapshot chunk. Note that increasing the chunk size provides
greater efficiency as the snapshot runs fewer snapshot queries of a greater
size. On the other hand, a larger chunk size requires more memory to buffer the
snapshot data. You should adjust the chunk size to a value that achieves
optimal performance in your environment.
Type: int
Default: 1024
read.only
Specifies whether to switch to alternative incremental snapshot watermarks
implementation to avoid writes to signal data collection.
Type: boolean
Default: false
provide.transaction.metadata
When set to true, the connector generates events with transaction
boundaries and enriches change event envelopes with transaction metadata. For
more details, see Transaction metadata.
Type: boolean
Default: false
event.processing.failure.handling.mode
Specifies how failures should be handled during the processing of events–that is,
when encountering a corrupted event.
Type: string
Valid values:

fail: Raises an exception indicating the corrupted event and its
position, causing the connector to be stopped.
warn: Does not raise an exception, instead the corrupted event and its
position will be logged and the event will be skipped.
ignore: Ignores the corrupted event completely with no logging.
Default: fail
topic.naming.strategy
The name of the TopicNamingStrategy class that should be used to determine the
topic name for data change, schema change, transaction, heartbeat event, etc.
Type: string
Default: io.debezium.schema.DefaultTopicNamingStrategy
topic.delimiter
Specifies the delimiter for topic name.  Defaults to ..
topic.cache.size
The size used for holding the topic names in bounded concurrent hash map. This
cache will help to determine the topic name corresponding to a given data
collection.
Type: int
Default: 10000
topic.heartbeat.prefix
Controls the name of the topic to which the connector sends heartbeat
messages. The topic name has the following pattern:
topic.heartbeat.prefix.topic.prefix.
Type: string
Default: __debezium-heartbeart
topic.transaction
Controls the name of the topic to which the connector sends transaction
metadata messages. The topic name has the following pattern:
topic.prefix.topic.transaction.
Type: string
Default: transaction
snapshot.max.threads
Specifies the number of threads the connector uses when performing an initial
snapshot. You can enable parallel initial snapshots—the connector will
process multiple tables at the same time—by setting this property to a value
greater than 1.
Type: int
Default: 1
snapshot.tables.order.by.row.count
Specifies the order in which the connector processes tables when it performs an
initial snapshot.
Type: int
Valid values:

descending: The connector snapshots tables in order, based on the number of rows from highest to lowest.
ascending: The connector snapshots tables in order, based on the number of rows from lowest to highest.
disabled: The connector disregards row count when performing an initial snapshot.
Default: disabled
errors.max.retries
The maximum number of retries on retriable errors (for example, connection
errors) before failing.
Type: int
Default: -1
Database history parameters¶
schema.history.internal.kafka.topic
The full name of the Kafka topic where the connector will store the database
schema history.
Important
Because Debezium uses multiple topics–of which certain limitations may
apply–for storing data, Confluent recommends you view Configuring Debezium
Topics
before you create a database schema history topic.
Type: string
Importance: high
Default: No default
schema.history.internal.kafka.bootstrap.servers
A list of host/port pairs that the connector will use for establishing an
initial connection to the Kafka cluster. This connection will be used for
retrieving database schema history previously stored by the connector, and for
writing each DDL statement read from the source database. This should point to
the same Kafka cluster used by the Kafka Connect process.
Type: list of strings
Importance: high
Default: No Default
schema.history.internal.kafka.recovery.poll.interval.ms
The maximum time duration, in milliseconds, the connector waits during startup or recovery while polling for persisted data.
Type: int
Default: 100
schema.history.internal.kafka.query.timeout.ms
The maximum time duration, in milliseconds, the connector waits for cluster information to be fetched using the Kafka admin client.
Type: int
Default: 3000
schema.history.internal.kafka.create.timeout.ms
An integer value that specifies the maximum number of milliseconds the
connector should wait while creating the Kafka history topic using the Kafka
admin client.
Type: int
Default: 30000
schema.history.internal.kafka.recovery.attempts
The maximum number of times that the connector should try to read persisted
history data before the connector recovery fails with an error. The maximum
amount of time to wait after receiving no data is recovery.attempts x
recovery.poll.interval.ms.
Type: int
Default: 100
schema.history.internal.skip.unparseable.ddl
A Boolean value that specifies whether the connector should ignore malformed
or unknown database statements or stop processing so a human can fix the
issue. The safe default is false. Skipping should be used only with care
as it can lead to data loss or mangling when the binlog is being processed.
Type: boolean
Default: false
schema.history.internal.store.only.captured.tables.ddl
A Boolean value that specifies whether the connector records schema structures
from all tables in a schema or database, or only from tables that are
designated for capture.
Type: boolean
Default: false
schema.history.internal.store.only.captured.databases.ddl
A Boolean value that specifies whether the connector should record schema
structures from all logical databases instance. When set to true the
connector records schema structures only for tables in the logical database
and schema from which Debezium captures change events. When set to false
the connector records schema structures for all logical databases.
Type: boolean
Default: false
Signal Parameters¶
signal.kafka.topic
The name of the Kafka topic that connector monitors for ad-hoc signals.
Type: string
Default: No default
signal.kafka.groupId
The name of the group ID that is used by Kafka consumers.
Type: string
Default: kafka-signal
signal.kafka.bootstrap.servers
A list of host/port pairs that the connector uses for establishing an initial
connection to the Kafka cluster. Each pair should point to the same Kafka
cluster used by the Kafka Connect process.
Type: list
Default: No default
signal.kafka.poll.timeout.ms
An integer value that specifies the maximum number of milliseconds the
connector should wait when polling signals. The default is 100 ms.
Type: int
Default: 100
Auto topic creation¶
For more information about Auto topic creation, see Configuring
Auto Topic Creation for Source
Connectors.
Configuration properties accept regular expressions (regex) that are defined as
Java regex.
topic.creation.groups
A list of group aliases that are used to define per-group topic configurations for matching topics. A default group always exists and matches all topics.
Type: List of String types
Default: empty
Possible Values: The values of this property refer to any additional groups. A default group is always defined for topic configurations.
topic.creation.$alias.replication.factor
The replication factor for new topics created by the connector. This value must not be larger than the number of brokers in the Kafka cluster. If this value is larger than the number of Kafka brokers, an error occurs when the connector attempts to create a topic. This is a required property for the default group. This property is optional for any other group defined in topic.creation.groups. Other groups use the Kafka broker default value.
Type: int
Default: n/a
Possible Values: >= 1 for a specific valid value or -1 to use the Kafka broker’s default value.
topic.creation.$alias.partitions
The number of topic partitions created by this connector. This is a required property for the default group. This property is optional for any other group defined in topic.creation.groups. Other groups use the Kafka broker default value.
Type: int
Default: n/a
Possible Values: >= 1 for a specific valid value or -1 to use the Kafka broker’s default value.
topic.creation.$alias.include
A list of strings that represent regular expressions that match topic names. This list is used to include topics with matching values, and apply this group’s specific configuration to the matching topics. $alias applies to any group defined in topic.creation.groups. This property does not apply to the default group.
Type: List of String types
Default: empty
Possible Values: Comma-separated list of exact topic names or regular expressions.
topic.creation.$alias.exclude
A list of strings representing regular expressions that match topic names. This list is used to exclude topics with matching values from getting the group’s specfic configuration. $alias applies to any group defined in topic.creation.groups. This property does not apply to the default group. Note that exclusion rules override any inclusion rules for topics.
Type: List of String types
Default: empty
Possible Values: Comma-separated list of exact topic names or regular expressions.
topic.creation.$alias.${kafkaTopicSpecificConfigName}
Any of the Changing Broker Configurations Dynamically for the version of the Kafka broker where the records will be written. The broker’s topic-level configuration value is used if the configuration is not specified for the rule. $alias applies to the default group as well as any group defined in topic.creation.groups.
Type: property values
Default: Kafka broker value
You can find more advanced configuration properties and details in the Debezium
connector for MySQL
documentation.
Portions of the information provided here derives from documentation
originally produced by the Debezium Community. Work
produced by Debezium is licensed under Creative Commons 3.0.
CSFLE configuration¶
csfle.enabled
Accepts a boolean value. CSFLE is enabled for the connector if csfle.enabled is set to True.
Type: boolean
Default: False
auto.register.schemas
Specifies if the Serializer should attempt to register the Schema with Schema Registry.
Type: boolean
Default: true
Importance: medium
use.latest.version
Only applies when auto.register.schemas is set to false. If auto.register.schemas is set to false and use.latest.version is set to true, then instead of deriving a schema for the object passed to the client for serialization, Schema Registry uses the latest version of the schema in the subject for serialization.
Type: boolean
Default: true
Importance: medium
        Sign up
      
By clicking "SIGN UP" you agree that your personal data will be processed in accordance with
        our Privacy Policy.