anchor.diff:      mysql> INSERT INTO customers VALUES (default, "Kenneth", "Anderson", "kander@acme.com");
   ----
   
  -Notice that in the terminal where we're running `watch-topic`, there's been no update. Also, we're still able to watch the topic because Kafka is still running. (In a production system, you would have enough brokers to handle the producers and consumers, and to maintain a minimum number of in sync replicas for each topic. So if enough brokers fail such that there are not the minimum number of ISRs, Kafka should become unavailable. Producers, like the Debezium connectors, and consumers will simply wait patiently for the Kafka cluster or network to recover. Yes, that means that your consumers might temporarily see no change events as data is changed in the databases, but that's because none are being produced. As soon as the Kafka cluster is restarted or the network recovers, Debezium will continue producing change events while your consumers will continue consuming events where they left off.)
  +Notice that in the terminal where we're running `watch-topic`, there's been no update. Also, we're still able to watch the topic because Kafka is still running. 
  +
  +[TIP]
  +====
  +In a production system, you would have enough brokers to handle the producers and consumers, and to maintain a minimum number of in sync replicas for each topic. So if enough brokers fail such that there are not the minimum number of ISRs, Kafka should become unavailable. Producers, like the Debezium connectors, and consumers will simply wait patiently for the Kafka cluster or network to recover. Yes, that means that your consumers might temporarily see no change events as data is changed in the databases, but that's because none are being produced. As soon as the Kafka cluster is restarted or the network recovers, Debezium will continue producing change events and your consumers will continue consuming events where they left off.
  +====
   
   Now, in a new terminal, start a new container using the _same_ command we used before:
   
   [source,bash,indent=0]
   ----
  -    $ docker run -it --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my-connect-configs -e OFFSET_STORAGE_TOPIC=my-connect-offsets -e ADVERTISED_HOST_NAME=$(echo $DOCKER_HOST | cut -f3  -d'/' | cut -f1 -d':') --link zookeeper:zookeeper --link kafka:kafka debezium/connect:0.3
  -----
  -
  -This creates a whole new container, and since we've intialized it with the same topic information the new service can connect to Kafka, read the previous service's configuration and start the registered connectors, which will continue where they last left off.
  -
  -Jump back to the terminal running `watch-topic`, and you should now see two new records we added to the MySQL database:
  -
  -[source,json,indent=0]
  -----
  -    {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"mysql-server-1.inventory.customers.Key"},"payload":{"id":1005}}   {"schema":{"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":"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,"field":"snapshot"}],"optional":false,"name":"io.debezium.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","version":1},"payload":{"before":null,"after":{"id":1005,"first_name":"Sarah","last_name":"Thompson","email":"kitt@acme.com"},"source":{"name":"mysql-server-1","server_id":223344,"ts_sec":1465583,"gtid":null,"file":"mysql-bin.000003","pos":1115,"row":0,"snapshot":null},"op":"c","ts_ms":1465583022619}}
  -    {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"mysql-server-1.inventory.customers.Key"},"payload":{"id":1006}}   {"schema":{"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":"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,"field":"snapshot"}],"optional":false,"name":"io.debezium.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","version":1},"payload":{"before":null,"after":{"id":1006,"first_name":"Kenneth","last_name":"Anderson","email":"kander@acme.com"},"source":{"name":"mysql-server-1","server_id":223344,"ts_sec":1465583,"gtid":null,"file":"mysql-bin.000003","pos":1429,"row":0,"snapshot":null},"op":"c","ts_ms":1465583031183}}
  +    $ docker run -it --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my-connect-offsets --link zookeeper:zookeeper --link kafka:kafka debezium/connect:0.3
  +----
  +
  +This creates a whole new container that runs the Kafka Connect distributed service, and since we've intialized it with the same topic information, the new service connects to Kafka, read the previous service's configuration, and starts the registered connectors that will continue exactly where they last left off.
  +
  +Here's the last few lines from this restarted service:
  +
  +[source,bash,indent=0]
  +----
  +...
  +2016-08-16 02:50:06,437 INFO   MySQL|dbserver1|task  Found existing offset: {ts_sec=1471315, file=mysql-bin.000003, pos=438, row=0, server_id=223344}   [io.debezium.connector.mysql.MySqlConnectorTask]
  +...
  +2016-08-16 02:50:06,477 INFO   MySQL|dbserver1|task  Kafka version : 0.10.0.1   [org.apache.kafka.common.utils.AppInfoParser]
  +2016-08-16 02:50:06,477 INFO   MySQL|dbserver1|task  Kafka commitId : a7a17cdec9eaa6c5   [org.apache.kafka.common.utils.AppInfoParser]
  +2016-08-16 02:50:06,581 INFO   MySQL|dbserver1|task  Discovered coordinator 172.17.0.3:9092 (id: 2147483646 rack: null) for group 3ad7c83a-da2f-4ede-b9c6-64c058ed3ba6.   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 02:50:06,583 INFO   MySQL|dbserver1|task  Revoking previously assigned partitions [] for group 3ad7c83a-da2f-4ede-b9c6-64c058ed3ba6   [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator]
  +2016-08-16 02:50:06,583 INFO   MySQL|dbserver1|task  (Re-)joining group 3ad7c83a-da2f-4ede-b9c6-64c058ed3ba6   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 02:50:06,593 INFO   MySQL|dbserver1|task  Successfully joined group 3ad7c83a-da2f-4ede-b9c6-64c058ed3ba6 with generation 1   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 02:50:06,594 INFO   MySQL|dbserver1|task  Setting newly assigned partitions [dbhistory.inventory-0] for group 3ad7c83a-da2f-4ede-b9c6-64c058ed3ba6   [org.apache.kafka.clients.consumer.internals.ConsumerCoordinator]
  +2016-08-16 02:50:07,105 INFO   MySQL|dbserver1|task  Step 0: Get all known binlogs from MySQL   [io.debezium.connector.mysql.MySqlConnectorTask]
  +...
  +2016-08-16 02:50:07,643 INFO   MySQL|dbserver1|binlog  Connected to MySQL binlog at 192.168.99.100:3306, starting at binlog file 'mysql-bin.000003', pos=438, row=0   [io.debezium.connector.mysql.BinlogReader]
  +2016-08-16 02:50:07,643 INFO   ||  Source task WorkerSourceTask{id=inventory-connector-0} finished initialization and start   [org.apache.kafka.connect.runtime.WorkerSourceTask]
  +----
  +
  +As you can see, these lines show that the service finds the offsets previously recorded by the last task before it was shut down, and that it then connects to the MySQL database, starts reading the binlog from that position, and generates events from any changes in the MySQL database since that point in time.
  +
  +Jump back to the terminal running `watch-topic`, and you should now see events for our two new records:
  +
  +[source,json,indent=0]
  +----
  +{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1005}}	{"schema":{"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":"dbserver1.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":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"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,"field":"snapshot"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1005,"first_name":"Sarah","last_name":"Thompson","email":"kitt@acme.com"},"source":{"name":"dbserver1","server_id":223344,"ts_sec":1471314,"gtid":null,"file":"mysql-bin.000003","pos":1046,"row":0,"snapshot":null},"op":"c","ts_ms":1471314885877}}
  +{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1006}}	{"schema":{"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":"dbserver1.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":"dbserver1.inventory.customers.Value","field":"after"},{"type":"struct","fields":[{"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,"field":"snapshot"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope","version":1},"payload":{"before":null,"after":{"id":1006,"first_name":"Kenneth","last_name":"Anderson","email":"kander@acme.com"},"source":{"name":"dbserver1","server_id":223344,"ts_sec":1471314,"gtid":null,"file":"mysql-bin.000003","pos":1356,"row":0,"snapshot":null},"op":"c","ts_ms":1471314885878}}
   ----
   
   These events are _create_ events that are similar to what we saw before. The important point to understand, though, is that Debezium will still report all of the changes in a database even when it is not running, as long as it is restarted before the MySQL database starts purging those commits we missed from its binlog.
  
positive.code:   {
      "schema": {
        "type": "struct",
        "name": "mysql-server-1.inventory.customers.Key"
        "optional": false,
        "fields": [
          {
  
negative.code:   }
  ----
  
  The event's key has two parts: a `schema` and `payload`. The `schema` contains a Kafka Connect schema describing what is in the payload, and in our case that means that the `payload` is a struct named `dbserver1.inventory.customers.Key` that is not optional and has one required field named `id` of type `int32`.
  
  If we look at the value of the key's `payload` field, we'll see that it is indeed a structure (which in JSON is just an object) with a single `id` field, whose value is `1004`.
  
  Therefore, we interpret this event as applying to the row in the `inventory.customers` table (output from the connector named `dbserver1`) whose `id` primary key column had a value of `1004`.
  
  Now let's look at the same event's _value_, which again we reformat to be easier to read:
  
  [source,json,indent=0]
  ----
  {
    "schema": {
      "name": "dbserver1.inventory.customers.Envelope",
      "version": 1,
      "optional": false,
      "type": "struct",
      "fields": [
        {
          "field": "before",
          "name": "dbserver1.inventory.customers.Value",
          "optional": true,
          "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"
            }
          ]
        },
        {
          "field": "after",
          "name": "dbserver1.inventory.customers.Value",
          "optional": true,
          "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"
            }
          ]
        },
        {
          "field": "source",
          "name": "io.debezium.connector.mysql.Source",
          "optional": false,
          "type": "struct",
          "fields": [
            {
              "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,
              "field": "snapshot"
            }
          ]
        },
        {
          "type": "string",
          "optional": false,
          "field": "op"
        },
        {
          "type": "int64",
          "optional": true,
          "field": "ts_ms"
        }
      ]
    },
    "payload": {
      "before": null,
      "after": {
        "id": 1004,
        "first_name": "Anne",
        "last_name": "Kretchmar",
        "email": "annek@noanswer.org"
      },
      "source": {
        "name": "dbserver1",
        "server_id": 0,
        "ts_sec": 0,
        "gtid": null,
        "file": "mysql-bin.000003",
        "pos": 154,
        "row": 0,
        "snapshot": true
      },
      "op": "c",
      "ts_ms": 1471310719597
    }
  }
  ----
  
  This portion of the event is much larger, but like the event's _key_ this, too, has a `schema` and a `payload`. The `schema` contains a Kafka Connect schema named `dbserver1.inventory.customers.Envelope` (version 1) that can contain 5 fields:
  
  * `op` is a mandatory field that contains a string value describing the type of operation. Values for the MySQL connector are `c` for create (or insert), `u` for update, `d` for delete, and `r` for read (in the case of a non-initial snapshot).
  * `before` is an optional field that if present contains the state of the row _before_ the event occurred. The structure will  be described by the `dbserver1.inventory.customers.Value` Kafka Connect schema, which the `dbserver1` connector uses for all rows in the `inventory.customers` table.
  * `after` is an optional field that if present contains the state of the row _after_ the event occurred. The structure is describe by the same `dbserver1.inventory.customers.Value` Kafka Connect schema used in `before`.
  * `source` is a mandatory field that conains a structure describing the source metadata for the event, which in the case of MySQL contains several fields: the connector name, the name of the binlog file where the event was recorded, the position in that binlog file where the event appeared, the row within the event (if there is more than one), whether this event was part of a snapshot, and if available the MySQL server ID, and the timestamp in seconds.
  * `ts_ms` is optional and if present contains the time (using the system clock in the JVM running the Kafka Connect task) at which the connector processed the event. 
  
  