Bug report

What Debezium connector do you use and what version?

ibmi connector 3.4.0-SNAPSHOT

What is the connector configuration?

snapshot mode = initial

What is the captured database version and mode of deployment?

On prem DB2 

What behavior do you expect?

connector to continue with the same offsets when restarting after snapshot has completed

What behavior do you see?

connector jumps to the latest offset when restarted even after snapshot has completed

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

Yes

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

can be supplied

How to reproduce the issue using our tutorial deployment?

create a connector with snapshot mode = "initial" and wait for the snapshot to complete

restart the connector and watch the logs for the initial offset, it will start with a new offset not the one in the offsets topic

Implementation ideas (optional)

The debezium connector like others checks the snapshotter `shouldStreamEventsStartingFromSnapshot` method and only uses the previous offset if false e.g. postgres:

https://github.com/debezium/debezium/blob/622b18b9870ec8dec0c87363a0721f9729f7d6d8/debezium-connector-postgres/src/main/java/io/debezium/connector/postgresql/PostgresSnapshotChangeEventSource.java#L156

the issue I believe is caused by the ibmi connector needing access to the schema to decode the entries in the journal so it always snapshots for the schema but normally doesn't snapshot for data.
I think it is assumed this will only be called if we are snapshotting data, though that isn't the case here.
Proposed fix is to add an additional check if we have snapshotted for data before also checking the  `shouldStreamEventsStartingFromSnapshot` method

our determineSnapshotOffset method would thus be something like:

boolean shouldSnapshotData = shouldSnapshotData(previousOffset, snapshotterService.getSnapshotter());

boolean useOffset = !shouldSnapshotData || (shouldSnapshotData && !snapshotterService.getSnapshotter().shouldStreamEventsStartingFromSnapshot());

if (previousOffset != null && previousOffset.isPositionSet() && useOffset) {

snapshotContext.offset = previousOffset;