feat: expose startMessageRollbackDurationInSec in ConsumerOptions#416
feat: expose startMessageRollbackDurationInSec in ConsumerOptions#416davider80 wants to merge 7 commits into
Conversation
The Pulsar protocol supports resetting the subscription cursor back by N seconds at subscribe time (CommandSubscribe.start_message_rollback_duration_sec, field 16), but the option was not exposed by the client. This is the protocol-native way for readers and non-durable consumers to start 'N seconds back' (e.g. to rebuild recent state after a restart without persisted positions). The alternative — seek(None, timestamp) after creation — forces a broker-side consumer close + reconnect, which is racy right after subscribe (see also streamnative#415). Zero/unset values are not sent on the wire (proto default 0 = no rollback). Includes a unit test on the generated CommandSubscribe.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 21be1b1193
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…progress Re-sending startMessageRollbackDurationInSec on every resubscribe would rewind the cursor by the whole window after each broker/network blip, replaying messages the consumer already processed. Mirror the Java client (ConsumerImpl only sends it while startMessageId still equals the initial one): track whether the engine has dequeued any message and strip the rollback from the options used on reconnect once it has.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 36c08d99ac
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…ribe Clearing the rollback alone meant a reconnect after the first dequeued message but before the window drained resubscribed at the default position (latest), losing the undrained remainder. Mirror the Java client fully (ConsumerImpl.clearReceiverQueue): track the last dequeued message id and resubscribe from it; the broker resumes delivery after that entry. A seek would have left that recorded position (and the original start_message_id) stale, so reconnects silently undid the seek: the consumer now notifies the engine (EngineMessage::SeekPosition), which rebases the subscribe options on the seek target and drops the rollback. Timestamp seeks carry no message id; reconnects then fall back to the subscription's default initial position, which is documented on the variant.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: c25b6c105a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Consumer::seek rebuilds a replacement TopicConsumer from config().clone() after seeking, so rebasing only the engine left the replacement subscribing with the stale pre-seek start position and re-applying the one-shot rollback, moving the cursor away from the seek target. Single-topic: the seek target becomes the config's start_message_id and the rollback is dropped. Multi-topic: only the rollback is dropped from the shared config — message ids are topic-specific, so a single seek target cannot be rebased into a config used by every topic (including the consumers update_topics() creates later).
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 5b3331e745
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Sending start_message_id on reconnect for a consumer that was not created with the rollback option repositions the broker cursor: a durable subscription with an in-flight unacked message would have its cursor moved past it, losing the message instead of redelivering it. The resume rewrite is now gated on track_resume_position, captured once at engine construction from the rollback option — consumers without it reconnect with their original options exactly as before this feature. In the multi-topic seek rebuild, each replacement consumer is now built from its own topic's config (which TopicConsumer::seek rebased for the seeked topics) instead of the shared central config, so per-topic seek positions survive the rebuild; the shared config remains the fallback for topics without an old consumer, and the one-shot rollback is dropped from all of them.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ab5d829750
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
CommandSubscribe already filters a zero rollback out of the wire as "no rollback"; the engine's resume-tracking gate must agree, or a consumer configured with an explicit zero would still get client-side resume rewrites — the durable-cursor hazard the gate exists to prevent.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: cab854d3bb
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…tions on multi seek Two related corrections to the reconnect/seek semantics: Non-durable subscriptions (readers included) now opt into resume tracking, not just rollback consumers: their broker-side cursor is dropped on disconnect, so a reconnect without a resume id re-pins to the configured start position — including a just-seeked target, forever. This matches the Java client (ConsumerImpl.clearReceiverQueue advances startMessageId to the last dequeued message for non-durable subscriptions) and bounds the reconnect replay by progress instead of replaying from the original start. Durable subscriptions without a rollback stay untouched: the broker cursor governs. The multi-topic seek rebuild now clones each topic's config as-is: seeked topics were already rebased by TopicConsumer::seek, and the topics that were NOT seeked keep their original options — including a configured rollback, whose bounded window replay is at-least-once, whereas dropping it resubscribed them at the default position and skipped messages. The central config keeps its rollback too: it only seeds genuinely new topics (update_topics discoveries), whose first subscribe legitimately applies it.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: df874582ce
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| fn tracks_resume_position(options: &ConsumerOptions) -> bool { | ||
| rollback_enabled(options) || options.durable == Some(false) | ||
| } |
There was a problem hiding this comment.
For a durable subscription configured with with_start_message_rollback_duration_secs(...) (including the default durable case where durable is unset), this gate opts the engine into rewriting reconnects to start_message_id = last_dequeued_message_id. If the connection drops after a message is delivered but before it is acked, a durable subscription's broker cursor would normally redeliver that in-flight message, but the rewritten subscribe can reposition the cursor past it and lose it. Fresh evidence is that the current gate still returns true for any positive rollback before considering whether the subscription is durable; the resume rewrite should be limited to non-durable/client-driven cursors, while durable rollback consumers only need the one-shot rollback cleared after progress.
Useful? React with 👍 / 👎.
1 participant
Motivation
The Pulsar protocol supports resetting the subscription cursor back by N seconds at subscribe time (
CommandSubscribe.start_message_rollback_duration_sec, field 16 — "If specified, the subscription will reset cursor's position back to specified seconds and will send messages from that point"), and the Java client exposes it asstartMessageRollbackDurationInSec. pulsar-rs does not expose it.This is the protocol-native way for readers and non-durable consumers to start "N seconds back" — our use case: a stateful view service rebuilding recent state after a restart without persisted positions. The current alternative is
seek(None, Some(timestamp))right after creation, which forces a broker-side consumer close + reconnect and is racy immediately after subscribe (we hit lookups failing mid-reconnect; related hardening in #415).Changes
ConsumerOptions.start_message_rollback_duration_secs: Option<u64>+with_start_message_rollback_duration_secs()builder, consistent with the existing options style.messages::subscribe; zero/unset values are not sent on the wire (proto default 0 = no rollback).CommandSubscribecarries the field (and omits it for 0/unset).Verification
cargo test --features tokio-runtime— the new unit test passes; the pre-existing integration tests require a live broker (unchanged).cargo clippyintroduces no new warnings.We run this client in production-grade workloads (two Rust services consuming a ~30k msg/s firehose); happy to adjust naming/shape to maintainer preference.