[Bug] Unsupported table metadata field type 216 when parsing binlog event

[leonardBang]

Search before asking

• I searched in the issues and found nothing similar.

Flink version

1.15.2

Flink CDC version

2.3.0

Database and its version

mysql version = 8.0.18

Minimal reproduce step

This error occurs when parsing the binlog.

What did you expect to see?

Support this kind of even type.

What did you see instead?

Caused by: io.debezium.DebeziumException: Failed to deserialize data of EventHeaderV4{timestamp=1679640836000, eventType=TABLE_MAP, serverId=1704062425, headerLength=19, dataLength=89, nextPosition=359034816, flags=0}
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1154)
... 5 more
Caused by: com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializationException: Failed to deserialize data of EventHeaderV4{timestamp=1679640836000, eventType=TABLE_MAP, serverId=1704062425, headerLength=19, dataLength=89, nextPosition=359034816, flags=0}
at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:309)
at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeTableMapEventData(EventDeserializer.java:281)
at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:228)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$1.nextEvent(MySqlStreamingChangeEventSource.java:230)
at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:952)
... 3 more
Caused by: java.io.IOException: Unsupported table metadata field type 216
at com.github.shyiko.mysql.binlog.event.deserialization.TableMapEventMetadataDeserializer.deserialize(TableMapEventMetadataDeserializer.java:52)
at com.github.shyiko.mysql.binlog.event.deserialization.TableMapEventDataDeserializer.deserialize(TableMapEventDataDeserializer.java:59)
at com.github.shyiko.mysql.binlog.event.deserialization.TableMapEventDataDeserializer.deserialize(TableMapEventDataDeserializer.java:37)
at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.deserializeEventData(EventDeserializer.java:303)
... 7 more

#2001 (comment)

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[ruanhang1993]

I add a test, but cannot reproduce the error.

MySQL : docker mysql:8.0
Table sql :

CREATE TABLE multi_value_index (
    id bigint NOT NULL,
    f1 json,
    PRIMARY KEY (id)
);

insert into multi_value_index(id, f1)
values (1, cast('{"inner":[1,3]}' as json)),
       (2, cast('{"inner":[1,3,9,8]}' as json));

Test:

@Test
    public void testMultiValueIndex() throws Throwable {
        fullTypesMySql8Database.createAndInitialize();
        String sourceDDL =
                String.format(
                        "CREATE TABLE multi_value_index (\n"
                                + "    `id` INT NOT NULL,\n"
                                + "    f1 STRING,\n"
                                + "    primary key (`id`) not enforced"
                                + ") WITH ("
                                + " 'connector' = 'mysql-cdc',"
                                + " 'hostname' = '%s',"
                                + " 'port' = '%s',"
                                + " 'username' = '%s',"
                                + " 'password' = '%s',"
                                + " 'database-name' = '%s',"
                                + " 'table-name' = '%s',"
                                + " 'scan.incremental.snapshot.enabled' = '%s',"
                                + " 'server-id' = '%s',"
                                + " 'server-time-zone' = 'UTC',"
                                + " 'scan.incremental.snapshot.chunk.size' = '%s'"
                                + ")",
                        MYSQL8_CONTAINER.getHost(),
                        MYSQL8_CONTAINER.getDatabasePort(),
                        fullTypesMySql8Database.getUsername(),
                        fullTypesMySql8Database.getPassword(),
                        fullTypesMySql8Database.getDatabaseName(),
                        "multi_value_index",
                        incrementalSnapshot,
                        getServerId(),
                        getSplitSize());
        tEnv.executeSql(sourceDDL);

        // async submit job
        TableResult result = tEnv.executeSql("SELECT id, f1 FROM multi_value_index");

        CloseableIterator<Row> iterator = result.collect();
        waitForSnapshotStarted(iterator);
        Thread.sleep(10000L);

        try (Connection connection = fullTypesMySql8Database.getJdbcConnection();
             Statement statement = connection.createStatement()) {
            statement.execute(
                    "ALTER TABLE `multi_value_index` ADD INDEX `k_source`  ( (CAST(f1->'$.inner' AS UNSIGNED ARRAY)) );");
            statement.execute(
                    "insert into multi_value_index(id, f1) values (3, cast('{\"inner\":[1,4,5,8]}' as json));");
        }

        while (true) {

        }
//        assertEqualsInAnyOrder(Arrays.asList(), fetchRows(iterator, 2));
//        result.getJobClient().get().cancel().get();
    }

