Mysql connector does not restart after a database outage [DBZ-5707]

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

h1. Bug report
h2. What Debezium connector do you use and what version?

io.debezium.connector.mysql.MySqlConnector

debezium/connect:1.9.5.Final
h2. What is the connector configuration?
{code:java}
curl -v --request PUT --header 'Accept:application/json' --header 'Content-Type:application/json' 'http://localhost:8083/connectors/my-mysql-connector/config' --data '{
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.ssl.mode": "required",
"database.serverTimezone": "UTC",
"database.useUnicode": "true",
"database.characterEncoding": "UTF-8",
"database.hostname": "mydb-db",
"database.port": "3306",
"database.user": "root",
"database.password": "password",
"database.server.name": "mydb",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "debezium-dbhistory",
"internal.custom.retriable.exception" : "(.*)",
"table.include.list": "mydb.outboxevent",
"database.whitelist": "",
"include.schema.changes": "false",
"tombstones.on.delete": "false",
"transforms": "outbox",
"transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
"transforms.outbox.table.fields.additional.placement": "aggregateid:header:aggregateId,tenant_id:header:tenant,type:header:eventType",
"transforms.outbox.table.expand.json.payload": "true",
"database.history.skip.unparseable.ddl": "true"
}' {code}
{color:#172b4d}Note: internal.custom.retriable.exception is configured!{color}
h2. What is the captured database version and mode of deployment?

Mysql 5.7.*

Running in docker
h2. What behaviour do you expect?

I would expect that Debezium keeps retrying until the database comes back up.
h2. What behaviour do you see?

I can see that the exception is retried once when I stop the mysql database:

 
{code:java}
Oct 10, 2022 7:54:18 AM com.github.shyiko.mysql.binlog.BinaryLogClient connect
INFO: Connected to mydb-db:3306 at bin-log.000009/5539027 (sid:6048, cid:74)
2022-10-10 07:54:18,997 INFO   MySQL|mydb|binlog  Connected to MySQL binlog at mydb-db:3306, starting at MySqlOffsetContext [sourceInfoSchema=Schema{io.debezium.connector.mysql.Source:STRUCT}, sourceInfo=SourceInfo [currentGtid=null, currentBinlogFilename=bin-log.000009, currentBinlogPosition=5539027, currentRowNumber=0, serverId=0, sourceTime=null, threadId=-1, currentQuery=null, tableIds=[], databaseName=null], snapshotCompleted=false, transactionContext=TransactionContext [currentTransactionId=null, perTableEventCount={}, totalEventCount=0], restartGtidSet=null, currentGtidSet=null, restartBinlogFilename=bin-log.000009, restartBinlogPosition=5539027, restartRowsToSkip=0, restartEventsToSkip=0, currentEventLengthInBytes=0, inTransaction=false, transactionId=null, incrementalSnapshotContext =IncrementalSnapshotContext [windowOpened=false, chunkEndPosition=null, dataCollectionsToSnapshot=[], lastEventKeySent=null, maximumKey=null]]   [io.debezium.connector.mysql.MySqlStreamingChangeEventSource]
2022-10-10 07:54:19,000 INFO   MySQL|mydb|streaming  Waiting for keepalive thread to start   [io.debezium.connector.mysql.MySqlStreamingChangeEventSource]
2022-10-10 07:54:19,003 INFO   MySQL|mydb|binlog  Creating thread debezium-mysqlconnector-mydb-binlog-client   [io.debezium.util.Threads]
2022-10-10 07:54:19,105 INFO   MySQL|mydb|streaming  Keepalive thread is running   [io.debezium.connector.mysql.MySqlStreamingChangeEventSource]
2022-10-10 07:54:51,824 ERROR  MySQL|mydb|binlog  Error during binlog processing. Last offset stored = {transaction_id=null, file=bin-log.000010, pos=154, server_id=1, event=1}, binlog reader near position = bin-log.000010/154   [io.debezium.connector.mysql.MySqlStreamingChangeEventSource]
2022-10-10 07:54:51,833 ERROR  MySQL|mydb|binlog  Producer failure   [io.debezium.pipeline.ErrorHandler]
io.debezium.DebeziumException: Failed to read next byte from position 693
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1189)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1234)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599)
    at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.io.EOFException: Failed to read next byte from position 693
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:213)
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readInteger(ByteArrayInputStream.java:52)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:35)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:27)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:221)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$1.nextEvent(MySqlStreamingChangeEventSource.java:230)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:952)
    ... 3 more
