anchor.diff:  
   [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=$(docker-machine ip) --link zookeeper:zookeeper --link kafka:kafka debezium/connect:0.3
  -----
  -
  -This runs a new container using version {debezium-docker-label} of the `debezium/connect` image, and assigns the name `connect` to this container. The `-it` flag makes the container interactive, meaning it attaches the terminal's standard input and output to the container so that you can see what is going on in the container. The command maps port 8083 in the container to the same port on the Docker host so that software outside of the container can use Kafka Connect's REST API to set up and manage new connector instances. The command uses the `--link zookeeper:zookeeper` and `--link kafka:kafka` argument to tell the container that it can find Zookeeper and Kafka in the container named `zookeeper` and `kafka`, respectively, running on the same Docker host. And finally, it also uses the `-e` option four times to set the `GROUP_ID`, `CONFIG_STORAGE_TOPIC`, `OFFSET_STORAGE_TOPIC`, and `ADVERTISED_HOST_NAME` environment variables; the first three are required by this container (you can use different values as desired), while the last variable is optional but is used by the Kafka Connect server process to tell clients and other service instances the address at which it is listening. In our case, we use the `docker-machine ip` command to optain the IP address of the Docker host.
  +    $ docker run -it --rm --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 runs a new container named `connect` using version {debezium-docker-label} of the `debezium/connect` image. The `-it` flag makes the container interactive, meaning it attaches the terminal's standard input and output to the container so that you can see what is going on in the container. The `--rm` flag instructs Docker to remove the container when it is stopped. The command maps port 8083 in the container to the same port on the Docker host so that software outside of the container can use Kafka Connect's REST API to set up and manage new connector instances. The command uses the `--link zookeeper:zookeeper` and `--link kafka:kafka` argument to tell the container that it can find Zookeeper and Kafka in the container named `zookeeper` and `kafka`, respectively, running on the same Docker host. And finally, it also uses the `-e` option three times to set the `GROUP_ID`, `CONFIG_STORAGE_TOPIC`, and `OFFSET_STORAGE_TOPIC` environment variables, which are all required by this Debezium image (though you can use different values as desired).
   
   You should see in your terminal the typical output of Kafka, ending with:
   
   [listing,indent=0,options="nowrap"]
   ----
  -    ...
  -    2016-06-09 15:35:22,257 - INFO  [DistributedHerder:AppInfoParser$AppInfo@82] - Kafka version : 0.9.0.1
  -    2016-06-09 15:35:22,259 - INFO  [DistributedHerder:AppInfoParser$AppInfo@83] - Kafka commitId : 23c69d62a0cabf06
  -    2016-06-09 15:35:22,491 - INFO  [main:Server@327] - jetty-9.2.12.v20150709
  -    2016-06-09 15:35:22,837 - INFO  [DistributedHerder:KafkaBasedLog@143] - Finished reading KafakBasedLog for topic my-connect-configs
  -    2016-06-09 15:35:22,837 - INFO  [DistributedHerder:KafkaBasedLog@145] - Started KafakBasedLog for topic my-connect-configs
  -    2016-06-09 15:35:22,838 - INFO  [DistributedHerder:KafkaConfigStorage@242] - Started KafkaConfigStorage
  -    2016-06-09 15:35:22,838 - INFO  [DistributedHerder:DistributedHerder@156] - Herder started
  -    2016-06-09 15:35:23,112 - INFO  [DistributedHerder:DistributedHerder$14@868] - Joined group and got assignment: Assignment{error=0, leader='connect-1-f84dd8fb-ec0d-485f-8b3d-657746927ef2', leaderUrl='http://172.17.0.5:8083/', offset=-1, connectorIds=[], taskIds=[]}
  -    2016-06-09 15:35:23,119 - INFO  [DistributedHerder:DistributedHerder@639] - Starting connectors and tasks using config offset -1
  -    2016-06-09 15:35:23,120 - INFO  [DistributedHerder:DistributedHerder@659] - Finished starting connectors and tasks
  -    Jun 09, 2016 3:35:23 PM org.glassfish.jersey.internal.Errors logErrors
  -    WARNING: The following warnings have been detected: WARNING: The (sub)resource method listConnectors in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
  -    WARNING: The (sub)resource method createConnector in org.apache.kafka.connect.runtime.rest.resources.ConnectorsResource contains empty path annotation.
  -    WARNING: The (sub)resource method serverInfo in org.apache.kafka.connect.runtime.rest.resources.RootResource contains empty path annotation.
  -
  -    2016-06-09 15:35:23,706 - INFO  [main:ContextHandler@744] - Started o.e.j.s.ServletContextHandler@b78a709{/,null,AVAILABLE}
  -    2016-06-09 15:35:23,722 - INFO  [main:AbstractConnector@266] - Started ServerConnector@2e58f579{HTTP/1.1}{172.17.0.5:8083}
  -    2016-06-09 15:35:23,722 - INFO  [main:Server@379] - Started @5447ms
  -    2016-06-09 15:35:23,724 - INFO  [main:RestServer@132] - REST server listening at http://172.17.0.5:8083/, advertising URL http://172.17.0.5:8083/
  -    2016-06-09 15:35:23,724 - INFO  [main:Connect@60] - Kafka Connect started
  -----
  -
  -The last line shown above reports that the service has started and is ready for connections. The terminal will continue to show additional output as the Kafka Connect service generates it.
  +...
  +2016-08-16 01:22:42,083 INFO   ||  Kafka version : 0.10.0.1   [org.apache.kafka.common.utils.AppInfoParser]
  +2016-08-16 01:22:42,083 INFO   ||  Kafka commitId : a7a17cdec9eaa6c5   [org.apache.kafka.common.utils.AppInfoParser]
  +2016-08-16 01:22:42,552 INFO   ||  Discovered coordinator 172.17.0.3:9092 (id: 2147483646 rack: null) for group 1.   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 01:22:42,565 INFO   ||  Finished reading KafkaBasedLog for topic my_connect_configs   [org.apache.kafka.connect.util.KafkaBasedLog]
  +2016-08-16 01:22:42,566 INFO   ||  Started KafkaBasedLog for topic my_connect_configs   [org.apache.kafka.connect.util.KafkaBasedLog]
  +2016-08-16 01:22:42,566 INFO   ||  Started KafkaConfigBackingStore   [org.apache.kafka.connect.storage.KafkaConfigBackingStore]
  +2016-08-16 01:22:42,573 INFO   ||  Herder started   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
  +2016-08-16 01:22:42,681 INFO   ||  Discovered coordinator 172.17.0.3:9092 (id: 2147483646 rack: null) for group 1.   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 01:22:42,682 INFO   ||  (Re-)joining group 1   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 01:22:42,883 INFO   ||  Successfully joined group 1 with generation 1   [org.apache.kafka.clients.consumer.internals.AbstractCoordinator]
  +2016-08-16 01:22:42,883 INFO   ||  Joined group and got assignment: Assignment{error=0, leader='connect-1-f0fc4cad-6ff3-47ef-83dc-d1acf3c4ed82', leaderUrl='http://172.17.0.4:9092/', offset=-1, connectorIds=[], taskIds=[]}   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
  +2016-08-16 01:22:42,884 INFO   ||  Starting connectors and tasks using config offset -1   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
  +2016-08-16 01:22:42,885 INFO   ||  Finished starting connectors and tasks   [org.apache.kafka.connect.runtime.distributed.DistributedHerder]
  +2016-08-16 01:22:45,920 INFO   ||  Reflections took 3270 ms to scan 69 urls, producing 3313 keys and 23410 values    [org.reflections.Reflections]
  +----
  +
  +The last few line shown above reports that the service has started and is ready for connections. The terminal will continue to show additional output as the Kafka Connect service generates it.
   
   [[kafka-connect-api]]
   ==== Using the Kafka Connect REST API
  
positive.code:   {
      "schema": {
        "type": "struct",
        "name": "mysql-server-1.inventory.customers.Key"
        "optional": false,
        "fields": [
          {
  
negative.code: 
  [source,http,indent=0]
  ----
  HTTP/1.1 201 Created
  Date: Tue, 16 Aug 2016 01:25:16 GMT
  Location: http://192.168.99.100:8083/connectors/inventory-connector
  Content-Type: application/json
  Content-Length: 480
  Server: Jetty(9.2.15.v20160210)
  
  {
     "name":"inventory-connector",
     "config":{  
        "connector.class":"io.debezium.connector.mysql.MySqlConnector",
        "tasks.max":"1",
        "database.hostname":"192.168.99.100",
        "database.port":"3306",
        "database.user":"debezium",
        "database.password":"dbz",
        "database.server.id":"184054",
        "database.server.name":"dbserver1",
        "database.whitelist":"inventory",
        "database.history.kafka.bootstrap.servers":"kafka:9092",
        "database.history.kafka.topic":"dbhistory.inventory",
        "name":"inventory-connector"
     },
     "tasks":[]
  }
  ----
  
  This response describes the connector resource `/connectors/inventory-connector` that the service just created and includes the connector's configuration and information about the tasks. Since the connector was just created, the service hasn't yet finished starting tasks. 
  
  We can even use the RESTful API to verify that our connector is included in the list of connectors:
  
  [source,bash,indent=0]
  ----
      $ curl -H "Accept:application/json" 192.168.99.100:8083/connectors/
  ----
  
  which should return the following:
  
  [source,json,indent=0]
  ----
      ["inventory-connector"]
  ----
  
  Recall that the Kafka Connect service uses connectors to start one or more tasks that do the work, and that it will automatically distribute the running tasks across the cluster of Kafka Connect services. Should any of the services stop or crash, those tasks will be redistributed to running services. We can see the tasks when we get the state of the connector:
  
  [source,bash,indent=0]
  ----
      $ curl -i -X GET -H "Accept:application/json" 192.168.99.100:8083/connectors/inventory-connector
  ----
  
  which returns:
  
  [source,http,indent=0]
  ----
  HTTP/1.1 200 OK
  Date: Tue, 16 Aug 2016 01:13:45 GMT
  Content-Type: application/json
  Content-Length: 534
  Server: Jetty(9.2.15.v20160210)
  
  {
    "name": "inventory-connector",
    "config": {
      "name": "inventory-connector",
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "tasks.max": "1",
      "database.hostname": "192.168.99.100",
      "database.port": "3306"
      "database.user": "debezium",
      "database.password": "dbz",
      "database.server.id": "184054",
      "database.server.name": "mysql-server-1",
      "database.history.kafka.bootstrap.servers": "kafka:9092",
      "database.history.kafka.topic": "schema-changes.inventory",
      "database.whitelist": "inventory",
    },
    "tasks": [
      {
        "connector": "inventory-connector",
        "task": 0
      }
    ]
  }
  ----
  
  Here, we can see that the connector is running a single task (e.g., task 0) to do its work. The MySQL connector only supports a single task, since MySQL records all of its activities in one sequential binlog and so the MySQL connector needs only one reader to get a consistent and totally ordered view of all of those events. 
  
  If we look at the output of our `connect` container, we should now see log statements that show the progress of the snapshot:
  
  [listing,indent=0,options="nowrap"]
  ----
  ...
  2016-08-16 01:25:19,908 INFO   MySQL|dbserver1|snapshot  Snapshot is using user 'debezium' with these MySQL grants:   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,910 INFO   MySQL|dbserver1|snapshot  	GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%'   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,911 INFO   MySQL|dbserver1|snapshot  MySQL server variables related to change data capture:   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,915 INFO   MySQL|dbserver1|snapshot  	binlog_cache_size                             = 32768                                           [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,916 INFO   MySQL|dbserver1|snapshot  	binlog_checksum                               = CRC32                                           [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,916 INFO   MySQL|dbserver1|snapshot  	binlog_direct_non_transactional_updates       = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,917 INFO   MySQL|dbserver1|snapshot  	binlog_error_action                           = ABORT_SERVER                                    [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,917 INFO   MySQL|dbserver1|snapshot  	binlog_format                                 = ROW                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,918 INFO   MySQL|dbserver1|snapshot  	binlog_group_commit_sync_delay                = 0                                               [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,918 INFO   MySQL|dbserver1|snapshot  	binlog_group_commit_sync_no_delay_count       = 0                                               [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,919 INFO   MySQL|dbserver1|snapshot  	binlog_gtid_simple_recovery                   = ON                                              [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,920 INFO   MySQL|dbserver1|snapshot  	binlog_max_flush_queue_time                   = 0                                               [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,920 INFO   MySQL|dbserver1|snapshot  	binlog_order_commits                          = ON                                              [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,921 INFO   MySQL|dbserver1|snapshot  	binlog_row_image                              = FULL                                            [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,921 INFO   MySQL|dbserver1|snapshot  	binlog_rows_query_log_events                  = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,921 INFO   MySQL|dbserver1|snapshot  	binlog_stmt_cache_size                        = 32768                                           [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,922 INFO   MySQL|dbserver1|snapshot  	enforce_gtid_consistency                      = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,922 INFO   MySQL|dbserver1|snapshot  	gtid_executed_compression_period              = 1000                                            [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,923 INFO   MySQL|dbserver1|snapshot  	gtid_mode                                     = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,923 INFO   MySQL|dbserver1|snapshot  	gtid_next                                     = AUTOMATIC                                       [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,924 INFO   MySQL|dbserver1|snapshot  	gtid_owned                                    =                                                 [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,924 INFO   MySQL|dbserver1|snapshot  	gtid_purged                                   =                                                 [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,925 INFO   MySQL|dbserver1|snapshot  	innodb_api_enable_binlog                      = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,925 INFO   MySQL|dbserver1|snapshot  	innodb_locks_unsafe_for_binlog                = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,926 INFO   MySQL|dbserver1|snapshot  	innodb_version                                = 5.7.14                                          [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,927 INFO   MySQL|dbserver1|snapshot  	log_statements_unsafe_for_binlog              = ON                                              [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,927 INFO   MySQL|dbserver1|snapshot  	max_binlog_cache_size                         = 18446744073709547520                            [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,927 INFO   MySQL|dbserver1|snapshot  	max_binlog_size                               = 1073741824                                      [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,927 INFO   MySQL|dbserver1|snapshot  	max_binlog_stmt_cache_size                    = 18446744073709547520                            [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,928 INFO   MySQL|dbserver1|snapshot  	protocol_version                              = 10                                              [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,928 INFO   MySQL|dbserver1|snapshot  	session_track_gtids                           = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,928 INFO   MySQL|dbserver1|snapshot  	slave_type_conversions                        =                                                 [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,929 INFO   MySQL|dbserver1|snapshot  	sync_binlog                                   = 1                                               [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,930 INFO   MySQL|dbserver1|snapshot  	tls_version                                   = TLSv1,TLSv1.1                                   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,932 INFO   MySQL|dbserver1|snapshot  	tx_isolation                                  = REPEATABLE-READ                                 [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,937 INFO   MySQL|dbserver1|snapshot  	tx_read_only                                  = OFF                                             [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,937 INFO   MySQL|dbserver1|snapshot  	version                                       = 5.7.14-log                                      [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,938 INFO   MySQL|dbserver1|snapshot  	version_comment                               = MySQL Community Server (GPL)                    [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,938 INFO   MySQL|dbserver1|snapshot  	version_compile_machine                       = x86_64                                          [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,939 INFO   MySQL|dbserver1|snapshot  	version_compile_os                            = Linux                                           [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,939 INFO   MySQL|dbserver1|snapshot  Step 0: disabling autocommit and enabling repeatable read transactions   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,941 INFO   MySQL|dbserver1|snapshot  Step 1: start transaction with consistent snapshot   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,943 INFO   MySQL|dbserver1|snapshot  Step 2: flush and obtain global read lock (preventing writes to database)   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,944 INFO   MySQL|dbserver1|snapshot  Step 3: read binlog position of MySQL master   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,946 INFO   MySQL|dbserver1|snapshot  	 using binlog 'mysql-bin.000003' at position '154' and gtid ''   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,946 INFO   MySQL|dbserver1|snapshot  Step 4: read list of available databases   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,948 INFO   MySQL|dbserver1|snapshot  	 list of available databases is: [information_schema, inventory, mysql, performance_schema, sys]   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,949 INFO   MySQL|dbserver1|snapshot  Step 5: read list of available tables in each database   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,952 INFO   MySQL|dbserver1|snapshot  	 'information_schema.CHARACTER_SETS' is filtered out, discarding   [io.debezium.connector.mysql.SnapshotReader]
  ...
  2016-08-16 01:25:19,988 INFO   MySQL|dbserver1|snapshot  	 'information_schema.INNODB_SYS_TABLESTATS' is filtered out, discarding   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,990 INFO   MySQL|dbserver1|snapshot  	 including 'inventory.customers'   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,990 INFO   MySQL|dbserver1|snapshot  	 including 'inventory.orders'   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,991 INFO   MySQL|dbserver1|snapshot  	 including 'inventory.products'   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,991 INFO   MySQL|dbserver1|snapshot  	 including 'inventory.products_on_hand'   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:19,993 INFO   MySQL|dbserver1|snapshot  	 'mysql.columns_priv' is filtered out, discarding   [io.debezium.connector.mysql.SnapshotReader]
  ...
  2016-08-16 01:25:20,133 INFO   MySQL|dbserver1|snapshot  	 'sys.x$waits_global_by_latency' is filtered out, discarding   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,134 INFO   MySQL|dbserver1|snapshot  Step 6: generating DROP and CREATE statements to reflect current database schemas:   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,158 INFO   MySQL|dbserver1|snapshot  	DROP TABLE IF EXISTS inventory.products_on_hand   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,190 INFO   MySQL|dbserver1|snapshot  	DROP TABLE IF EXISTS inventory.customers   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,193 INFO   MySQL|dbserver1|snapshot  	DROP TABLE IF EXISTS inventory.orders   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,195 INFO   MySQL|dbserver1|snapshot  	DROP TABLE IF EXISTS inventory.products   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,202 INFO   MySQL|dbserver1|snapshot  	DROP DATABASE IF EXISTS inventory   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,206 INFO   MySQL|dbserver1|snapshot  	CREATE DATABASE inventory   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,209 INFO   MySQL|dbserver1|snapshot  	USE inventory   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,223 INFO   MySQL|dbserver1|snapshot  	CREATE TABLE `customers` (
    `id` int(11) NOT NULL AUTO_INCREMENT,
    `first_name` varchar(255) NOT NULL,
    `last_name` varchar(255) NOT NULL,
    `email` varchar(255) NOT NULL,
    PRIMARY KEY (`id`),
    UNIQUE KEY `email` (`email`)
  ) ENGINE=InnoDB AUTO_INCREMENT=1005 DEFAULT CHARSET=latin1   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,250 INFO   MySQL|dbserver1|snapshot  	CREATE TABLE `orders` (
    `order_number` int(11) NOT NULL AUTO_INCREMENT,
    `order_date` date NOT NULL,
    `purchaser` int(11) NOT NULL,
    `quantity` int(11) NOT NULL,
    `product_id` int(11) NOT NULL,
    PRIMARY KEY (`order_number`),
    KEY `order_customer` (`purchaser`),
    KEY `ordered_product` (`product_id`),
    CONSTRAINT `orders_ibfk_1` FOREIGN KEY (`purchaser`) REFERENCES `customers` (`id`),
    CONSTRAINT `orders_ibfk_2` FOREIGN KEY (`product_id`) REFERENCES `products` (`id`)
  ) ENGINE=InnoDB AUTO_INCREMENT=10005 DEFAULT CHARSET=latin1   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,265 INFO   MySQL|dbserver1|snapshot  	CREATE TABLE `products` (
    `id` int(11) NOT NULL AUTO_INCREMENT,
    `name` varchar(255) NOT NULL,
    `description` varchar(512) DEFAULT NULL,
    `weight` float DEFAULT NULL,
    PRIMARY KEY (`id`)
  ) ENGINE=InnoDB AUTO_INCREMENT=110 DEFAULT CHARSET=latin1   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,271 INFO   MySQL|dbserver1|snapshot  	CREATE TABLE `products_on_hand` (
    `product_id` int(11) NOT NULL,
    `quantity` int(11) NOT NULL,
    PRIMARY KEY (`product_id`),
    CONSTRAINT `products_on_hand_ibfk_1` FOREIGN KEY (`product_id`) REFERENCES `products` (`id`)
  ) ENGINE=InnoDB DEFAULT CHARSET=latin1   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,283 INFO   MySQL|dbserver1|snapshot  Step 7: releasing global read lock to enable MySQL writes   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,285 INFO   MySQL|dbserver1|snapshot  Step 7: blocked writes to MySQL for a total of 00:00:00.341   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,287 INFO   MySQL|dbserver1|snapshot  Step 8: scanning contents of 4 tables   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,289 INFO   MySQL|dbserver1|snapshot  Step 8: - scanning table 'inventory.customers' (1 of 4 tables)   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,293 INFO   MySQL|dbserver1|snapshot  Step 8: - 4 of 4 rows scanned from table 'inventory.customers' after 00:00:00.004   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,294 INFO   MySQL|dbserver1|snapshot  Step 8: - scanning table 'inventory.orders' (2 of 4 tables)   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,302 INFO   MySQL|dbserver1|snapshot  Step 8: - 4 of 4 rows scanned from table 'inventory.orders' after 00:00:00.008   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,304 INFO   MySQL|dbserver1|snapshot  Step 8: - scanning table 'inventory.products' (3 of 4 tables)   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,306 INFO   MySQL|dbserver1|snapshot  Step 8: - 9 of 9 rows scanned from table 'inventory.products' after 00:00:00.001   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,306 INFO   MySQL|dbserver1|snapshot  Step 8: - scanning table 'inventory.products_on_hand' (4 of 4 tables)   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,308 INFO   MySQL|dbserver1|snapshot  Step 8: - 9 of 9 rows scanned from table 'inventory.products_on_hand' after 00:00:00.002   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,309 INFO   MySQL|dbserver1|snapshot  Step 8: scanned 26 rows in 4 tables in 00:00:00.021   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,309 INFO   MySQL|dbserver1|snapshot  Step 9: committing transaction   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,311 INFO   MySQL|dbserver1|snapshot  Completed snapshot in 00:00:00.713   [io.debezium.connector.mysql.SnapshotReader]
  2016-08-16 01:25:20,620 WARN   ||  Error while fetching metadata with correlation id 0 : {dbserver1=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
  2016-08-16 01:25:20,855 WARN   ||  Error while fetching metadata with correlation id 3 : {dbserver1.inventory.customers=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
  2016-08-16 01:25:21,076 WARN   ||  Error while fetching metadata with correlation id 8 : {dbserver1.inventory.orders=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
  2016-08-16 01:25:21,310 WARN   ||  Error while fetching metadata with correlation id 11 : {dbserver1.inventory.products=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
  2016-08-16 01:25:21,548 WARN   ||  Error while fetching metadata with correlation id 15 : {dbserver1.inventory.products_on_hand=LEADER_NOT_AVAILABLE}   [org.apache.kafka.clients.NetworkClient]
  2016-08-16 01:25:21,754 INFO   MySQL|dbserver1|binlog  Connected to MySQL binlog at 192.168.99.100:3306, starting at binlog file 'mysql-bin.000003', pos=154, row=0   [io.debezium.connector.mysql.BinlogReader]
  2016-08-16 01:26:19,201 INFO   ||  Finished WorkerSourceTask{id=inventory-connector-0} commitOffsets successfully in 10 ms   [org.apache.kafka.connect.runtime.WorkerSourceTask]
  ...
  ----
  
  Let's look into this output in more detail. First, Debezium log output makes use of _mapped diagnostic contexts_, or MDC, which allow the log messages to include thread-specific information like the connector type (e.g., `MySQL` in the above log messages after "INFO" or "WARN" fields), the logical name of the connector (e.g., `dbserver1` above), and the connector's activity (e.g., `snapshot` and `binlog`). Hopefully these will make it easier to understand what is going on in the multi-threaded Kafka Connect service.
  
  Now, if we look at these log statements, we can see that the connector starts, describes information about the MySQL server and the user the connector is using, performs a consistent snapshot with 9 steps, and then starts reading the binlog at the same point where the snapshot was taken. Since our `inventory` database is quite small, the snapshot process goes quite quickly: 0.713 seconds as shown in one of the log messages above, and actually much of that was spent writing out the many log lines that make it easy to track the progress of the connector. This will take longer with larger databases, but the log messages do describe which of the 9 steps are performed with a global read lock on the MySQL server. (See the link:/docs/connectors/mysql[MySQL connector documentation] for more details.)
  
  [NOTE]
  ====
  Debezium 0.3 uses result set streaming during the snapshot process to page through the results. This makes it possible to snapshot very large databases without memory problems.
  ====
  
  After the snapshot completes, the MySQL connector will generally output very little information using `INFO` or `WARN` level messages. 
  
  There's one more thing in these log messages to mention. The five warning log messages near the end of the sample output above sound ominous, but are basically telling us that new Kafka topics were created and Kafka had to assign a new leader. Note the names of the topics:
  
  * `dbserver1.inventory.products`
  * `dbserver1.inventory.products_on_hand`
  * `dbserver1.inventory.customers`
  * `dbserver1.inventory.orders`
  
  As described in the link:/docs/connectors/mysql/#topic-names[MySQL connector documentation], each topic names start with `dbserver1`, which is the logical name we gave our connector. Each topic name also includes `inventory`, which is the name of the database. Finally, each topic name concludes with the name of one of the tables in the `inventory` database. In other words, all of the data change events describing rows in the each table appear in separate topics.
  
  Let's look at all of the data change events in the `dbserver1.inventory.customers` topic. Again, we'll use the `debezium/kafka` Docker image to start a new container that connects to Kafka to watch the topic from the beginning of the topic:
  
  [source,bash,indent=0]
  ----
      $ docker run -it --name watcher --rm --link zookeeper:zookeeper debezium/kafka:0.3 watch-topic -a -k dbserver1.inventory.customers
  ----
  
  Again, we use the `--rm` flag since we want the container to be removed when it stops, and we use the `-a` flag on `watch-topic` to signal that we want to see _all_ events since the beginning of the topic. (If we were to remove the `-a` flag, we'd see only the events that are recorded in the topic _after_ we start watching.) The `-k` flag specifies that the output should include the event's key, which in our case contains the row's primary key. Here's the output:
  
  [source,bash,indent=0]
  ----
      ...
  Contents of topic dbserver1.inventory.customers:
  {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1001}}	{"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":1001,"first_name":"Sally","last_name":"Thomas","email":"sally.thomas@acme.com"},"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}}
  {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1002}}	{"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":1002,"first_name":"George","last_name":"Bailey","email":"gbailey@foobar.com"},"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}}
  {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1003}}	{"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":1003,"first_name":"Edward","last_name":"Walker","email":"ed@walker.com"},"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}}
  {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1004}}	{"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":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}}
  ----
  
  [NOTE]
  