// Metadata created by nebel
//
[id="mysql-connector-events_{context}"]
= MySQL connector events

All data change events produced by the {prodname} MySQL connector contain a key and a value. The change event key and the change event value each contain a _schema_ and a _payload_ where the schema describes the structure of the payload and the payload contains the data.

WARNING: The MySQL connector ensures that all Kafka Connect schema names adhere to the link:http://avro.apache.org/docs/current/spec.html#names[Avro schema name format]. This is important as any character that is not a latin letter or underscore is replaced by an underscore which can lead to unexpected conflicts in schema names when the logical server names, database names, and table names container other characters that are replaced with these underscores.

== Change event key

For any given table, the change event's key has a structure that contains a field for each column in the `PRIMARY KEY` (or unique constraint) at the time the event was created. Let us look at an example table and then how the schema and payload would appear for the table.

.example table
[source,sql]
----
CREATE TABLE customers (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE KEY
) AUTO_INCREMENT=1001;
----

=====
.example change event key
[source,json]
----
{
 "schema": { <1>
    "type": "struct",
 "name": "mysql-server-1.inventory.customers.Key", <2>
 "optional": false, <3>
 "fields": [ <4>
      {
        "field": "id",
        "type": "int32",
        "optional": false
      }
    ]
  },
 "payload": { <5>
    "id": 1001
  }
}
----
<1> The `schema` describes what is in the `payload`.
<2> The `mysql-server-1.inventory.customers.Key` is the name of the schema which defines the structure where `mysql-server-1` is the connector name, `inventory` is the database, and `customers` is the table.
<3> Denotes that the `payload` is not optional.
<4> Specifies the type of fields expected in the `payload`.
<5> The payload itself, which in this case only contains a single `id` field.

This key describes the row in the `inventory.customers` table which is out from the connector entitled `mysql-server-1` whose `id` primary key column has a value of `1001`.
=====

== Change event value

The change event value contains a schema and a payload section. There are three types of change event values which have an envelope structure. The fields in this structure are explained below and marked on each of the change event value examples.

* <<Create change event value>>
* <<Update change event value>>
* <<Delete change event value>>

[cols="1,2,7"]
|===
|Item |Field name |Description

|1
| `name`
| `mysql-server-1.inventory.customers.Key` is the name of the schema which defines the structure where `mysql-server-1` is the connector name, `inventory` is the database and `customers` is the table

|2
|`op`
a| A *mandatory* string that describes the type of operation.

.values
* `c` = create
* `u` = update
* `d` = delete
* `r` = read (_initial snapshot_ only)

|3
|`before`
| An optional field that specifies the state of the row before the event occurred.

|4
|`after`
| An optional field that specifies the state of the row after the event occurred.

|5
|`source`
a| A *mandatory* field that describes the source metadata for the event including:

* the {prodname} version
* the connector name
* the binlog name where the event was recorded
* the binlog position
* the row within the event
* if the event was part of a snapshot
* the name of the affected database and table
* the id of the MySQL thread creating the event (non-snapshot only)
* the MySQL server ID (if available)
* timestamp

NOTE: If the {link-prefix}:{link-mysql-connector}#enable-query-log-events-for-cdc_{context}[binlog_rows_query_log_events] option is enabled and the connector has the `include.query` option enabled, a `query` field is displayed which contains the original SQL statement that generated the event.

|6
|`ts_ms`
a| An optional field that displays the time at which the connector processed the event.

NOTE: The time is based on the system clock in the JVM running the Kafka Connect task.

|===

Let us look at an example table and then how the schema and payload would appear for the table.

.example table
[source,sql]
----
CREATE TABLE customers (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  first_name VARCHAR(255) NOT NULL,
  last_name VARCHAR(255) NOT NULL,
  email VARCHAR(255) NOT NULL UNIQUE KEY
) AUTO_INCREMENT=1001;
----

=== Create change event value

This example shows a _create_ event for the `customers` table:

[source,json,options="nowrap",subs="+attributes"]
----
{
  "schema": { // <1>
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "mysql-server-1.inventory.customers.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "mysql-server-1.inventory.customers.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "version"
          },
          {
            "type": "string",
            "optional": false,
            "field": "connector"
          },
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_ms"
          },
          {
            "type": "boolean",
            "optional": true,
            "default": false,
            "field": "snapshot"
          },
          {
            "type": "string",
            "optional": false,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "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": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "query"
          }
        ],
        "optional": false,
        "name": "io.product.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "mysql-server-1.inventory.customers.Envelope"
  },
  "payload": { // <2>
    "op": "c",
    "ts_ms": 1465491411815,
    "before": null,
    "after": {
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": {
      "version": "{debezium-version}",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 0,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 154,
      "row": 0,
      "thread": 7,
      "query": "INSERT INTO customers (first_name, last_name, email) VALUES ('Anne', 'Kretchmar', 'annek@noanswer.org')"
    }
  }
}
----
<1> The `schema` portion of this event’s _value_ shows the schema for the envelope, the schema for the source structure (which is specific to the MySQL connector and reused across all events), and the table-specific schemas for the `before` and `after` fields.
ifdef::community[]
+
[TIP]
====
The names of the schemas for the `before` and `after` fields are of the form `<logicalName>.<tableName>.Value`, and thus are entirely independent from all other schemas for all other tables. This means that when using the {link-prefix}:{link-avro-serialization}[Avro Converter], the resulting Avro schemas for each table in each logical source have their own evolution and history.
====
endif::community[]

