DBZ-60 Added MySQL server ID and timestamp to event's source info

[rhauch]

Added to the Debezium event message's source information the MySQL server ID for the cluster process that recorded the event and the MySQL timestamp at which the event was recorded.

The following is a sample of the new source struct in a message's envelope:

      "source" : {
        "name" : "kafka-connect",
        "server-id" : 112233,
        "ts" : 1463753313000,
        "file" : "mysql-bin.000003",
        "pos" : 1081,
        "row" : 0
      },

Note that the server-id and ts fields are both new, and the name field was previously called server and contains the connector's logical name of the server; the rest of the fields in source are unchanged, as are any other part of the event structure. The schema for the source struct is now:

   {
        "type" : "struct",
        "fields" : [ {
          "type" : "string",
          "optional" : false,
          "field" : "name"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "server-id"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "ts"
        }, {
          "type" : "string",
          "optional" : false,
          "field" : "file"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "pos"
        }, {
          "type" : "int32",
          "optional" : false,
          "field" : "row"
        } ],
        "optional" : false,
        "name" : "io.debezium.connector.mysql.Source",
        "field" : "source"
      }

And finally, the following shows the source struct and schema in the context of a DELETE event's message value:

{
    "schema" : {
      "type" : "struct",
      "fields" : [ {
        "type" : "struct",
        "fields" : [ {
          "type" : "int32",
          "optional" : false,
          "field" : "id"
        }, {
          "type" : "string",
          "optional" : false,
          "field" : "name"
        }, {
          "type" : "string",
          "optional" : true,
          "field" : "description"
        }, {
          "type" : "double",
          "optional" : true,
          "field" : "weight"
        } ],
        "optional" : true,
        "name" : "connector_test.products",
        "field" : "before"
      }, {
        "type" : "struct",
        "fields" : [ {
          "type" : "int32",
          "optional" : false,
          "field" : "id"
        }, {
          "type" : "string",
          "optional" : false,
          "field" : "name"
        }, {
          "type" : "string",
          "optional" : true,
          "field" : "description"
        }, {
          "type" : "double",
          "optional" : true,
          "field" : "weight"
        } ],
        "optional" : true,
        "name" : "connector_test.products",
        "field" : "after"
      }, {
        "type" : "struct",
        "fields" : [ {
          "type" : "string",
          "optional" : false,
          "field" : "name"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "server-id"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "ts"
        }, {
          "type" : "string",
          "optional" : false,
          "field" : "file"
        }, {
          "type" : "int64",
          "optional" : false,
          "field" : "pos"
        }, {
          "type" : "int32",
          "optional" : false,
          "field" : "row"
        } ],
        "optional" : false,
        "name" : "io.debezium.connector.mysql.Source",
        "field" : "source"
      }, {
        "type" : "string",
        "optional" : false,
        "field" : "op"
      }, {
        "type" : "int64",
        "optional" : true,
        "field" : "ts"
      } ],
      "optional" : false,
      "name" : "kafka-connect.connector_test.products",
      "version" : 1
    },
    "payload" : {
      "before" : {
        "id" : 1001,
        "name" : "roy",
        "description" : "old robot",
        "weight" : 1234.56005859375
      },
      "after" : null,
      "source" : {
        "name" : "kafka-connect",
        "server-id" : 112233,
        "ts" : 1463753313000,
        "file" : "mysql-bin.000003",
        "pos" : 1081,
        "row" : 0
      },
      "op" : "d",
      "ts" : 1463753314594
    }
}

[criccomini]

TODO clean this up? :)

[rhauch]

I've intentionally left these as I often want to enable printing things when running the tests, and it's far easier to uncomment these than repeatedly figure out the best place to put them.

[criccomini]

LGTM one nit on some commented out test code