Skip to main content

mysql_cdc

Enables MySQL streaming for Expanso Edge.

# Common config fields, showing default values
input:
  label: ""
  mysql_cdc:
    flavor: "mysql"
    dsn: "" # No default (required)
    tables: [] # No default (required)
    checkpoint_cache: "" # No default (required)
    checkpoint_key: "mysql_binlog_position"
    snapshot_max_batch_size: 1000
    stream_snapshot: false # No default (required)
    max_parallel_snapshot_tables: 1
    auto_replay_nacks: true
    checkpoint_limit: 1024
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""

Metadata​

This input adds the following metadata fields to each message:

  • operation: The type of operation (insert, update, delete, or read for snapshot messages)
  • table: The name of the table
  • binlog_position: The binlog position (for CDC messages only, not set for snapshot messages)
  • schema: The table schema in Expanso Edge common schema format, compatible with processors like parquet_encode

Fields​

flavor​

The type of MySQL database to connect to.

Type: string
Default: "mysql"

OptionSummary
mariadbMariaDB flavored databases.
mysqlMySQL flavored databases.

dsn​

The DSN of the MySQL database to connect to.

Type: string

tables​

A list of tables to stream from the database.

Type: array of string

checkpoint_cache​

A cache resource to use for storing the current latest BinLog Position that has been successfully delivered, this allows Expanso Edge to continue from that BinLog Position upon restart, rather than consume the entire state of the table.

Type: string

checkpoint_key​

The key to use to store the snapshot position in checkpoint_cache. An alternative key can be provided if multiple CDC inputs share the same cache.

Type: string
Default: "mysql_binlog_position"

snapshot_max_batch_size​

The maximum number of rows to be streamed in a single batch when taking a snapshot.

Type: int
Default: 1000

max_reconnect_attempts​

The maximum number of attempts the MySQL driver will try to re-establish a broken connection before Connect attempts reconnection. A zero or negative number means infinite retry attempts.

Type: int
Default: 10

stream_snapshot​

If set to true, the connector will query all the existing data as a part of snapshot process. Otherwise, it will start from the current binlog position.

Type: bool

max_parallel_snapshot_tables​

Specifies the number of tables that will be snapshotted in parallel.

Type: int
Default: 1

auto_replay_nacks​

Whether messages that are rejected (nacked) at the output level should be automatically replayed indefinitely, eventually resulting in back pressure if the cause of the rejections is persistent. If set to false these messages will instead be deleted. Disabling auto replays can greatly improve memory efficiency of high throughput streams as the original shape of the data can be discarded immediately upon consumption and mutation.

Type: bool
Default: true

checkpoint_limit​

The maximum number of messages that can be processed at a given time. Increasing this limit enables parallel processing and batching at the output level. Any given BinLog Position will not be acknowledged unless all messages under that offset are delivered in order to preserve at least once delivery guarantees.

Type: int
Default: 1024

tls​

Using this field overrides the SSL/TLS settings in the environment and DSN.

Type: object

tls.skip_cert_verify​

Whether to skip server side certificate verification.

Type: bool
Default: false

tls.enable_renegotiation​

Whether to allow the remote server to repeatedly request renegotiation. Enable this option if you're seeing the error message local error: tls: no renegotiation.

Type: bool
Default: false

tls.root_cas​

An optional root certificate authority to use. This is a string, representing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

tls.root_cas_file​

An optional path of a root certificate authority file to use. This is a file, often with a .pem extension, containing a certificate chain from the parent trusted root certificate, to possible intermediate signing certificates, to the host certificate.

Type: string
Default: ""

tls.client_certs​

A list of client certificates to use. For each certificate either the fields cert and key, or cert_file and key_file should be specified, but not both.

Type: array of object
Default: []

tls.client_certs[].cert​

A plain text certificate to use.

Type: string
Default: ""

tls.client_certs[].key​

A plain text certificate key to use.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

tls.client_certs[].cert_file​

The path of a certificate to use.

Type: string
Default: ""

tls.client_certs[].key_file​

The path of a certificate key to use.

Type: string
Default: ""

tls.client_certs[].password​

A plain text password for when the private key is password encrypted in PKCS#1 or PKCS#8 format. The obsolete pbeWithMD5AndDES-CBC algorithm is not supported for the PKCS#8 format.

Because the obsolete pbeWithMD5AndDES-CBC algorithm does not authenticate the ciphertext, it is vulnerable to padding oracle attacks that can let an attacker recover the plaintext.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string
Default: ""

aws​

AWS IAM authentication configuration for MySQL instances. When enabled, IAM credentials are used to generate temporary authentication tokens instead of a static password.

Type: object

aws.enabled​

Enable AWS IAM authentication for MySQL. When enabled, an IAM authentication token is generated and used as the password. When using IAM authentication ensure max_reconnect_attempts is set to a low value to ensure it can refresh credentials.

Type: bool
Default: false

aws.region​

The AWS region where the MySQL instance is located. If no region is specified then the environment default will be used.

Type: string

aws.endpoint​

The MySQL endpoint hostname (e.g., mydb.abc123.us-east-1.rds.amazonaws.com).

Type: string

aws.id​

The ID of credentials to use.

Type: string

aws.secret​

The secret for the credentials being used.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

aws.token​

The token for the credentials being used, required when using short term credentials.

Secret

This field contains sensitive information. Use a secret reference rather than a literal value.

Type: string

aws.role​

Optional AWS IAM role ARN to assume for authentication. Alternatively, use roles array for role chaining instead.

Type: string

aws.role_external_id​

Optional external ID for the role assumption. Only used with the role field. Alternatively, use roles array for role chaining instead.

Type: string

aws.roles​

Optional array of AWS IAM roles to assume for authentication. Roles can be assumed in sequence, enabling chaining for purposes such as cross-account access. Each role can optionally specify an external ID.

Type: array of object

aws.roles[].role​

AWS IAM role ARN to assume.

Type: string
Default: ""

aws.roles[].role_external_id​

Optional external ID for the role assumption.

Type: string
Default: ""

batching​

Allows you to configure a batching policy.

Type: object

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.check​

A Bloblang query that should return a boolean value indicating whether a message should end a batch.

Type: string
Default: ""

batching.processors​

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array of processor