2022-10-10 07:54:51,849 INFO   MySQL|mydb|binlog  Stopped reading binlog after 0 events, last recorded offset: {transaction_id=null, file=bin-log.000010, pos=154, server_id=1, event=1}   [io.debezium.connector.mysql.MySqlStreamingChangeEventSource]
2022-10-10 07:54:52,120 WARN   ||  Going to restart connector after 10 sec. after a retriable exception   [io.debezium.connector.common.BaseSourceTask]
2022-10-10 07:54:52,125 INFO   MySQL|mydb|streaming  Finished streaming   [io.debezium.pipeline.ChangeEventSourceCoordinator]
2022-10-10 07:54:52,142 INFO   ||  Connection gracefully closed   [io.debezium.jdbc.JdbcConnection]
2022-10-10 07:54:52,146 INFO   ||  [Producer clientId=mydb-dbhistory] Closing the Kafka producer with timeoutMillis = 30000 ms.   [org.apache.kafka.clients.producer.KafkaProducer]
2022-10-10 07:54:52,161 INFO   ||  Metrics scheduler closed   [org.apache.kafka.common.metrics.Metrics]
2022-10-10 07:54:52,162 INFO   ||  Closing reporter org.apache.kafka.common.metrics.JmxReporter   [org.apache.kafka.common.metrics.Metrics]
2022-10-10 07:54:52,162 INFO   ||  Metrics reporters closed   [org.apache.kafka.common.metrics.Metrics]
2022-10-10 07:54:52,166 INFO   ||  App info kafka.producer for mydb-dbhistory unregistered   [org.apache.kafka.common.utils.AppInfoParser]
2022-10-10 07:54:52,170 WARN   ||  WorkerSourceTask{id=mydb-mysql-connector-0} failed to poll records from SourceTask. Will retry operation.   [org.apache.kafka.connect.runtime.WorkerSourceTask]
org.apache.kafka.connect.errors.RetriableException: An exception occurred in the change event producer. This connector will be restarted.
    at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:46)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1234)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:980)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:599)
    at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:857)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: io.debezium.DebeziumException: Failed to read next byte from position 693
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1189)
    ... 5 more
Caused by: java.io.EOFException: Failed to read next byte from position 693
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.read(ByteArrayInputStream.java:213)
    at com.github.shyiko.mysql.binlog.io.ByteArrayInputStream.readInteger(ByteArrayInputStream.java:52)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:35)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventHeaderV4Deserializer.deserialize(EventHeaderV4Deserializer.java:27)
    at com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer.nextEvent(EventDeserializer.java:221)
    at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$1.nextEvent(MySqlStreamingChangeEventSource.java:230)
    at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:952)
    ... 3 more
2022-10-10 07:54:52,174 INFO   ||  Awaiting end of restart backoff period after a retriable error   [io.debezium.connector.common.BaseSourceTask]
2022-10-10 07:54:54,176 INFO   ||  Awaiting end of restart backoff period after a retriable error   [io.debezium.connector.common.BaseSourceTask]
2022-10-10 07:54:56,185 INFO   ||  Awaiting end of restart backoff period after a retriable error   [io.debezium.connector.common.BaseSourceTask] {code}
After 10 seconds it tries to restart the connector, but it immediately stops the connector after restarting the connector:

 
{code:java}
ERROR || WorkerSourceTask{id=my-mysql-connector-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted [org.apache.kafka.connect.runtime.WorkerTask]
io.debezium.DebeziumException: Unexpected error while connecting to MySQL and looking at BINLOG_FORMAT mode:{code}
The database is still down, but I would expect that it keeps retrying for a while at least.

 
h2. Do you see the same behaviour using the latest released Debezium version?

Also tested and seen this behaviour in version 2.0.0 final
h2. How to reproduce the issue using our [tutorial|https://github.com/debezium/debezium-examples/tree/main/tutorial] deployment?
# Configure mysql connector
# Make sure that it connects successfully to the bin log
# Stop mysql server, wait for 10 seconds, start mysql server again and observe the logs

Start with MySqlStreamingChangeEventSource and BaseSourceTask, using the reported restart logs to trace what happens when MySQL remains unavailable after the retry backoff. Reproduce the outage scenario from the issue with the Debezium tutorial deployment and verify that the connector continues retrying until the database returns, rather than requiring a manual restart.

Written by the indexing model from the issue text.