Kafka Connect: recover from metadata commit stall after Kafka cluster recreation

[Schiewatir]

Problem

After a Kafka cluster is recreated (old cluster deleted, new one started), the Apache Iceberg Kafka Connect sink stops committing metadata. Parquet files accumulate in object storage but no snapshots are created, and the logs show "Coordinator … found nothing to commit to table" on every commit cycle. (Reported in #15293.)

Root cause: The control topic resets to offset 0 on the new cluster, but the Iceberg snapshot still stores committed offsets from the old cluster (e.g. {"0":100}). The deduplication filter in Coordinator.commitToTable() rejects every DataWritten event whose offset is below the stored baseline, so all data files are silently dropped.

// before fix – stale minOffset = 100, envelope.offset() = 0..4 → all filtered
Long minOffset = committedOffsets.get(envelope.partition());
return minOffset == null || envelope.offset() >= minOffset;

Fix

When a partition's current control-topic offset is lower than its stored committed offset, the committed offset for that partition is reset to the current offset minus one (so the next write is accepted). This is detected per-partition during commitToTable(), so there is no behaviour change for the normal (non-reset) path.

After the reset, the filter accepts the new events, a snapshot is committed, and its stored offset is reset to the new baseline — replication resumes.

Fixes #15293