Outdated binlog references used when a connector is restarted before snapshotting is completed [DBZ-3803]

Migrated from [DBZ-3803](https://issues.redhat.com/browse/DBZ-3803)

I've come across an issue with restarting a mysql connector task where I am wondering if it is a bug or a feature. The issue is as follows:

*T0:* Start a new connector for a mysql database with snapshot mode "when_needed"

*T1:* Connector records the mysql binlog position *B1* before beginning snapshotting

*T2:* Connector begins snapshotting schemas and data at schema version *V1*

*T3:* Database row is updated and produces a binlog event *E1* with the schema *V1* as it was at *T2*

*T4:* Mysql schema changes during snapshotting resulting in schema version *V2*

*T5:* Connector task fails before completing snapshotting

*T6:* Connector task is restarted via the REST API

*T7:* Restarted task finds binlog position *B1* previously recorded at *T1* and uses it

*T8:* Snapshotting of schema and data starts over and finishes successfully recording schema version *V2*

*T9:* Connector starts processing binlog events starting at binlog position *B1* established at *T1*

*T10:* Connector encounters *E1* on the binlog where the schema of the record is from schema version *V1* does not match the expected version *V2*


*T11:* Connector fails with an exception like below



I wonder if at T7 when the task was restarted resulting in a restart of snapshotting if the connector should record a more current binlog reference at *T7* rather than use the old binlog reference *B1* from the initial connector start at *T1*?

{noformat}

[2021-06-25 03:58:31,532] ERROR [prod573-debezium-20200623|task-0] Producer failure (io.debezium.pipeline.ErrorHandler)
io.debezium.DebeziumException: Error processing binlog event
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleEvent(MySqlStreamingChangeEventSource.java:369)
at com.github.shyiko.mysql.binlog.BinaryLogClient.notifyEventListeners(BinaryLogClient.java:1118)
at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:966)
at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606)
at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.kafka.connect.errors.ConnectException: Error while processing event at offset {transaction_id=null, ts_sec=1624483800, file=binlog.000045, pos=965901764, row=1, server_id=174673616, event=26}
at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:232)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.lambda$handleInsert$3(MySqlStreamingChangeEventSource.java:681)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleChange(MySqlStreamingChangeEventSource.java:730)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleInsert(MySqlStreamingChangeEventSource.java:680)
at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.handleEvent(MySqlStreamingChangeEventSource.java:352)
... 5 more
Caused by: org.apache.kafka.connect.errors.ConnectException: Data row is smaller than a column index, internal schema representation is probably out of sync with real database schema
at io.debezium.relational.TableSchemaBuilder.validateIncomingRowToInternalMetadata(TableSchemaBuilder.java:221)
at io.debezium.relational.TableSchemaBuilder.lambda$createValueGenerator$5(TableSchemaBuilder.java:250)
at io.debezium.relational.TableSchema.valueFromColumnData(TableSchema.java:141)
at io.debezium.relational.RelationalChangeRecordEmitter.emitCreateRecord(RelationalChangeRecordEmitter.java:69)
at io.debezium.relational.RelationalChangeRecordEmitter.emitChangeRecords(RelationalChangeRecordEmitter.java:45)
at io.debezium.pipeline.EventDispatcher.dispatchDataChangeEvent(EventDispatcher.java:202)
... 9 more{noformat}
[https://gitter.im/debezium/user?at=60d5f66a8a40b117282a4cbd|http://example.com]

Start by tracing restart and snapshot position handling in the MySQL connector, then read MySqlStreamingChangeEventSource.java and TableSchemaBuilder.java around the reported failure. Reproduce the documented when_needed sequence if possible, focusing on the REST-triggered restart and reuse of binlog position B1. Done means the restart no longer causes the stale schema/binlog mismatch, with regression coverage for the scenario.

Written by the indexing model from the issue text.