TINYINT debezium/connect issue.

test-case.md

If I create a table:

CREATE TABLE test.Test (
    id int not null primary key auto_increment,
    a TINYINT(1) not null,
    b TINYINT(1),
    c TINYINT(2) not null,
    d TINYINT(2)
);

Insert a couple of rows:

INSERT INTO test.Test VALUES (NULL, 5, 5, 5, 5), (NULL, 6, NULL, 6, NULL);

I end up with the following in MySQL:

id	a	b	c	d
1	5	5	5	5
2	6	NULL	6	NULL

Then I start a debezium connector, it takes a snapshot, and gets the following into kafka:

Schema:

"fields": [
  { "name": "id", "type": "int" },
  { "name": "a", "type": { "type": "int", "connect.type": "int16" } },
  { "name": "b", "type": [ "null", { "type": "int", "connect.type": "int16" } ], "default": null },
  { "name": "c", "type": { "type": "int", "connect.type": "int16" } },
  { "name": "d", "type": [ "null", { "type": "int", "connect.type": "int16" } ], "default": null }
]

Messages:

{ "id": 1, "a": 1, "b": { "int": 1 }, "c": 5, "d": { "int": 5 } }
{ "id": 2, "a": 1, "b": null, "c": 6, "d": null }

As you can see, the "a" columns are set (incorrectly) to 1, but the "c" fields are set correctly even though they are the exact same type in the kafka schema. It seems to treat the TINYINT(2) fine, but from my testing, only allow the TINYINT(1) be null, int16(0) or int16(1).

Now that debezium/connect is running, I insert another row:

INSERT INTO test.Test VALUES (NULL, 7, 7, 7, 7);

So the database now looks like:

id	a	b	c	d
1	5	5	5	5
2	6	NULL	6	NULL
3	7	7	7	7

And get a new message on the topic:

{ "id": 3, "a": 7, "b": { "int": 7 }, "c": 7, "d": { "int": 7 } }

And here, after the snapshot, the column values are interpreted correctly.