kafka connect: data loss with default configs due to persistent zombie coordinators

[twthorn]

Apache Iceberg version

1.11.0 (latest release)

Query engine

None

Please describe the bug 🐞

Data loss is observed with the kafka connect iceberg sink with the following configs & timeline.

Configs

• iceberg.kafka.auto.offset.reset: latest (default)
• control topic retention.ms / retention.bytes: some constant (not infinite retention)

Timeline

1. Coordinator A elected, healthy, commits control-topic offset 100, assigned -coord group 0 partition in consumer group
2. Coordinator B elected, healthy, commits offset 105, assigned -coord group 0 partition in consumer group
3. Coordinator A fails to exit (eg due to any cause in iceberg-kafka-connect zombie coordinator thread when writing to S3 fails #16016) now a zombie, ie assigned no partitions, last control-topic offset in memory is 100
4. Coordinator A regularly commits control-topic offset 100 on each commit, zero table commits
5. The record at coord offset 100 is removed from kafka due to retention (bytes or time), while coordinator A still exists (ie Coordinator A's pod & thread exist without crashing)
6. During this time sawtooth commit pattern oscillating between old and recent offset (see Control topic offsets for coordinator resets to some older value after making progress. #13995)
7. Coordinator B dies
8. Coordinator A writes its old control-topic offset 100 once more, last writer wins
9. Coordinator A dies
10. Coordinator C elected, attempts to read from control-topic offset 100, no longer in retention
11. Coordinator C Auto offset reset to latest, skips all control topic records up to tip, data loss occurs for all records between B's last commit and C's election

Solution

We propose to make the coordinator read the currently committed coord group offset before committing, and never commit an offset lower than it. This provides cross-process monotonicity on the shared group -coord offset. The result will prevent data loss (note this is distinct from the in-process local monotonic map proposed in #17340 which cannot prevent this data loss scenario). The only potential functionality it will remove is an operator wanting to rewind a coordinator. But typically that is not done (only source offsets ie worker offsets are rewound).

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[twthorn]

PR published #17552