[FLINK-37191][mysql] Avoid back filling if the lowWatermark of BinlogSplit is equal to highWatermark.

[lvyanquan]

To fix timeout in test case of PolardbxSourceITCase.

Avoid extra backfill when there is no binlog event between lowWatermark and highWatermark.

[lvyanquan]

@yuxiqian @leonardBang PTAL.

[leonardBang]

CI failed....

[yuxiqian]

BinlogOffset#compareTo is a general-purposed comparison method, and this change might affect more than backfill necessity check. Is it possible to tweak SnapshotSplitReader#isBackfillRequired only?

[yuxiqian]

Seems testBinlogOffsetCompareWithSnapshotAndBinlogPhase is failing due to this change. DataStream migration test is irrelevant, and should be fixed in #3884.

[lvyanquan]

Hi, @JNSimba.

I found that the change of #3415 will cause the following problem:

1. Even though there is not record between lowWatermark and highWatermark, we still need a backfill procedure, and the timestamp of highWatermark is greater than lowWatermark.
2. If there is no more record after highWatermark, the backfill procedure will never stop, as we will need to wait for a record that is at or after the highWatermark(as the following code displayed).
   

   flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java

   Line 102 in 7717779

   if (currentBinlogOffset.isAtOrAfter(binlogSplit.getEndingOffset())) {

I thank that the change of #3415 mainly to solve the problem of the comparson of highWatermark and the last record, but in reality, we don't need to read this record, so we can skip it directly.

flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java

Line 102 in 7717779

if (currentBinlogOffset.isAtOrAfter(binlogSplit.getEndingOffset())) {

Notice that this will not cause data loss, as this record will be read in backfill phase.

What do you think?

[JNSimba]

@lvyanquan Is the backfill phase not stopped because the ts_sec of offsetContext is always 0, which is less than the ts_sec of highwatermark(binlogSplit.getEndingOffset)?

flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/task/MySqlBinlogSplitReadTask.java

Lines 99 to 102 in 7717779

	final BinlogOffset currentBinlogOffset =
	RecordUtils.getBinlogPosition(offsetContext.getOffset());
	// reach the high watermark, the binlog reader should finished
	if (currentBinlogOffset.isAtOrAfter(binlogSplit.getEndingOffset())) {

2. If so, the following situation may also cause problems.
   Since the server_id of highwatermark is equal to 0, but the server_id of offsetContext is not equal to 0, the comparison will trigger the comparison of timestamp, so the ts_sec of offset will be less than the ts_sec of highwatermark, and it will also get stuck?
   

   flink-cdc/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/offset/BinlogOffset.java

   Lines 228 to 236 in 7717779

   	if (serverId != targetServerId) {
   	// These are from different servers, and their binlog coordinates are not related. So
   	// the only thing we can do
   	// is compare timestamps, and we have to assume that the server timestamps can be
   	// compared ...
   	long timestamp = this.getTimestampSec();
   	long targetTimestamp = that.getTimestampSec();
   	return Long.compare(timestamp, targetTimestamp);
   	}

[lvyanquan]

@lvyanquan Is the backfill phase not stopped because the ts_sec of offsetContext is always 0, which is less than the ts_sec of highwatermark(binlogSplit.getEndingOffset)?

Yes.

and it will also get stuck?

Make sense. So I would like to revert the change of adding timestamp to highWaterMark and skip the record that is equal to the offset of highWaterMark.

The test case of BinlogSplitReaderTest will be remained to verify this.

[JNSimba]

LGTM

[yuxiqian]

Pulled and tested locally, I could confirm that PolarDBX test case could run without problem.

But seems BinlogSplitReaderTest#testBinlogOffsetCompareWithSnapshotAndBinlogPhase (introduced in #3415) will always fail on this assertion:

assertTrue(sourceRecords.isEmpty());

Expecting empty but was: [SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1739012804, file=mysql-bin.000003, pos=15691, row=1, server_id=223344, event=2}} ConnectRecord{topic='mysql_binlog_source.customer_tesuwt.customers', kafkaPartition=null, key=Struct{id=999999}, keySchema=Schema{mysql_binlog_source.customer_tesuwt.customers.Key:STRUCT}, value=Struct{after=Struct{id=999999,name=user_22,address=Shanghai,phone_number=123567891234},source=Struct{version=1.9.8.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1739012804000,db=customer_tesuwt,table=customers,server_id=223344,file=mysql-bin.000003,pos=15831,row=0,thread=5},op=c,ts_ms=1739012808026}, valueSchema=Schema{mysql_binlog_source.customer_tesuwt.customers.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}]

[lvyanquan]

This change is necessary because that if the TimestampSec is 0, the assumption of These are from different servers, and their binlog coordinates are not related is not correct, these two BinlogOffsets could come from the same server.

[yuxiqian]

Pulled, ran tests locally and looks good.

Considering BinlogOffset is part of public API and has complex comparison logic, maybe we can add some tests later to cover it in case if we break something accidentally again in the future.