anchor.diff:  ifdef::community[]
   Typically, you configure a {prodname} MySQL connector in a `.json` file that sets configuration properties for the connector. Following is an example configuration for a MySQL connector that connects to a MySQL server on port 3306 at 192.168.99.100, whose logical name is `fullfillment`.
   
  -For details, see {link-prefix}:{link-mysql-connector}#mysql-connector-configuration-properties[MySQL connector configuration properties].
  +For details, see {link-prefix}:{link-mysql-connector}#mysql-configuration-properties[MySQL connector configuration properties].
   
   [source,json]
   ----
  
positive.code: 
  MySQL has a binary log (binlog) that records all operations in the order in which they are committed to the database. This includes changes to table schemas as well as changes to the data in tables. MySQL uses the binlog for replication and recovery.
  
  The {prodname} MySQL connector reads the binlog and produces change events for row-level `INSERT`, `UPDATE`, and `DELETE` operations and records the change events in a Kafka topic. Client applications read those Kafka topics.
  
  As MySQL is typically set up to purge binlogs after a specified period of time, the MySQL connector performs an initial _consistent snapshot_ of each of your databases. The MySQL connector reads the binlog from the point at which the snapshot was made.
  
  
negative.code: 
  endif::product[]
  
  // Type: concept
  // ModuleID: how-debezium-mysql-connectors-handle-schema-changes
  // Title: How {prodname} MySQL connectors handle schema changes
  [[mysql-schema-history-topic]]
  === Database schema history
  
  When a database client queries a database, the client uses the database’s current schema. However, the database schema can be changed at any time, which means that the connector must be able to identify what the schema was at the time each insert, update, or delete operation was recorded. Also, a connector cannot just use the current schema because the connector might be processing events that are relatively old and may have been recorded before the tables' schemas were changed. 
  
  To handle this, MySQL includes in the binlog not only the row-level changes to the data, but also the DDL statements that are applied to the database. As the connector reads the binlog and comes across these DDL statements, it parses them and updates an in-memory representation of each table’s schema. The connector uses this schema representation to identify the structure of the tables at the time of each insert, update, or delete operation and to produce the appropriate change event. In a separate database history Kafka topic, the connector records all DDL statements along with the position in the binlog where each DDL statement appeared.
  
  When the connector restarts after having crashed or been stopped gracefully, the connector starts reading the binlog from a specific position, that is, from a specific point in time. The connector rebuilds the table structures that existed at this point in time by reading the database history Kafka topic and parsing all DDL statements up to the point in the binlog where the connector is starting.
  
  This database history topic is for connector use only. The connector can optionally emit schema change events to a different topic that is intended for consumer applications. See {link-prefix}:{link-mysql-connector}#how-the-mysql-connector-exposes-schema-changes[how the MySQL connector exposes schema changes].
  
  When the MySQL connector captures changes in a table to which a schema change tool such as `gh-ost` or `pt-online-schema-change` is applied there are helper tables created during the migration process. The connector needs to be configured to capture change to these helper tables. If consumers do not need the records generated for helper tables then a simple message transform can be applied to filter them out.
  
  See {link-prefix}:{link-mysql-connector}#mysql-topic-names[default names for topics] that receive {prodname} event records.
  
  // Type: concept
  // ModuleID: how-debezium-mysql-connectors-expose-schema-changes
  // Title: How {prodname} MySQL connectors exposes schema changes
  [[mysql-schema-change-topic]]
  === Schema change topic
  
  You can configure a {prodname} MySQL connector to produce schema change events that include all DDL statements applied to databases in the MySQL server. The connector emits these events to a Kafka topic named _serverName_ where _serverName_ is the name of the connector as specified by the `database.server.name` connector configuration property.
  
  If you choose to use _schema change events_, ensure that you consume records from the schema change topic. The database history topic is for connector use only. 
  
  IMPORTANT: A global order for events emitted to the schema change topic is vital. Therefore, you must not partition the database history topic. This means that you must specify a partition count of `1` when creating the database history topic. When relying on auto topic creation, make sure that Kafka’s `num.partitions` configuration option, which specifies the default number of partitions, is set to `1`.
  
  Each message that is written to the schema change topic contains a message key that includes the name of the connected database used when applying DDL statements, for example: 
  
  [source,json,subs="+attributes"]
  ----
  {
    "schema": {
      "type": "struct",
      "name": "io.debezium.connector.mysql.SchemaChangeKey",
      "optional": false,
      "fields": [
        {
          "field": "databaseName",
          "type": "string",
          "optional": false
        }
      ]
    },
    "payload": {
      "databaseName": "inventory"
    }
  }
  ----
  
  The schema change event record value contains a structure that includes the DDL statements, the database to which the statements were applied, and the position in the binlog where the statements appeared, for example: 
  
  [source,json,subs="attributes"]
  ----
  {
    "schema": {
      "type": "struct",
      "name": "io.debezium.connector.mysql.SchemaChangeValue",
      "optional": false,
      "fields": [
        {
          "field": "databaseName",
          "type": "string",
          "optional": false
        },
        {
          "field": "ddl",
          "type": "string",
          "optional": false
        },
        {
          "field": "source",
          "type": "struct",
          "name": "io.debezium.connector.mysql.Source",
          "optional": false,
          "fields": [
            {
              "type": "string",
              "optional": true,
              "field": "version"
            },
            {
              "type": "string",
              "optional": false,
              "field": "name"
            },
            {
              "type": "int64",
              "optional": false,
              "field": "server_id"
            },
            {
              "type": "int64",
              "optional": false,
              "field": "ts_sec"
            },
            {
              "type": "string",
              "optional": true,
              "field": "gtid"
            },
            {
              "type": "string",
              "optional": false,
              "field": "file"
            },
            {
              "type": "int64",
              "optional": false,
              "field": "pos"
            },
            {
              "type": "int32",
              "optional": false,
              "field": "row"
            },
            {
              "type": "boolean",
              "optional": true,
              "default": false,
              "field": "snapshot"
            },
            {
              "type": "int64",
              "optional": true,
              "field": "thread"
            },
            {
              "type": "string",
              "optional": true,
              "field": "db"
            },
            {
              "type": "string",
              "optional": true,
              "field": "table"
            },
            {
              "type": "string",
              "optional": true,
              "field": "query"
            }
          ]
        }
      ]
    },
    "payload": {
      "databaseName": "inventory",
      "ddl": "CREATE TABLE products ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255) NOT NULL, description VARCHAR(512), weight FLOAT ); ALTER TABLE products AUTO_INCREMENT = 101;",
      "source" : {
        "version": "0.10.0.Beta4",
        "name": "mysql-server-1",
        "server_id": 0,
        "ts_sec": 0,
        "gtid": null,
        "file": "mysql-bin.000003",
        "pos": 154,
        "row": 0,
        "snapshot": true,
        "thread": null,
        "db": null,
        "table": null,
        "query": null
      }
    }
  }
  ----
  
  The `ddl` field might contain multiple DDL statements. Each statement applies to the database in the `databaseName` field. The statements appear in the order in which they were applied to the database. The `source` field is structured exactly as a standard data change event written to table-specific topics. This field is useful to correlate events on different topics.
  
  [source,json,subs="+attributes"]
  ----
  ....
  "payload": {
      "databaseName": "inventory",
      "ddl": "CREATE TABLE products ( id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,...)",
      "source" : {
          ...
      }
  }
  ....
  ----
  
  A client can submit multiple DDL statements to be applied to multiple databases. If MySQL applies them atomically, the connector takes the DDL statements in order, groups them by database, and creates a schema change event for each group. If MySQL applies them individually, the connector creates a separate schema change event for each statement.
  
  See also: {link-prefix}:{link-mysql-connector}#mysql-schema-history-topic[Database schema history topic].
  
  // Type: concept
  // ModuleID: how-debezium-mysql-connectors-perform-database-snapshots
  