Configuring Connection Details

This section provides instructions for configuring the connection details required to establish communication between the Micro-Integration and your third-party system.

For information about configuring the connection to the event broker, see Connecting to Your Event Broker.

The name of the binder for MySQL CDC is mysqlcdc.

MySQL CDC Database Prerequisites

Before you use the MySQL CDC, you must configure your MySQL CDC database with the following settings:

• Binary logging enabled

  MySQL CDC must be configured with binary logging enabled in ROW format. Add the following settings to your MySQL CDC configuration file (my.cnf or my.ini):

  server-id = 1
  log_bin = mysql-bin
  binlog_format = ROW
  binlog_row_image = FULL
  expire_logs_days = 10

• Required user privileges

  The MySQL CDC user specified in the connection configuration must have the following privileges:

  • REPLICATION SLAVE—Required to read the binary log

  • REPLICATION CLIENT—Required to use SHOW MASTER STATUS, SHOW SLAVE STATUS, and SHOW BINARY LOGS

  • SELECT—Required on all tables being monitored

  To grant these privileges, run the following SQL commands:

  CREATE USER 'mysqluser'@'%' IDENTIFIED BY 'password';
  GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'mysqluser'@'%';
  GRANT SELECT ON <database_name>.* TO 'mysqluser'@'%';
  FLUSH PRIVILEGES;

• GTID mode (recommended)

  We strongly recommend using Global Transaction Identifiers (GTID) mode for production deployments because it provides better consistency and failover support. To enable GTID mode, add the following settings to your MySQL CDC configuration:

  gtid_mode = ON
  enforce_gtid_consistency = ON

  For more information about MySQL CDC binary logging and GTID configuration, consult your MySQL CDC server administrator or refer to the MySQL CDC documentation for your specific MySQL CDC version.

MySQL CDC Connection Details

To manually configure the MySQL CDC connection details, set the following properties in application.yml:

spring:
  cloud:
    stream:
      mysqlcdc:
        bindings:
          input-0:
            consumer:
              endpoint:
                query-parameters:
                  databaseHostname: <hostname>
                  databasePort: <port>
                  databaseUser: <database user name>
                  databaseServerId: <unique server ID>
                  <authentication parameter>: <authentication parameter value>
                  <authentication parameter>: <authentication parameter value>
                  ...

Where:

• databaseHostname is the hostname or IP address of the MySQL CDC database server.

• databasePort is the connection port of the MySQL CDC database server.

• databaseUser is the MySQL CDC user to log in as. The MySQL CDC user must have replication privileges (REPLICATION SLAVE, REPLICATION CLIENT).

• databaseServerId is a unique server ID for replication. This must be a unique integer identifier for the connector instance.

• <authentication parameter> is a parameter required for authenticating with the MySQL CDC database server. Choose one of:

  • Basic Authentication—Use a username and password to authenticate. For more information, see Basic Authentication Parameters.

  • Client Certificate Authentication—Use a client certificate to authenticate. For more information, see Client Certificate Authentication Parameters.

Basic Authentication Parameters

Field	Type	Description
databaseUser	String	The username to use for logging in to MySQL CDC. The user must have replication privileges.
databasePassword	String	The password to use for authenticating with the MySQL CDC database.

Client Certificate Authentication Parameters

For SSL/TLS certificate-based authentication, configure the following SSL properties in addition to the basic connection parameters.

Field	Type	Description
databaseSslMode	String	Specifies whether to use an encrypted connection. Set to one of: disabled: Establish an unencrypted connection. preferred: Establish an encrypted connection if the server supports encrypted connections, falling back to an unencrypted connection if an encrypted connection cannot be established. This is the default if databaseSslMode is not specified. required: Establish an encrypted connection if the server supports encrypted connections. The connection attempt fails if an encrypted connection cannot be established. verify_ca: Like required, but additionally verify the server Certificate Authority (CA) certificate against the configured CA certificates. The connection attempt fails if no valid matching CA certificates are found. verify_identity: Like verify_ca, but additionally perform host name identity verification by checking the host name the client uses for connecting to the server against the identity in the certificate that the server sends to the client. For more information see the MySQL CDC documentation for SSL Mode.
databaseSslKeystore	String	The path to the Java keystore file containing the client certificate and private key.
databaseSslKeystorePassword	String	The password for the keystore file.
databaseSslTruststore	String	The path to the Java truststore file containing the trusted CA certificates for verifying the MySQL CDC server certificate.
databaseSslTruststorePassword	String	The password for the truststore file.

Checkpoint Store Configuration

The MySQL CDC stores the information about the current progress of processing change events in a checkpoint store backed by a Last Value Queue (LVQ) on a Solace event broker. The checkpoint store contains information about the binlog position and GTID that have been processed.

For example, add the following settings to the application.yml file:

solace:
  mysqlcdc:
    checkpoint:
      lvqName: lvq-checkpoint
      autoProvisionLvq: true

The following table lists the configuration options for the Checkpoint Store.

Config Option	Type	Valid Values	Default Value	Description
solace.mysqlcdc.checkpoint.lvqName	String			Required. The name of the event broker Last Value Queue (spool size 0) to be used for checkpointing. The queue must exist on the same event broker and Message VPN as the target queue. If the LVQ is deleted (administratively) or a message from the LVQ is deleted or consumed by another consumer, the MySQL CDC will not be able to resume from the last checkpoint. In addition, the LVQ should not be shared by multiple instances of the MySQL CDC.
solace.mysqlcdc.checkpoint.autoProvisionLvq	boolean	true, false	false	Optional. Set to true to auto-provision the LVQ queue if it does not exist.

MySQL CDC Binder Configuration Options

The following properties are available at the MySQL CDC binder level.

Config Option	Type	Valid Values	Description
spring.cloud.stream.bindings.input-<workflow-id>.destination	String	Format: database.tableName For example, inventory.customers	The MySQL CDC table from which to consume change events. The destination must include the database name and the name of the table.
spring.cloud.stream.bindings.input-<workflow-id>.binder	String	mysqlcdc	This property must be set to mysqlcdc.

MySQL CDC Consumer Configuration Options

The following table lists the source-specific properties for Debezium MySQL CDC.

All properties must be prefixed with spring.cloud.stream.mysqlcdc.bindings.input-<workflow-id>.consumer.endpoint.query-parameters.

Example configuration:

spring:
  cloud:
    stream:
      mysqlcdc:
        bindings:
           input-0:
             consumer:
               endpoint:
                 query-parameters:
                   <connection parameters>
                   databaseServerId: 12345
                   snapshotMode: no_data
                   skipDeleteEvents: false
                   offsetFlushIntervalMs: 60000
                   bigintUnsignedHandlingMode: long
                   decimalHandlingMode: precise
                   timePrecisionMode: adaptive_time_microseconds

Config Option	Type	Valid Values	Default Value	Description
databaseHostname	String			Required. The MySQL CDC server hostname or IP address.
databasePort	Integer			Required. The MySQL CDC server port.
databaseUser	String			Required. The database username with replication privileges.
databasePassword	String			Required. The Database password.
databaseServerId	Integer			Required. The unique server ID for replication.
skipDeleteEvents	Boolean	true, false	false	Indicates whether to skip DELETE operation events to avoid transformation issues with empty payloads.
snapshotMode	String	initial, schema_only, when_needed, never, no_data, initial_only	no_data	Controls initial snapshot behavior. Select one of the following snapshot options: initial—The default. Takes a snapshot on first startup, then streams changes. schema_only—Schema structure only, no data. Deprecated, see no_data. when_needed—Takes a snapshot when the Micro-Integration determines it's needed. never—Never takes snapshots. no_data—Schema structure only, no data. initial_only—Takes a snapshot but doesn't transition to streaming.
skippedOperations	String	c, u, d, t, none	t	Operations to skip: c (create), u (update), d (delete), t (truncate), none.
offsetFlushIntervalMs	Integer		60000	The interval for flushing offsets (milliseconds).
bigintUnsignedHandlingMode	String	precise, long	long	Specify how BIGINT UNSIGNED columns should be represented in change events: precise—use java.math.BigDecimal long—represent values using Java long
decimalHandlingMode	String	precise, string, double	precise	Specify how DECIMAL and NUMERIC columns should be represented in change events: precise—use java.math.BigDecimal string—use string double—use Java double
timePrecisionMode	String	adaptive_time_microseconds, connect	adaptive_time_microseconds	Specifies precision for time, date and timestamps values: adaptive_time_microseconds—precision is based on the database column's precision connect—uses millisecond precision regardless of database columns' precision.

Headers

The Micro-Integration injects the following CDC-related headers in each message:

Header Name	Type	Description
cdc_operation	String	The event operation type: c (create), u (update), d (delete), r (read/snapshot), t (truncate).
cdc_db_name	String	The source database name.
cdc_table_name	String	The source table name.
cdc_key	JSON String	The primary key of the record.
cdc_timestamp	Long	The event processing timestamp.
cdc_before	JSON String	The record state before change (UPDATE/DELETE).
cdc_file	String	The binlog file name.
cdc_gtid	String	The Global Transaction ID (if enabled).
cdc_pos	Long	The position in binlog file.
cdc_query	String	The DDL query (for schema changes).
cdc_row	Integer	The row number in event.
cdc_server_id	Integer	The MySQL server ID.
cdc_thread	Integer	The MySQL thread ID.
cdc_after_schema	String	The schema of the payload.