[ruanhang1993]

Debug

The value is:

{"source":{"file":"mysql-bin.000005","pos":1001,"server_id":223344},"position":{"transaction_id":null,"ts_sec":1686219039,"file":"mysql-bin.000005","pos":1240,"gtids":"9444d53c-05e4-11ee-b649-0242ac110004:1-17","server_id":223344},"databaseName":"column_type_test_mysql8_1gppbph","ddl":"ALTER TABLE `multi_value_index` ADD INDEX `k_source`  ( (CAST(f1->'$.inner' AS UNSIGNED ARRAY)) )","tableChanges":[]}

[racoon945]

I add a test, but cannot reproduce the error.

MySQL : docker mysql:8.0 Table sql :

CREATE TABLE multi_value_index (
    id bigint NOT NULL,
    f1 json,
    PRIMARY KEY (id)
);

insert into multi_value_index(id, f1)
values (1, cast('{"inner":[1,3]}' as json)),
       (2, cast('{"inner":[1,3,9,8]}' as json));

Test:

@Test
    public void testMultiValueIndex() throws Throwable {
        fullTypesMySql8Database.createAndInitialize();
        String sourceDDL =
                String.format(
                        "CREATE TABLE multi_value_index (\n"
                                + "    `id` INT NOT NULL,\n"
                                + "    f1 STRING,\n"
                                + "    primary key (`id`) not enforced"
                                + ") WITH ("
                                + " 'connector' = 'mysql-cdc',"
                                + " 'hostname' = '%s',"
                                + " 'port' = '%s',"
                                + " 'username' = '%s',"
                                + " 'password' = '%s',"
                                + " 'database-name' = '%s',"
                                + " 'table-name' = '%s',"
                                + " 'scan.incremental.snapshot.enabled' = '%s',"
                                + " 'server-id' = '%s',"
                                + " 'server-time-zone' = 'UTC',"
                                + " 'scan.incremental.snapshot.chunk.size' = '%s'"
                                + ")",
                        MYSQL8_CONTAINER.getHost(),
                        MYSQL8_CONTAINER.getDatabasePort(),
                        fullTypesMySql8Database.getUsername(),
                        fullTypesMySql8Database.getPassword(),
                        fullTypesMySql8Database.getDatabaseName(),
                        "multi_value_index",
                        incrementalSnapshot,
                        getServerId(),
                        getSplitSize());
        tEnv.executeSql(sourceDDL);

        // async submit job
        TableResult result = tEnv.executeSql("SELECT id, f1 FROM multi_value_index");

        CloseableIterator<Row> iterator = result.collect();
        waitForSnapshotStarted(iterator);
        Thread.sleep(10000L);

        try (Connection connection = fullTypesMySql8Database.getJdbcConnection();
             Statement statement = connection.createStatement()) {
            statement.execute(
                    "ALTER TABLE `multi_value_index` ADD INDEX `k_source`  ( (CAST(f1->'$.inner' AS UNSIGNED ARRAY)) );");
            statement.execute(
                    "insert into multi_value_index(id, f1) values (3, cast('{\"inner\":[1,4,5,8]}' as json));");
        }

        while (true) {

        }
//        assertEqualsInAnyOrder(Arrays.asList(), fetchRows(iterator, 2));
//        result.getJobClient().get().cancel().get();
    }

I noticed that you are using docker mysql:8.0
Can you try with mysql:8.0.18 ?
docker pull mysql:8.0.18
The problem occurred again recently, reproducible in MySQL 8.0.18 but not in 8.0.22-13.

[ruanhang1993]

Hi, @racoon945 .
I change the version to mysql:8.0.18, but the error does not occur.

[hanpi14]

I was using flink1.16.2 and mysql 5.6 for data capture, and such errors also occurred,cdc versions 2.4.0 and 2.4.1 have the same error

[ruanhang1993]

Hi, @hanpi14.
Which MySQL version are you using when the error occurs?
And do you know what sql will cause this error ? Thanks.

[ruanhang1993]

Close this issue as no reply received.

If the problem still exists, please re-create your issue in English on Apache Jira under project Flink with component tag Flink CDC. Thank you!