<2> The `payload` portion of this event’s _value_ shows the information in the event, namely that it is describing that the row was created (because `op=c`), and that the `after` field value contains the values of the new inserted row's `id`, `first_name`, `last_name`, and `email` columns.
ifdef::community[]
+
[TIP]
====
It may appear that the JSON representations of the events are much larger than the rows they describe. This is because the JSON representation must include the schema and the payload portions of the message.
However, by using the {link-prefix}:{link-avro-serialization}[Avro Converter], you can dramatically decrease the size of the actual messages written to the Kafka topics.
====
endif::community[]

=== Update change event value

The value of an _update_ change event on the `customers` table has the exact same schema as a _create_ event. The payload is structured the same, but holds different values. Here is an example (formatted for readability):

[source,json,options="nowrap",subs="+attributes"]
----
{
  "schema": { ... },
  "payload": {
    "before": { // <1>
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": { // <2>
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "source": { // <3>
      "version": "{debezium-version}",
      "name": "mysql-server-1",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 1465581,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 484,
      "row": 0,
      "thread": 7,
      "query": "UPDATE customers SET first_name='Anne Marie' WHERE id=1004"
    },
    "op": "u", // <4>
    "ts_ms": 1465581029523 // <5>
  }
}
----

Comparing this to the value in the _insert_ event, you can see a couple of differences in the `payload` section:

<1> The `before` field now has the state of the row with the values before the database commit.
<2> The `after` field now has the updated state of the row, and the `first_name` value is now `Anne Marie`. You can compare the `before` and `after` structures to determine what actually changed in this row because of the commit.
<3> The `source` field structure has the same fields as before, but the values are different (this event is from a different position in the binlog). The `source` structure shows information about MySQL’s record of this change (providing traceability). It also has information you can use to compare to other events in this and other topics to know whether this event occurred before, after, or as part of the same MySQL commit as other events.
<4> The `op` field value is now `u`, signifying that this row changed because of an update.
<5> The `ts_ms` field shows the timestamp when {prodname} processed this event.

[NOTE]
====
When the columns for a row’s primary or unique key are updated, the value of the row’s key is changed and {prodname} outputs three events: a _DELETE_ event and tombstone event with the old key for the row, followed by an _INSERT_ event with the new key for the row.
====

=== Delete change event value

The value of a _delete_ change event on the `customers` table has the exact same schema as _create_ and _update_ events. The payload is structured the same, but holds different values. Here is an example (formatted for readability):


[source,json,options="nowrap",subs="+attributes"]
----
{
  "schema": { ... },
  "payload": {
    "before": { // <1>
      "id": 1004,
      "first_name": "Anne Marie",
      "last_name": "Kretchmar",
      "email": "annek@noanswer.org"
    },
    "after": null, // <2>
    "source": { // <3>
      "version": "{debezium-version}",
      "connector": "mysql",
      "name": "mysql-server-1",
      "ts_ms": 1465581,
      "snapshot": false,
      "db": "inventory",
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 805,
      "row": 0,
      "thread": 7,
      "query": "DELETE FROM customers WHERE id=1004"
    },
    "op": "d", // <4>
    "ts_ms": 1465581902461 // <5>
  }
}
----
Comparing the `payload` portion to the payloads in the _create_ and _update_ events, you can see some differences:

<1> The `before` field now has the state of the row that was deleted with the database commit.
<2> The `after` field is `null`, signifying that the row no longer exists.
<3> The `source` field structure has many of the same values as before, except the `ts_sec` and `pos` fields have changed (and the file might have changed in other scenarios).
<4> The `op` field value is now `d`, signifying that this row was deleted.
<5> The `ts_ms` shows the timestamp when {prodname} processed this event.

This event provides a consumer with the information that it needs to process the removal of this row. The old values are included because some consumers might require them in order to properly handle the removal.

The MySQL connector’s events are designed to work with link:{link-kafka-docs}#compaction[Kafka log compaction], which allows for the removal of some older messages as long as at least the most recent message for every key is kept. This allows Kafka to reclaim storage space while ensuring the topic contains a complete data set and can be used for reloading key-based state.

When a row is deleted, the _delete_ event value listed above still works with log compaction, because Kafka can still remove all earlier messages with that same key. If the message value is `null`, Kafka knows that it can remove all messages with that same key. To make this possible, {prodname}’s MySQL connector always follows a _delete_ event with a special tombstone event that has the same key but a `null` value.
