numinnex opened a new pull request, #4734: URL: https://github.com/apache/iggy/pull/4734
## Which issue does this PR address? Closes #4233 Closes #4234 Closes #4235 Relates to #4232, #4236, #4237, #2927, #2928 ## Rationale A sink stores the offset of a poll when the server answers it, before the plugin writes anything, so a failed or interrupted write loses the batch. The 503-then-restart reproduction in #4232 drops the rejected record, and a local Iceberg test lost 500 rows to sink kills and 4,700 to a single catalog restart. This PR states the delivery contract that #4233 asks for and implements it as an opt-in mode in the SDK, the runtime and the HTTP sink. ## What changed? The runtime stored a sink's offset on read, so it could not hand a failed batch over again. A sink with `[delivery] guarantee = "at_least_once"` now polls without storing, the plugin accepts batches in `consume`, and the runtime stores an offset only after the plugin's `flush` proved the records up to it delivered. A failure is retried from the first unflushed record or stops the sink with its partition and offset, and an invalid record stops it unless `invalid_messages = "skip"`. Sinks without the table keep today's behaviour, plugins built with an older SDK keep loading, and the HTTP sink is the first plugin to opt in. ## Delivery contract (#4233) Merging this PR records the agreement #4233 asks for. Version 1 implements every rule below except the parts marked as later or #4238. **Modes.** `at_most_once`, the unchanged default, can lose a record on failure. `at_least_once` never loses a record while Iggy still holds it, and can deliver it more than once: ```toml [delivery] guarantee = "at_least_once" retry_max_attempts = 10 # unset retries until the destination recovers retry_initial_backoff = "1s" retry_max_backoff = "30s" flush_interval = "5s" # unset flushes after every batch flush_max_messages = 10000 invalid_messages = "stop" # or "skip" ``` **Rules for `at_least_once`:** 1. **Checkpoint.** The stored offset of a partition is its last complete record: delivered, filtered by a transform, or skipped under `skip`. It never passes the first incomplete record, even when later records succeeded. 2. **Acknowledgment.** A successful `consume` only accepts records. A successful `flush` proves every accepted record of the listed partitions delivered, up to each listed offset. The runtime flushes after every batch, or with `flush_interval` when it runs out or a partition holds `flush_max_messages` unflushed records, and at shutdown. A plugin that writes inside `consume` keeps the default no-op flush. 3. **Failures.** A call or a flush covers all of its records, and an uncertain outcome counts as a failure. A failure crosses the plugin boundary as transient or permanent, and the plugin logs the reason. A plugin built before the contract reports every failure as transient. 4. **Retries.** A transient failure hands the unflushed records over again with a capped exponential backoff, up to `retry_max_attempts`. A permanent failure, or the last attempt, stops the consumer task with the partition and offset in the sink error, and a restart resumes at the stored offset. A failed offset store is retried and never treated as undelivered data. 5. **Duplicates.** A failed call or flush, a crash, an ownership change or an uncertain store can deliver records above the stored offset again, with the same stream, topic, partition and offset to deduplicate on. 6. **Invalid records.** A record that fails to decode, transform or serialize stops the task, unless `invalid_messages = "skip"` skips it with an error count and its offset in the log. A dead-letter policy can come later. 7. **Lifecycle.** A stop, restart or reconfiguration flushes once and stores what the flush proved. A crash resumes at the stored offsets. A result of a closed or replaced plugin instance, or of a partition generation that is no longer owned, stores nothing. Version 1 has no hook before a rebalance takes a partition away, so the next owner redelivers from the stored offset. Flushing before release is later. 8. **Isolation.** Version 1 isolates the consumer task, one per configured topic: a stopped partition stops its topic, and the sink's other topics keep delivering. 9. **Retention and durability.** A partition whose next record is gone, after retention or a new partition history, stops with an error instead of skipping ahead. Records lost by Iggy itself are out of scope, so use `persisted` durability and a retention longer than the longest expected sink outage. **Compatibility.** The SDK gains two optional exports, `iggy_sink_enable_at_least_once` and `iggy_sink_flush`, and a status code for a permanent failure. No existing symbol changes, so no major SDK version is needed. A sink configured for the guarantee refuses to start with a plugin that does not opt in, and a plugin can refuse a configuration it cannot support. The status values stay as they are, and a stopped task reports `error`. Changing the mode keeps the consumer group and its offsets and replays nothing. The default switch, migration guidance and per-partition visibility follow in #4238, and the other bundled sinks in #4236. **Selected approach.** The runtime-owned commit and the rule that a failure is never stepped over come from #3203 and #3518. The flush barrier is new, because sinks such as S3 buffer across calls. A commit callback driven by the plugin (#3203) was rejected because it splits checkpoint ownership, and per-batch acknowledgments with batch ids (#3855) because they need a callback registry and gap tracking. In Kafka Connect terms, `consume` is `put`, a flush followed by a checkpoint is `flush`/`preCommit`, and `invalid_messages` is `errors.tolerance`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
