We are using debezium embedded engine and we have noticed that in source databases that we have lots of non stop writes and our debezium connector is restarted during the time records are being read by debezium, on the following start of the connector it always skips one record. I am guessing that it might be something to the way offsets are being saved. I can see in the offset_storage topic that the last offset is saved with field "row=1" indeed this row is skipped on the start of the engine. This happens only in cases when new sourceRecords are being read while the connector is stopping(gracefully or not). 

Bug report

For bug reports, provide this information, please:

What Debezium connector do you use and what version?

2.5.1.Final . 
We have recently migrated from 1.9.7.Final where the same issue was tested and we didn't found a problem. 

What is the connector configuration?

I am attaching a minimum reproducable java spring boot project for you to be able to test and simulate the same behaviour.

What is the captured database version and mode of depoyment?

On-prem

What behaviour do you expect?

I expect that, when the connector is restarted to not skip records, since this record that is missing(always it is 1 record) is never processed by the engine. It is present in the binlog thought.

What behaviour do you see?

I think it is something related to the way offsets are being saved. But i am not sure

Do you see the same behaviour using the latest relesead Debezium version?

I tried with the latest version, i still see the same behaviour

Do you have the connector logs, ideally from start till finish?

In the attached java project you can see all the logs. In the README.md I have explained a step by step on how to reproduce the issue.

How to reproduce the issue using our tutorial deployment?

In the attached java project you can see all the logs. In the README.md I have explained a step by step on how to reproduce the issue.
I have attached two files but use the latest one debezium-engine-simulation2.zip.  (I can't remove the first one)

Feature request or enhancement

-

Which use case/requirement will be addressed by the proposed feature?

-

Implementation ideas (optional)

-