A restart during a replication stall on the target duplicates a message, and the next restart skips one #49

Hi! On main (2dcedd3) I can get a duplicate and then a silently skipped message in the target. It takes a restart, such as a crash followed by a supervisor starting the relay again, while one of the target partition's in-sync followers has stopped replicating. The README says offset tracking between source and target allows de-duplication without external stores, so I thought this was worth reporting.

What happens (in my test, source offset N holds the value N+1):

1. The target has caught up with source offsets 0–99. Then one follower of the target partition stops fetching but keeps checking in with the controller, so it stays in the in-sync set for up to replica.lag.time.max.ms (30 s by default). In my test I drop its fetch traffic; a broken link between two brokers would look the same.
2. The relay copies source offset 100. The leader appends it at offset 100, but the high watermark stays at 100 until the stuck follower has it, so the write isn't acknowledged yet (commit_ack_type = "cluster").
3. The relay is killed and started again. At startup it resumes the source at the target's end offset (main.go L80-L91), which comes from ListEndOffsets (common.go L357-L369): the high watermark, 100. The message the old process already stored isn't counted, so the new process logs "initializing new source consumer","offsets":{"0":{"At":100,... and copies source offset 100 again, under a new producer ID, so the broker can't tell it's a retry.
4. Once the follower catches up, both copies become visible: the target holds value 101 at offsets 100 and 101.
5. From then on the target is one message ahead of the source, so every later start resumes one offset too far. The next restart logged "At":201, and source offset 200 (value 201) never reached the target. Nothing in the logs points at either problem.

How often I saw it, with Kafka 4.3.1 (KRaft, 3 brokers, replication factor 3, min.insync.replicas=2), a one-partition topic, --mode single and the config.sample.toml settings (without SASL, and with the target's request_timeout raised to 1s). kaf-relay pins franz-go v1.20.6; I also built it against v1.20.7 and v1.22.2 to rule the client out:

Every failed run had exactly the duplicate and the skip above. A frozen broker stops checking in with the controller, which fences it after about 9 s and lets the high watermark move, so a restart only hits the window if it reads the end offsets before that. Every restart that logged "At":100 failed, and every one that logged "At":101 was clean. Kafka's own console consumer shows the same layout in the target as my checker.

Possible directions, though you know the design much better than I do:

• Resuming from a source offset stored in a header of the last copied record, as zerodha/kaf-relay#44 proposes, would stop the skip: the second restart would resume at source offset 200 despite the duplicate. It wouldn't stop the duplicate itself, because at the first restart the last record the target shows is still the one before the stuck message.
• Avoiding the duplicate as well probably needs transactions on top of that: a restart with the same transactional ID aborts whatever the old process hadn't committed, and readers in read_committed mode never see it. Transaction markers take offsets too, so that would need the header-based resume anyway.

Happy to test a fix.