feat(connectors): let sinks control when offsets are committed - #3954
feat(connectors): let sinks control when offsets are committed#3954kriti-sc wants to merge 2 commits into
Conversation
|
Thanks for the PR. It is labeled Slash commands (own line, regular comment) move it around the queue:
See CONTRIBUTING.md for details. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #3954 +/- ##
============================================
+ Coverage 84.98% 85.03% +0.04%
Complexity 1402 1402
============================================
Files 1226 1225 -1
Lines 180701 179798 -903
Branches 147005 146102 -903
============================================
- Hits 153575 152889 -686
+ Misses 23070 22897 -173
+ Partials 4056 4012 -44
🚀 New features to boost your workflow:
|
|
This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. If you need a review, please ensure CI is green and the PR is rebased on the latest master. Don't hesitate to ping the maintainers - either Thank you for your contribution! |
There was a problem hiding this comment.
plz read this end to end. the core claim holds - i manually checked code, from after_consuming through the server's PollingKind::Next handling and a restarted consumer really does resume at stored_offset + 1. the tests are honest too: the after_polling one asserts the loss instead of hiding it.
three things before this lands.
the failure semantics change on the default path, not just the opt-in one. SinkContainer::consume maps every Err from Sink::consume to 1, so after this each of the 15 shipped sinks permanently stops its connector on a single transient failure - no retry, no backoff, no auto-restart, restart_connector only reachable over HTTP. "nothing changes for existing configs" in the description is not right, and your own given_after_polling_when_sink_rejects_batch_should_still_advance_offset asserts the new stop. either gate the hard stop behind AfterConsuming, or add backoff.
the sibling-task shutdown is a second behaviour change riding along, and it drops partial batches. it deserves at least a line in the description.
a question i could not settle: the batch is not partition-scoped but the commit is. consume_messages pushes into one batch regardless of message.partition_id, overwriting partition_id and message_offset each iteration, then commits only the last message's partition. a group member owning several partitions could flush a batch spanning a switch and leave the other partition's offset stale. i could not build a trigger - the server bounds a read by commit_offset and batch_size equals batch_length - so it may be unreachable by construction. is it?
on "closes #2928": the literal condition in that issue survives by default. its stated impact is resolved opt-in and #2927 is fixed outright, so this may well be fine - just flagging it rather than asserting it.
ci is red on two one-line fixes (module ordering, a stray blank line in one toml). everything else red is matrix fail-fast cancellation rather than real failures, which also means clippy, sort, machete, doctest and the test suite have never reported on this head. the new config unit tests have not run anywhere, since cargo build skips cfg(test). the three --locked cross-builds did pass.
| // modes stop: continuing would hand every later batch to the same | ||
| // failing target, and under `AfterPolling` each of those is already | ||
| // committed at poll time, so the topic would drain into nothing. | ||
| return Err(RuntimeError::SinkRejectedBatch(plugin_id, consume_result)); |
There was a problem hiding this comment.
critical: this fires under the default after_polling too, and the SDK maps every Err from a sink to 1, so one transient write failure now permanently stops the connector. nothing auto-restarts it - restart_connector is only reachable over HTTP.
| // failure condemns the rest. Stopping them here keeps the | ||
| // failure domain the same as the recovery domain: the whole | ||
| // connector goes down, and `restart_connector` brings it back. | ||
| let _ = shutdown_tx.send(()); |
There was a problem hiding this comment.
critical: a sibling task woken by this breaks out and drops its half-filled batch. under after_polling those offsets are already committed, so one topic failing silently loses in-flight messages on the others.
|
|
||
| if offset_commit == OffsetCommitMode::AfterConsuming | ||
| && let Err(error) = consumer | ||
| .store_offset(message_offset, Some(partition_id)) |
There was a problem hiding this comment.
critical: any IggyError here, including a transient disconnect, kills the connector with no retry. RuntimeError::IggyError prints just "Iggy error", so last_error tells an operator nothing about this path.
| return Err(RuntimeError::SinkRejectedBatch(plugin_id, consume_result)); | ||
| } | ||
|
|
||
| if offset_commit == OffsetCommitMode::AfterConsuming |
There was a problem hiding this comment.
warning: this commits the batch's last offset, marking everything up to it consumed - including messages already dropped by decode or transform failures. after_consuming is at-least-once per batch, not per message.
| Err(_) => 0, | ||
| }; | ||
|
|
||
| let (processed_count, decode_us, prepare_us, ffi_us) = match &result { |
There was a problem hiding this comment.
warning: a rejected batch takes the Ok arm, so the decode/prepare/ffi histograms and the benchmark event record it as processed. contradicts the comment right above about sub-stages only on success.
| @@ -27,6 +27,8 @@ pub enum RuntimeError { | |||
| FailedToSerializeMessagesMetadata, | |||
| #[error("Failed to serialize raw messages")] | |||
There was a problem hiding this comment.
nit: positional tuple variant - named fields read better, and this message is the only thing a test or an operator can match on.
| let batch_index = state.batches_consumed; | ||
| let topic_selected = | ||
| self.reject_topics.is_empty() || self.reject_topics.contains(&topic_metadata.topic); | ||
| let should_fail = topic_selected |
There was a problem hiding this comment.
nit: batches_consumed is one counter across all topics but the threshold is checked per topic. fine at 0, meaningless at any other value.
|
|
||
| mod benchmark; | ||
| mod error_isolation; | ||
| mod offset_commit; |
There was a problem hiding this comment.
nit: rustfmt wants this after mod http_state; - this is the cargo fmt failure.
| # specific language governing permissions and limitations | ||
| # under the License. | ||
|
|
||
|
|
There was a problem hiding this comment.
nit: extra blank line, drop one. this alone fails both taplo and the license header check.
|
|
||
| let consume_result = match &result { | ||
| Ok(timing) => timing.consume_result, | ||
| Err(_) => 0, |
There was a problem hiding this comment.
simplification: the Err arm is unreachable - line 446 returns first. this block also repeats the match &result directly below it.
github-actions
Bot
added
S-waiting-on-author
2 participants
Closes #2928, #2927
Related: #3203
Rationale
Sinks must have an option to decide exactly when it is safe to advance the offset, as only the sink knows when a write is truly durable from the target's perspective.
What changed?
This PR adds an
offset_commitsetting to sink configs:after_polling(default) - existing behaviour of offset being committed when messages are polled. Nothing changes for existing configs.after_consuming- the offset is stored only once the sink reports that it wrote the batch.The offset commit decision hinges upon the i32 returned by the sink.
This also required binding the i32 returned by the sink, which was previous discarded. Now, the sink exits if i32 != 0, regardless of offset commit configuration, which is the correct behaviour as a sink returns error for failures lying outside its purview.
This PR also introduces integration tests that inject failures into a running sink and verify the desired commit behaviour.