[source-mssql] CDC Sync runs infinitely during heartbeat change events when the last heartbeat timestamp is null

### Connector Name

source-mssql

### Connector Version

4.1.27

### What step the error happened?

During the sync

### Relevant information

We currently see an issue where CDC syncs run for an infinite amount of time after . This happens when there is a table that has no activity for days. These are the steps that you can try to reproduce. This is how we were able to reproduce on our customer's environment.

Steps to Reproduce:
1. Create a table in MSSQL in incremental mode and shove some good amount of random data into it.
2. Create a connection in airbyte with just that table and run the sync. The first sync will finish with a historical load and then add some more data but don't start the sync yet.
3. Now do not make any updates to the table for the next 20 or 30 days and you should be able to reproduce the issue
4. Make sure that the source has a initial wait time set to max i.e 1 hour and now run the sync and you will see that the connection goes in an idle state where debezium is reading mssql heartbeat (this heartbeat is different from your platform heartbeat) change events which means that the connector lsn heartbeat is not progressing

Generally if the initial wait time out is set to 1 hour, the sync should finish right after the 1 hour mark when the heartbeat messages are in an idle state due to the logic in DebeziumRecordIterator.kt which knows how to handle a situation when heartbeat is not progressing. The firstRecordWaitTime is the intial wait time out

```
private fun heartbeatPosNotChanging(): Boolean {
if (this.tsLastHeartbeat == null) {
return false
}

val timeElapsedSinceLastHeartbeatTs =
Duration.between(this.tsLastHeartbeat, LocalDateTime.now())
return timeElapsedSinceLastHeartbeatTs.compareTo(firstRecordWaitTime) > 0
}
```

Sometimes (as mentioned in steps for reproduction) what happens is that the loop that calls the above function, its stuck in an infinite state which is when the value of this.tsLastHeartbeat is null and i never see the value of this timestamp get updated.
In the below logs, you should see the timestamp where it starts emitting the heartbeat events and after a few spaces I am giving the output of the last time stamp that i saw the heartbeat event. Note that I had to create a custom connector to prove this, so the log messages might be different from what you see on your platform. The sync was running in an idle state for 4 hours even though we had a 1 hour initial wait time out set. I had to cancel the sync otherwise it would run for an indefinite amount of time

Solution:

The easiest way to solve this problem is to check wether this.tsLastHeartbeat has been null for a period of time. This could be set to the initial wait time but ideally this logic should be replaced with a special heartbeat time out since the initial wait time is for debezium to figure out the first record it can fetch in the initial wait time.

### Relevant log output

```shell
2025-05-01 12:26:10 source INFO main i.a.c.i.d.i.RecordWaitTimeUtil(getFirstRecordWaitTime):68 First record waiting time: 3600 seconds
2025-05-01 12:26:10 source INFO main i.a.i.s.m.i.MssqlInitialReadUtil(getCdcReadIterators):197 First record waiting time: 3600 seconds

2025-05-01 12:26:47 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=3, polls=0
2025-05-01 12:26:47 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.47050344S after 187 previous call(s) which were not logged.
2025-05-01 12:26:47 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 12:26:47 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 12:26:47 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 12:26:47 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 12:26:57 source INFO pool-5-thread-1 i.d.c.c.BaseSourceTask(logStatistics):351 2 records sent during previous 00:00:20.548, last recorded offset of {server=FacetsReport, database=FacetsReport} partition is {event_serial_no=2, commit_lsn=00019fda:001324bb:00b7, change_lsn=00019fda:001322e1:012b}
2025-05-01 12:26:57 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=4, polls=0
2025-05-01 12:26:57 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.020363236S after its previous call which was also logged.
2025-05-01 12:26:57 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 12:26:57 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 12:26:57 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 12:26:57 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 12:27:07 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=5, polls=0
2025-05-01 12:27:07 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.521561502S after its previous call which was also logged.
2025-05-01 12:27:07 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 12:27:07 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 12:27:07 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 12:27:07 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 12:27:18 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=6, polls=0
2025-05-01 12:27:18 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.52164078S after its previous call which was also logged.
2025-05-01 12:27:18 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 12:27:18 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 12:27:18 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 12:27:18 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null

2025-05-01 16:01:06 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 16:01:16 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=1230, polls=0
2025-05-01 16:01:16 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.019839935S after its previous call which was also logged.
2025-05-01 16:01:16 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 16:01:16 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 16:01:16 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 16:01:16 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 16:01:26 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=1231, polls=0
2025-05-01 16:01:26 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.520768844S after its previous call which was also logged.
2025-05-01 16:01:26 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 16:01:26 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 16:01:26 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 16:01:26 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
2025-05-01 16:01:37 source INFO pool-5-thread-1 i.a.c.i.d.AirbyteDebeziumHandler$CapacityReportingBlockingQueue(reportQueueUtilization):48 CDC events queue stats: size=0, cap=10000, puts=1232, polls=0
2025-05-01 16:01:37 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):90 CDC events queue poll(): blocked for PT10.520810749S after its previous call which was also logged.
2025-05-01 16:01:37 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):143 GAURAV'S HEARTBEAT STOPPED EVENT: heartbeat position: 00019fda:001324bb:00b7, last heartbeat position: 00019fda:001324bb:00b7
2025-05-01 16:01:37 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(computeNext):147 CDC events queue poll(): returned a heartbeat event: no progress since last heartbeat.
2025-05-01 16:01:37 source INFO main i.a.i.s.m.MssqlCdcTargetPosition(reachedTargetPosition):117 Checking if target position is reached. Target LSN: 00019fdf:000605ba:0003, Position from heartbeat: 00019fda:001324bb:00b7
2025-05-01 16:01:37 source INFO main i.a.c.i.d.i.DebeziumRecordIterator(heartbeatPosNotChanging):269 Checking if heartbeat position is not changing, last heartbeat: null
```

### Contribute

- [x] Yes, I want to contribute