实现指定timestamp消费binlog

[hiliuxg]

背景

业务有需求，从指定timestamp去消费binlog，类似flink cdc 1.0 的'scan.startup.mode'='timestamp'

初步实现思路

1. 实现一个EarliestMySqlBinlogSplitAssigner 继承自MySqlSplitAssigner
2. 在getNext方法里面返回MySqlBinlogSplit，然后MySqlBinlogSplit中的binlogoffset指定从BinlogOffset.INITIAL_OFFSET开始
3. 实现一个DebeziumDeserializationSchema，序列化所有SourceRecord，然后判断时间是否符合指定的timestamp之后，如果是，则放到下游处理

出现问题

1. 发现能满足预期，从指定的timestamp开始处理
2. 但是发现从checkpoint启动或者savepoint启动，都报错

2022-04-29 18:17:47,080 ERROR io.debezium.connector.mysql.MySqlStreamingChangeEventSource  [] - Error during binlog processing. Last offset stored = null, binlog reader near position = /4
2022-04-29 18:17:47,080 ERROR io.debezium.pipeline.ErrorHandler                            [] - Producer failure
io.debezium.DebeziumException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires. Error code: 1236; SQLSTATE: HY000.
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1146) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185) [flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) [flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) [flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) [flink-connector-mysql-cdc-2.2.0.jar:?]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_252]
Caused by: com.github.shyiko.mysql.binlog.network.ServerException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:937) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	... 3 more
2022-04-29 18:17:47,085 INFO  io.debezium.connector.mysql.MySqlStreamingChangeEventSource  [] - Stopped reading binlog after 0 events, no new offset was recorded
2022-04-29 18:17:47,525 ERROR org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager [] - Received uncaught exception.
java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:148) ~[flink-connector-kafka_2.11-1.13.5.jar:?]
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:103) [flink-connector-kafka_2.11-1.13.5.jar:?]
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_252]
	at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_252]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_252]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_252]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_252]
Caused by: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
	at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.ververica.cdc.connectors.mysql.debezium.task.context.MySqlErrorHandler.setProducerThrowable(MySqlErrorHandler.java:72) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	... 1 more
Caused by: io.debezium.DebeziumException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires. Error code: 1236; SQLSTATE: HY000.
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1146) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	... 1 more
Caused by: com.github.shyiko.mysql.binlog.network.ServerException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:937) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850) ~[flink-connector-mysql-cdc-2.2.0.jar:?]
	... 1 more
2022-04-29 18:17:47,527 INFO  com.ververica.cdc.connectors.mysql.source.reader.MySqlSplitReader [] - Close current debezium reader com.ververica.cdc.connectors.mysql.debezium.reader.BinlogSplitReader
2022-04-29 18:17:47,530 INFO  org.apache.kafka.clients.producer.KafkaProducer              [] - [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 0 ms.
2022-04-29 18:17:47,530 INFO  org.apache.kafka.clients.producer.KafkaProducer              [] - [Producer clientId=producer-1] Proceeding to force close the producer since pending requests could not be completed within timeout 0 ms.
2022-04-29 18:17:47,535 INFO  io.debezium.jdbc.JdbcConnection                              [] - Connection gracefully closed
2022-04-29 18:17:47,536 INFO  org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Split fetcher 0 exited.
2022-04-29 18:17:47,537 INFO  org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Closing Source Reader.
2022-04-29 18:17:47,539 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: TableSourceScan(table=[[default_catalog, default_database, mysql_source_table]], fields=[binlog_record]) -> Sink: Sink(table=[default_catalog.default_database.kafka_sink_table], fields=[binlog_record]) (1/1)#0 (9f2413ac0cbc1d54d11bb17b6e09998c) switched from RUNNING to FAILED with failure cause: java.lang.RuntimeException: One or more fetchers have encountered exception
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:223)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:154)
	at org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116)
	at org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:294)
	at org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:69)
	at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
	at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623)
	at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
	at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received unexpected exception while polling the records
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:148)
	at org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:103)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	... 1 more
Caused by: org.apache.kafka.connect.errors.ConnectException: An exception occurred in the change event producer. This connector will be stopped.
	at io.debezium.pipeline.ErrorHandler.setProducerThrowable(ErrorHandler.java:42)
	at com.ververica.cdc.connectors.mysql.debezium.task.context.MySqlErrorHandler.setProducerThrowable(MySqlErrorHandler.java:72)
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource$ReaderThreadLifecycleListener.onCommunicationFailure(MySqlStreamingChangeEventSource.java:1185)
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:973)
	at com.github.shyiko.mysql.binlog.BinaryLogClient.connect(BinaryLogClient.java:606)
	at com.github.shyiko.mysql.binlog.BinaryLogClient$7.run(BinaryLogClient.java:850)
	... 1 more
Caused by: io.debezium.DebeziumException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires. Error code: 1236; SQLSTATE: HY000.
	at io.debezium.connector.mysql.MySqlStreamingChangeEventSource.wrap(MySqlStreamingChangeEventSource.java:1146)
	... 5 more
Caused by: com.github.shyiko.mysql.binlog.network.ServerException: The slave is connecting using CHANGE MASTER TO MASTER_AUTO_POSITION = 1, but the master has purged binary logs containing GTIDs that the slave requires.
	at com.github.shyiko.mysql.binlog.BinaryLogClient.listenForEventPackets(BinaryLogClient.java:937)
	... 3 more

[dik111]

有比较好的方案吗？我也有从指定时间开始同步的需求