diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d24d86b00..ce401ff0f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,7 +21,11 @@ jobs: - run: cargo fmt -- --check - run: cargo clippy -- -Dwarnings - run: cargo clippy --tests -- -Dwarnings + - run: cargo clippy --features kip-932 -- -Dwarnings + - run: cargo clippy --features kip-932 --tests -- -Dwarnings - run: cargo test --doc + - run: cargo test --features kip-932 --lib consumer::share + - run: cargo test --features kip-932 --test test_share_consumer check: strategy: @@ -33,6 +37,8 @@ jobs: rdkafka-sys-features: cmake-build,libz-static,curl-static - os: ubuntu-24.04 features: tracing + - os: ubuntu-24.04 + features: kip-932 - os: ubuntu-24.04 features: cmake-build,ssl-vendored,gssapi-vendored,libz-static,curl-static,zstd rdkafka-sys-features: cmake-build,ssl-vendored,gssapi-vendored,libz-static,curl-static,zstd diff --git a/Cargo.toml b/Cargo.toml index e556fd18c..02bddfd2c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -66,9 +66,14 @@ zstd-pkg-config = ["rdkafka-sys/zstd-pkg-config"] external-lz4 = ["rdkafka-sys/external-lz4"] external_lz4 = ["rdkafka-sys/external_lz4"] static-linking = ["rdkafka-sys/static-linking"] +# Opt in to the KIP-932 (Queues for Kafka) share consumer surface. The Rust +# types ship now so downstream crates can compile against the API shape, but +# runtime methods return `KafkaError::Unsupported` until librdkafka exposes +# the public share consumer C API (tracking confluentinc/librdkafka#5441). +kip-932 = [] [package.metadata.docs.rs] # docs.rs doesn't allow writing to ~/.cargo/registry (reasonably), so we have to # use the CMake build for a proper out-of-tree build. -features = ["cmake-build", "naive-runtime", "tracing", "tokio"] +features = ["cmake-build", "naive-runtime", "tracing", "tokio", "kip-932"] rustdoc-args = ["--cfg", "docsrs"] diff --git a/PLAN.md b/PLAN.md new file mode 100644 index 000000000..94147b706 --- /dev/null +++ b/PLAN.md @@ -0,0 +1,431 @@ +# Implementation plan: KIP-932 (Queues for Kafka) scaffolding + +Target branch: `worktree-kip-932-queues` (fork: `j7nw4r/rust-rdkafka`). +Target PR: draft against `j7nw4r/rust-rdkafka:master`. Upstream +(`fede1024/rust-rdkafka`) is a follow-up once librdkafka exposes the +public KIP-932 C API. + +Reference: +Upstream tracking: librdkafka issue +. + +## 0. Working mode + +This plan is a living document. It ships in a draft PR on the fork +and is iterated on in-tree while we wait for librdkafka to land +public KIP-932 C API. Expect the document to evolve as upstream +shapes solidify; treat the latest commit on this branch (not the +initial version) as the source of truth. + +Practical consequences: + +- The draft PR opens as soon as phase 1 lands and stays draft until + librdkafka's public API is available, the scaffolding is wired to + it, and the validation gate (section 7) passes. +- Phases 2 through 4 (types, error variant, trait + stub) can be + executed against the current librdkafka pin since they introduce + no FFI calls. +- Phase 5 onward (tests, CI lane, docs) may be paused or revised + pending the upstream API shape, to avoid churn against a moving + target. +- API shapes in section 5 are provisional. When librdkafka's public + header lands, this document is updated first, then the code. +- The "Risks and follow-ups" section is the working backlog; + add entries as we learn more rather than starting a separate + tracking doc. + +## 1. Context and constraints + +- rust-rdkafka is a Rust FFI wrapper over librdkafka via `rdkafka-sys`. +- rdkafka-sys is currently pinned to librdkafka `2.10.0` + (`rdkafka-sys` crate version `4.9.0+2.10.0`). +- As of today (2026-05-22), librdkafka master `src/rdkafka.h` exposes + **zero** public symbols for KIP-932: no `rd_kafka_share_consumer_*`, + no `RD_KAFKA_SHARE_*`, no share group / share session entrypoints. + Verified by direct fetch of `src/rdkafka.h` and listing of `src/`. +- librdkafka has multiple in-progress `[KIP-932]` PRs (#5436, #5437, + #5442, #5443, #5444, #5449, #5451, #5452, #5453, #5455). Tracking + issue #5441 is open. +- Consequence: a functional ShareConsumer is **not implementable today** + in rust-rdkafka. This PR delivers the **public Rust API surface** and + scaffolding so that wiring real FFI is a mechanical follow-up once + librdkafka ships a public API. + +## 2. Goal and non-goals + +### Goal + +Land a draft PR that: + +1. Adds the public Rust API for KIP-932 share consumers, behind a + cargo feature flag `kip-932` (off by default). +2. Compiles cleanly with `--features kip-932` and without. +3. Has unit tests that exercise the public surface (enum + round-trips, config builder, trait method dispatch through a + stubbed implementation). +4. Drives CI green: `cargo build`, `cargo build --features kip-932`, + `cargo test`, `cargo test --features kip-932`, `cargo fmt -- --check`, + `cargo clippy --features kip-932 -- -D warnings`. + +### Non-goals + +- No real broker RPC traffic. No `ShareFetch`, `ShareAcknowledge`, + `ShareGroupHeartbeat` wire calls. +- No changes to `rdkafka-sys/librdkafka` submodule pin. No new C bindings. +- No integration tests against a Kafka broker for share groups. The + existing docker-compose harness is not extended. +- No CLI / admin parity for `kafka-share-groups.sh` operations on + `AdminClient`. That belongs in a follow-up. +- No changes to the existing `BaseConsumer` / `StreamConsumer` / + `Consumer` trait. KIP-932 is a parallel surface, not a subtype. + +## 3. Module layout + +``` +src/ + consumer/ + mod.rs (unchanged trait surface; re-export gated) + base_consumer.rs (unchanged) + stream_consumer.rs (unchanged) + share/ (NEW, gated on feature = "kip-932") + mod.rs (module declarations + re-exports) + acknowledge.rs (AcknowledgeType, AcknowledgementCommitCallback) + config.rs (ShareConsumerConfig builder + parse helpers) + context.rs (ShareConsumerContext trait, default impl) + consumer.rs (ShareConsumer trait + BaseShareConsumer stub) + records.rs (ShareConsumerRecords, ShareRecord wrappers) + error.rs (KipNotSupported helpers, ShareError shapes) +tests/ + test_share_consumer.rs (NEW, feature-gated) +``` + +Rationale for a `share/` submodule rather than flat files: KIP-932 +introduces five tightly-coupled types and a parallel consumer surface; +a sub-module keeps the diff isolated and lets the feature flag gate at +the `pub mod share;` line. + +Re-exports from `consumer/mod.rs`: + +```rust +#[cfg(feature = "kip-932")] +pub mod share; + +#[cfg(feature = "kip-932")] +#[doc(inline)] +pub use self::share::{ + AcknowledgeType, AcknowledgementCommitCallback, BaseShareConsumer, + ShareConsumer, ShareConsumerConfig, ShareConsumerContext, + ShareConsumerRecords, ShareRecord, +}; +``` + +## 4. Cargo feature flag + +Add to `Cargo.toml`: + +```toml +[features] +# ... existing ... +kip-932 = [] +``` + +- Off by default. Not in the `default` feature list. +- Documented in `Cargo.toml` comment block and in `lib.rs` rustdoc. +- `[package.metadata.docs.rs]` updated to add `kip-932` so docs.rs + renders the share API. + +No changes to `rdkafka-sys` features. No bindings change. + +## 5. Public API surface + +All types live in `crate::consumer::share::*` and are re-exported per +section 3. Types track the KIP's Java surface but use idiomatic Rust. + +### 5.1 `AcknowledgeType` + +```rust +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash)] +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub enum AcknowledgeType { + Accept, + Release, + Reject, +} +``` + +`Display` implementation maps to the KIP string forms `accept`, +`release`, `reject`. `FromStr` accepts the same case-insensitively. + +### 5.2 `ShareConsumerConfig` + +A typed builder backed by a `ClientConfig` for librdkafka-style passthrough +properties. Validated fields per the KIP: + +| Property | Type | Default | +|---|---|---| +| `group_id` | `String` | required | +| `share_acknowledgement_mode` | `AcknowledgementMode::{Implicit, Explicit}` | `Implicit` | +| `share_auto_offset_reset` | `AutoOffsetReset::{Earliest, Latest}` | `Latest` | +| `share_isolation_level` | `IsolationLevel::{ReadUncommitted, ReadCommitted}` | `ReadUncommitted` | +| `group_share_delivery_attempt_limit` | `u32` | `5` | +| `group_share_record_lock_duration_ms` | `u32` | `30_000` | +| `group_share_heartbeat_interval_ms` | `Option` | `None` (broker default) | +| `group_share_session_timeout_ms` | `Option` | `None` (broker default) | + +Method `into_client_config(self) -> ClientConfig` writes these as +canonical librdkafka property keys (`group.id`, +`share.acknowledgement.mode`, etc.), matching the KIP names verbatim. + +### 5.3 `ShareConsumerContext` + +Mirrors `ConsumerContext`: + +```rust +pub trait ShareConsumerContext: ClientContext + Sized { + fn acknowledgement_commit_callback( + &self, + results: &[(crate::message::OwnedHeaders /* topic */, /* partition */ i32, /* offset */ i64, Result)], + ) {} +} + +#[derive(Clone, Debug, Default)] +pub struct DefaultShareConsumerContext; +impl ClientContext for DefaultShareConsumerContext {} +impl ShareConsumerContext for DefaultShareConsumerContext {} +``` + +(Exact callback signature finalised during implementation; intent is +to surface the KIP's `AcknowledgementCommitCallback` results.) + +### 5.4 `ShareConsumer` trait + `BaseShareConsumer` + +```rust +pub trait ShareConsumer +where C: ShareConsumerContext, +{ + fn client(&self) -> &Client; + fn context(&self) -> &Arc { self.client().context() } + + fn subscribe(&self, topics: &[&str]) -> KafkaResult<()>; + fn unsubscribe(&self) -> KafkaResult<()>; + fn subscription(&self) -> KafkaResult>; + + fn poll>(&self, timeout: T) + -> KafkaResult>; + + fn acknowledge(&self, record: &ShareRecord<'_>, ack: AcknowledgeType) + -> KafkaResult<()>; + fn acknowledge_offset( + &self, + topic: &str, + partition: i32, + offset: i64, + ack: AcknowledgeType, + ) -> KafkaResult<()>; + + fn commit_sync>(&self, timeout: T) -> KafkaResult<()>; + fn commit_async(&self) -> KafkaResult<()>; + + fn client_instance_id>(&self, timeout: T) + -> KafkaResult; + + fn wakeup(&self); + fn close(&self) -> KafkaResult<()>; +} +``` + +`BaseShareConsumer` is a concrete struct that implements +`ShareConsumer`. In this PR every method body is: + +```rust +Err(KafkaError::Unsupported( + "KIP-932 share consumer support is not yet available in librdkafka; \ + see https://github.com/confluentinc/librdkafka/issues/5441", +)) +``` + +`wakeup()` and `close()` are no-ops that log a warning under the +`log` crate at `trace` level (no allocation in hot path; matches +the `tracing` opt-in pattern from `Cargo.toml`). + +`FromClientConfig` and `FromClientConfigAndContext` impls construct a +`BaseShareConsumer` from a `ClientConfig` produced by +`ShareConsumerConfig::into_client_config`. Construction returns `Ok` +so that downstream tests can exercise dispatch; only the runtime +methods return `Unsupported`. + +### 5.5 `ShareConsumerRecords` and `ShareRecord` + +Borrowed wrappers around `Vec>`. `ShareRecord` exposes: + +- `topic() -> &str` +- `partition() -> i32` +- `offset() -> i64` +- `delivery_count() -> u32` +- `key() -> Option<&[u8]>` +- `payload() -> Option<&[u8]>` +- `headers() -> Option<&BorrowedHeaders<'_>>` +- `timestamp() -> Timestamp` +- `acknowledge(&self, ack: AcknowledgeType) -> KafkaResult<()>` + +Backed by the same `BorrowedMessage`-style lifetime story as the +existing consumer. In stub form `ShareConsumerRecords::empty()` is +the only constructor reachable; `BaseShareConsumer::poll` returns +`Unsupported` rather than empty so callers don't silently hang. + +### 5.6 Error variants + +`src/error.rs`: + +```rust +pub enum KafkaError { + // ... existing variants ... + /// Returned when an API surface exists in rust-rdkafka but is not + /// yet backed by librdkafka. Currently used by the KIP-932 share + /// consumer scaffolding. + Unsupported(&'static str), +} +``` + +`Display`, `Error`, and `IsError` impls extended accordingly. +`Unsupported` reports as a non-retriable error. + +## 6. Implementation phases (commit breakdown) + +Each phase is one commit. All Conventional Commits. PR opens after +phase 1 as a draft. + +### Phase 1: `feat(consumer): add kip-932 feature flag and module skeleton` + +- Add `kip-932 = []` to `[features]` in `Cargo.toml`. +- Create `src/consumer/share/mod.rs` with empty submodules. +- Wire `pub mod share;` into `src/consumer/mod.rs` behind + `#[cfg(feature = "kip-932")]`. +- Verify `cargo build` and `cargo build --features kip-932`. + +### Phase 2: `feat(consumer/share): add AcknowledgeType and config types` + +- `AcknowledgeType` enum + `Display` + `FromStr` + tests. +- `AcknowledgementMode`, `AutoOffsetReset`, `IsolationLevel` enums. +- `ShareConsumerConfig` builder and `into_client_config`. +- Unit tests: round-trip parse, default values, key formatting. + +### Phase 3: `feat(error): add KafkaError::Unsupported variant` + +- Extend `KafkaError`, `Display`, and `IsError`. +- Single-purpose commit so reviewers see the new error surface + clearly. Used by phase 4. + +### Phase 4: `feat(consumer/share): add ShareConsumer trait and stub` + +- `ShareConsumerContext` trait + `DefaultShareConsumerContext`. +- `ShareConsumer` trait. +- `BaseShareConsumer` struct with `FromClientConfig` / + `FromClientConfigAndContext` impls and all methods returning + `KafkaError::Unsupported`. +- `ShareConsumerRecords` and `ShareRecord` skeleton types. +- Re-exports from `consumer/mod.rs`. + +### Phase 5: `test(consumer/share): public surface smoke tests` + +- `tests/test_share_consumer.rs` feature-gated on `kip-932`. +- Cases: + - Build a `ShareConsumerConfig` and convert to `ClientConfig`, + assert canonical key names. + - Create a `BaseShareConsumer` from config; assert `client()` + returns a live client. + - Assert `poll`, `acknowledge`, `commit_sync` return + `KafkaError::Unsupported`. + - `AcknowledgeType` Display / FromStr round-trip. +- Doc test on `ShareConsumer` trait that shows the intended usage + pattern and is annotated with `ignore` (`should_panic` is wrong + here; doc tests don't run under feature flags by default). + +### Phase 6: `ci: build and test rust-rdkafka with kip-932 feature` + +- Inspect `.github/workflows/*.yml` (or equivalent CI config). +- Add a matrix entry or extra step that runs: + - `cargo build --features kip-932` + - `cargo test --features kip-932` + - `cargo clippy --features kip-932 -- -D warnings` +- Add a CI lane for the no-default-features build to confirm the + default path is untouched. + +### Phase 7: `docs: document KIP-932 scaffolding and limitations` + +- `README.md`: short subsection under "Features" noting KIP-932 + scaffolding behind `kip-932` and the librdkafka tracking issue. +- `changelog.md`: entry under the next unreleased version. +- `src/lib.rs` module-level rustdoc on the share module: explain + the gating, link the KIP and librdkafka issue, set expectations. +- No new top-level docs file unless `README.md`'s structure forces it. + +## 7. Validation gate + +Run before declaring CI-green and ready-for-review: + +```bash +cargo fmt --all -- --check +cargo build +cargo build --features kip-932 +cargo build --no-default-features +cargo test +cargo test --features kip-932 +cargo clippy --all-targets -- -D warnings +cargo clippy --all-targets --features kip-932 -- -D warnings +cargo doc --no-deps --features kip-932 +``` + +If any step regresses on `master` it is fixed in-place, not +papered over. + +## 8. Tests we intentionally do not write + +- Broker integration tests against a real Kafka 4.x cluster with + share groups enabled. librdkafka cannot speak the protocol yet, + so these would be vapor tests. +- Mocked `ShareFetch` / `ShareAcknowledge` responses. The mock + surface in rdkafka-sys does not yet model share sessions. + +These tests land in the follow-up PR that wires real FFI. + +## 9. Upstream coordination + +- Open a tracking issue on `j7nw4r/rust-rdkafka` titled + "KIP-932 share consumer support (scaffolding + tracking)". Body + links the KIP, librdkafka #5441, and this PR. +- Do not file anything on `fede1024/rust-rdkafka` until the API + shape has settled and CI is green on the fork. +- Subscribe to librdkafka #5441 so we know when the public C API + ships. + +## 10. Risks and follow-ups + +| Risk | Mitigation | +|---|---| +| librdkafka's eventual public API may differ from the KIP shapes we mirror, requiring breaking changes in rust-rdkafka. | Gate the entire surface behind `kip-932` and document it as `unstable: API will change to match librdkafka when KIP-932 ships`. Tag types with `#[doc(hidden)]` notes if needed. | +| KafkaError gets a new public variant; consumers of the enum that match exhaustively will need to update. | `KafkaError` is already marked non-exhaustive in spirit (large enum). Confirm `#[non_exhaustive]` is present; add if not. | +| Feature combinations explode CI time. | Add only one new `--features kip-932` lane, not a full matrix. | +| Stub methods returning `Unsupported` get accidentally used in production by a user enabling the feature flag. | Module-level rustdoc warns loudly; `ShareConsumerConfig::into_client_config` writes a log warning at construction; the error message itself names the issue. | + +### Follow-up PRs (out of scope here) + +1. `rdkafka-sys`: regenerate bindings against librdkafka with KIP-932 + public C API, bump pinned version. +2. `consumer/share`: replace `Unsupported` returns with real FFI calls. +3. `admin`: `alter_share_group_offsets`, `delete_share_group_offsets`, + `list_share_groups`, `describe_share_groups`. +4. Integration tests against the docker-compose Kafka cluster + (requires a Kafka image with share groups enabled). +5. Stream-style async wrapper analogous to `StreamConsumer` for + share consumers (likely `ShareStreamConsumer`). + +## 11. PR shape + +- Title: `feat(consumer): scaffold KIP-932 share consumer surface` +- Draft from the start. Do not mark ready-for-review until the + validation gate passes. +- Body: 2-3 paragraphs summarising scope, the librdkafka blocker, + and the follow-up roadmap. Include the "Test plan" checklist + matching section 7. +- Target: `j7nw4r/rust-rdkafka:master`. diff --git a/README.md b/README.md index 3708c9c12..00145fd28 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,12 @@ The main features provided at the moment are: - Access to producer and consumer metrics, errors and callbacks. - Exactly-once semantics (EOS) via idempotent and transactional producers and read-committed consumers. +- Scaffolding for the KIP-932 (Queues for Kafka) share consumer surface + behind the `kip-932` cargo feature (off by default). Runtime methods + return `KafkaError::Unsupported` until librdkafka exposes the public + share consumer C API; see [librdkafka#5441][lrk-5441]. + +[lrk-5441]: https://github.com/confluentinc/librdkafka/issues/5441 ### One million messages per second diff --git a/changelog.md b/changelog.md index a7280b01d..712eac847 100644 --- a/changelog.md +++ b/changelog.md @@ -4,7 +4,19 @@ See also the [rdkafka-sys changelog](rdkafka-sys/changelog.md). ## Unreleased -None +* Add scaffolding for the KIP-932 (Queues for Kafka) share consumer + surface behind a new `kip-932` cargo feature (off by default). The + Rust types (`ShareConsumer`, `BaseShareConsumer`, `ShareConsumerConfig`, + `AcknowledgeType`, `ShareConsumerContext`, `ShareConsumerRecords`, + `ShareRecord`) compile and the configuration builder emits the + canonical librdkafka property keys, but runtime methods return the + new `KafkaError::Unsupported` variant until librdkafka ships the + public share consumer C API (tracking + [confluentinc/librdkafka#5441](https://github.com/confluentinc/librdkafka/issues/5441)). +* Add `KafkaError::Unsupported(&'static str)` for surfaces that exist + in rust-rdkafka but are not backed by the underlying librdkafka + build. `KafkaError` is `#[non_exhaustive]` so this is not a breaking + change for callers using a wildcard arm. ## 0.38.0 (2025-07-05) diff --git a/src/admin.rs b/src/admin.rs index 998053fe0..595743ef3 100644 --- a/src/admin.rs +++ b/src/admin.rs @@ -6,6 +6,7 @@ use std::collections::HashMap; use std::ffi::{c_void, CStr, CString}; +use std::fmt; use std::future::Future; use std::pin::Pin; use std::sync::atomic::{AtomicBool, Ordering}; @@ -43,6 +44,15 @@ pub struct AdminClient { handle: Option>, } +impl fmt::Debug for AdminClient { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut debug = f.debug_struct("AdminClient"); + debug.field("has_handle", &self.handle.is_some()); + debug.field("stop_requested", &self.should_stop.load(Ordering::Relaxed)); + debug.finish() + } +} + impl AdminClient { /// Creates new topics according to the provided `NewTopic` specifications. /// diff --git a/src/config.rs b/src/config.rs index 04d6bde6f..c0217094d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -22,8 +22,9 @@ //! //! [librdkafka-config]: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md -use std::collections::HashMap; +use std::collections::{BTreeMap, HashMap}; use std::ffi::CString; +use std::fmt::Debug; use std::iter::FromIterator; use std::os::raw::c_char; use std::ptr; @@ -36,6 +37,16 @@ use crate::error::{IsError, KafkaError, KafkaResult}; use crate::log::{log_enabled, DEBUG, INFO, WARN}; use crate::util::{ErrBuf, KafkaDrop, NativePtr}; +const SENSITIVE_CONFIG_KEYS: &[&str] = &[ + "sasl.password", + "ssl.key.password", + "ssl.keystore.password", + "ssl.truststore.password", + "sasl.oauthbearer.client.secret", +]; + +const SANITIZED_VALUE_PLACEHOLDER: &str = "[sanitized for safety]"; + /// The log levels supported by librdkafka. #[derive(Copy, Clone, Debug)] pub enum RDKafkaLogLevel { @@ -181,7 +192,7 @@ impl NativeClientConfig { } /// Client configuration. -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct ClientConfig { conf_map: HashMap, /// The librdkafka logging level. Refer to [`RDKafkaLogLevel`] for the list @@ -189,6 +200,27 @@ pub struct ClientConfig { pub log_level: RDKafkaLogLevel, } +impl Debug for ClientConfig { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let sanitized: BTreeMap<&str, &str> = self + .conf_map + .iter() + .filter_map(|(key, value)| { + if SENSITIVE_CONFIG_KEYS.contains(&key.as_str()) { + None + } else { + Some((key.as_str(), value.as_str())) + } + }) + .collect(); + + let mut debug_struct = f.debug_struct("ClientConfig"); + debug_struct.field("log_level", &self.log_level); + debug_struct.field("conf_map", &sanitized); + debug_struct.finish() + } +} + impl Default for ClientConfig { fn default() -> Self { Self::new() @@ -204,9 +236,21 @@ impl ClientConfig { } } - /// Gets a reference to the underlying config map - pub fn config_map(&self) -> &HashMap { - &self.conf_map + /// Returns a sanitized view of the underlying config map. + /// + /// Sensitive keys have their values replaced with a placeholder string so they never appear in + /// clear text when inspected. + pub fn config_map(&self) -> BTreeMap<&str, &str> { + self.conf_map + .iter() + .map(|(key, value)| { + if SENSITIVE_CONFIG_KEYS.contains(&key.as_str()) { + (key.as_str(), SANITIZED_VALUE_PLACEHOLDER) + } else { + (key.as_str(), value.as_str()) + } + }) + .collect() } /// Gets the value of a parameter in the configuration. diff --git a/src/consumer/mod.rs b/src/consumer/mod.rs index 5ce8b05b1..d8cb61085 100644 --- a/src/consumer/mod.rs +++ b/src/consumer/mod.rs @@ -19,12 +19,23 @@ use crate::util::{KafkaDrop, NativePtr, Timeout}; pub mod base_consumer; pub mod stream_consumer; +#[cfg(feature = "kip-932")] +pub mod share; + // Re-exports. #[doc(inline)] pub use self::base_consumer::BaseConsumer; #[doc(inline)] pub use self::stream_consumer::{MessageStream, StreamConsumer}; +#[cfg(feature = "kip-932")] +#[doc(inline)] +pub use self::share::{ + AcknowledgeType, AcknowledgementCommitResult, AcknowledgementMode, AutoOffsetReset, + BaseShareConsumer, DefaultShareConsumerContext, IsolationLevel, ShareConsumer, + ShareConsumerConfig, ShareConsumerContext, ShareConsumerRecords, ShareRecord, +}; + /// Rebalance information. #[derive(Clone, Debug)] pub enum Rebalance<'a> { diff --git a/src/consumer/share/acknowledge.rs b/src/consumer/share/acknowledge.rs new file mode 100644 index 000000000..2b4460a5a --- /dev/null +++ b/src/consumer/share/acknowledge.rs @@ -0,0 +1,136 @@ +//! Acknowledgement primitives for the share consumer surface. +//! +//! See [`KIP-932`][kip] for the semantics of each acknowledgement type +//! and the record state machine they drive. +//! +//! [kip]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka + +use std::fmt; +use std::str::FromStr; + +use crate::error::KafkaError; + +/// The disposition of a record delivered to a share consumer. +/// +/// Drives the broker-side record state machine described by KIP-932. +/// `Accept` moves the record to the Acknowledged state, `Release` returns +/// it to Available for redelivery (subject to the delivery attempt limit), +/// and `Reject` archives it (poison message). +#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] +pub enum AcknowledgeType { + /// The record was processed successfully. + Accept, + /// The record could not be processed but may be retried. Returns the + /// record to the Available state for redelivery. + Release, + /// The record cannot be processed (poison message). Moves the record + /// to the Archived state. + Reject, +} + +impl AcknowledgeType { + /// Returns the canonical lowercase wire name (`"accept"`, `"release"`, + /// `"reject"`). + pub fn as_str(&self) -> &'static str { + match self { + AcknowledgeType::Accept => "accept", + AcknowledgeType::Release => "release", + AcknowledgeType::Reject => "reject", + } + } +} + +impl fmt::Display for AcknowledgeType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +/// Error returned when [`AcknowledgeType::from_str`] receives an +/// unrecognised string. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ParseAcknowledgeTypeError(String); + +impl fmt::Display for ParseAcknowledgeTypeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "unknown acknowledge type {:?}; expected accept, release, or reject", + self.0 + ) + } +} + +impl std::error::Error for ParseAcknowledgeTypeError {} + +impl FromStr for AcknowledgeType { + type Err = ParseAcknowledgeTypeError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "accept" => Ok(AcknowledgeType::Accept), + "release" => Ok(AcknowledgeType::Release), + "reject" => Ok(AcknowledgeType::Reject), + _ => Err(ParseAcknowledgeTypeError(s.to_owned())), + } + } +} + +/// Result of an asynchronous acknowledgement commit, reported to +/// [`ShareConsumerContext::acknowledgement_commit`][cb]. +/// +/// [cb]: crate::consumer::share::ShareConsumerContext::acknowledgement_commit +#[derive(Clone, Debug)] +pub struct AcknowledgementCommitResult { + /// Topic of the acknowledged record. + pub topic: String, + /// Partition of the acknowledged record. + pub partition: i32, + /// Offset of the acknowledged record. + pub offset: i64, + /// The acknowledgement disposition, or the error reported by the + /// broker. + pub result: Result, +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn display_round_trip() { + for ack in [ + AcknowledgeType::Accept, + AcknowledgeType::Release, + AcknowledgeType::Reject, + ] { + let s = ack.to_string(); + assert_eq!(s.parse::().unwrap(), ack); + } + } + + #[test] + fn parse_is_case_insensitive() { + assert_eq!( + "ACCEPT".parse::().unwrap(), + AcknowledgeType::Accept + ); + assert_eq!( + "Release".parse::().unwrap(), + AcknowledgeType::Release + ); + } + + #[test] + fn parse_rejects_unknown() { + let err = "discard".parse::().unwrap_err(); + assert!(err.to_string().contains("discard")); + } + + #[test] + fn as_str_matches_display() { + assert_eq!(AcknowledgeType::Accept.as_str(), "accept"); + assert_eq!(AcknowledgeType::Release.to_string(), "release"); + assert_eq!(AcknowledgeType::Reject.as_str(), "reject"); + } +} diff --git a/src/consumer/share/config.rs b/src/consumer/share/config.rs new file mode 100644 index 000000000..927c180be --- /dev/null +++ b/src/consumer/share/config.rs @@ -0,0 +1,418 @@ +//! Share consumer configuration builder. +//! +//! Wraps a [`ClientConfig`] and exposes typed setters for the new +//! KIP-932 properties. Call [`ShareConsumerConfig::into_client_config`] +//! to obtain a [`ClientConfig`] suitable for +//! [`BaseShareConsumer::from_config`][bsc]. +//! +//! Property keys are written verbatim from KIP-932 so that, once +//! librdkafka exposes a public share consumer API, the existing +//! configuration plumbing flows through unchanged. +//! +//! [bsc]: crate::consumer::share::BaseShareConsumer + +use std::fmt; +use std::str::FromStr; + +use crate::config::ClientConfig; + +/// Whether the share consumer auto-acknowledges records on the next +/// `poll` (`Implicit`) or requires the application to acknowledge each +/// record explicitly (`Explicit`). +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +pub enum AcknowledgementMode { + /// The next call to `poll` or `commit_sync`/`commit_async` + /// acknowledges every record returned by the previous `poll`. + #[default] + Implicit, + /// The application must call `acknowledge` on every record before + /// the next `poll`, otherwise an error is raised. + Explicit, +} + +impl AcknowledgementMode { + /// Returns the canonical librdkafka string for this mode. + pub fn as_str(&self) -> &'static str { + match self { + AcknowledgementMode::Implicit => "implicit", + AcknowledgementMode::Explicit => "explicit", + } + } +} + +impl fmt::Display for AcknowledgementMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for AcknowledgementMode { + type Err = ParseConfigEnumError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "implicit" => Ok(AcknowledgementMode::Implicit), + "explicit" => Ok(AcknowledgementMode::Explicit), + _ => Err(ParseConfigEnumError { + value: s.to_owned(), + expected: "implicit or explicit", + }), + } + } +} + +/// Initial position for the share-partition start offset (SPSO) when a +/// share group reads a topic for the first time. +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +pub enum AutoOffsetReset { + /// Start from the earliest available offset. + Earliest, + /// Start from the latest (next-produced) offset. + #[default] + Latest, +} + +impl AutoOffsetReset { + /// Returns the canonical librdkafka string for this reset policy. + pub fn as_str(&self) -> &'static str { + match self { + AutoOffsetReset::Earliest => "earliest", + AutoOffsetReset::Latest => "latest", + } + } +} + +impl fmt::Display for AutoOffsetReset { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for AutoOffsetReset { + type Err = ParseConfigEnumError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "earliest" => Ok(AutoOffsetReset::Earliest), + "latest" => Ok(AutoOffsetReset::Latest), + _ => Err(ParseConfigEnumError { + value: s.to_owned(), + expected: "earliest or latest", + }), + } + } +} + +/// Per-share-group transactional isolation. Applies to the entire share +/// group, not per consumer. +#[derive(Clone, Copy, Debug, Default, Eq, Hash, PartialEq)] +pub enum IsolationLevel { + /// All produced records are visible (including aborted transactions). + #[default] + ReadUncommitted, + /// Only committed records are visible; aborted records are filtered + /// out by the broker. + ReadCommitted, +} + +impl IsolationLevel { + /// Returns the canonical librdkafka string for this isolation level. + pub fn as_str(&self) -> &'static str { + match self { + IsolationLevel::ReadUncommitted => "read_uncommitted", + IsolationLevel::ReadCommitted => "read_committed", + } + } +} + +impl fmt::Display for IsolationLevel { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for IsolationLevel { + type Err = ParseConfigEnumError; + + fn from_str(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "read_uncommitted" => Ok(IsolationLevel::ReadUncommitted), + "read_committed" => Ok(IsolationLevel::ReadCommitted), + _ => Err(ParseConfigEnumError { + value: s.to_owned(), + expected: "read_uncommitted or read_committed", + }), + } + } +} + +/// Error returned by the share-config enum [`FromStr`] impls when the +/// input does not match any known variant. +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct ParseConfigEnumError { + value: String, + expected: &'static str, +} + +impl fmt::Display for ParseConfigEnumError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "unknown share config value {:?}; expected {}", + self.value, self.expected + ) + } +} + +impl std::error::Error for ParseConfigEnumError {} + +/// Typed builder for the configuration values introduced by KIP-932. +/// +/// The builder is layered on top of [`ClientConfig`]: KIP-932 specific +/// fields are typed; arbitrary librdkafka properties (for example +/// `bootstrap.servers`) flow through the embedded [`ClientConfig`]. Call +/// [`ShareConsumerConfig::into_client_config`] to materialise the final +/// [`ClientConfig`] understood by +/// [`BaseShareConsumer::from_config`][bsc]. +/// +/// All field names match the KIP verbatim; canonical librdkafka key +/// formatting is applied by [`Self::into_client_config`]. +/// +/// [bsc]: crate::consumer::share::BaseShareConsumer +#[derive(Clone, Debug)] +pub struct ShareConsumerConfig { + base: ClientConfig, + group_id: Option, + acknowledgement_mode: AcknowledgementMode, + auto_offset_reset: AutoOffsetReset, + isolation_level: IsolationLevel, + delivery_attempt_limit: u32, + record_lock_duration_ms: u32, + heartbeat_interval_ms: Option, + session_timeout_ms: Option, +} + +impl ShareConsumerConfig { + /// KIP-932 default for `group.share.delivery.attempt.limit`. + pub const DEFAULT_DELIVERY_ATTEMPT_LIMIT: u32 = 5; + /// KIP-932 default for `group.share.record.lock.duration.ms`. + pub const DEFAULT_RECORD_LOCK_DURATION_MS: u32 = 30_000; + + /// Creates an empty configuration with KIP-932 defaults. + pub fn new() -> Self { + Self { + base: ClientConfig::new(), + group_id: None, + acknowledgement_mode: AcknowledgementMode::default(), + auto_offset_reset: AutoOffsetReset::default(), + isolation_level: IsolationLevel::default(), + delivery_attempt_limit: Self::DEFAULT_DELIVERY_ATTEMPT_LIMIT, + record_lock_duration_ms: Self::DEFAULT_RECORD_LOCK_DURATION_MS, + heartbeat_interval_ms: None, + session_timeout_ms: None, + } + } + + /// Sets the share group identifier (`group.id`). Required. + pub fn group_id(&mut self, group_id: impl Into) -> &mut Self { + self.group_id = Some(group_id.into()); + self + } + + /// Sets `share.acknowledgement.mode`. + pub fn acknowledgement_mode(&mut self, mode: AcknowledgementMode) -> &mut Self { + self.acknowledgement_mode = mode; + self + } + + /// Sets `share.auto.offset.reset`. + pub fn auto_offset_reset(&mut self, reset: AutoOffsetReset) -> &mut Self { + self.auto_offset_reset = reset; + self + } + + /// Sets `share.isolation.level`. + pub fn isolation_level(&mut self, level: IsolationLevel) -> &mut Self { + self.isolation_level = level; + self + } + + /// Sets `group.share.delivery.attempt.limit` (default `5`). + pub fn delivery_attempt_limit(&mut self, limit: u32) -> &mut Self { + self.delivery_attempt_limit = limit; + self + } + + /// Sets `group.share.record.lock.duration.ms` (default `30000`). + pub fn record_lock_duration_ms(&mut self, ms: u32) -> &mut Self { + self.record_lock_duration_ms = ms; + self + } + + /// Sets `group.share.heartbeat.interval.ms`. Leave unset to use the + /// broker default. + pub fn heartbeat_interval_ms(&mut self, ms: u32) -> &mut Self { + self.heartbeat_interval_ms = Some(ms); + self + } + + /// Sets `group.share.session.timeout.ms`. Leave unset to use the + /// broker default. + pub fn session_timeout_ms(&mut self, ms: u32) -> &mut Self { + self.session_timeout_ms = Some(ms); + self + } + + /// Sets an arbitrary librdkafka property on the embedded + /// [`ClientConfig`] (for example `bootstrap.servers`). + pub fn set(&mut self, key: K, value: V) -> &mut Self + where + K: Into, + V: Into, + { + self.base.set(key, value); + self + } + + /// Materialises the final [`ClientConfig`]. + /// + /// Returns `None` if no `group_id` was set, since KIP-932 requires a + /// share group identifier and downstream construction would + /// otherwise reject the configuration with an unhelpful error. + pub fn into_client_config(self) -> Option { + let group_id = self.group_id?; + let mut config = self.base; + config.set("group.id", group_id); + config.set( + "share.acknowledgement.mode", + self.acknowledgement_mode.as_str(), + ); + config.set("share.auto.offset.reset", self.auto_offset_reset.as_str()); + config.set("share.isolation.level", self.isolation_level.as_str()); + config.set( + "group.share.delivery.attempt.limit", + self.delivery_attempt_limit.to_string(), + ); + config.set( + "group.share.record.lock.duration.ms", + self.record_lock_duration_ms.to_string(), + ); + if let Some(ms) = self.heartbeat_interval_ms { + config.set("group.share.heartbeat.interval.ms", ms.to_string()); + } + if let Some(ms) = self.session_timeout_ms { + config.set("group.share.session.timeout.ms", ms.to_string()); + } + Some(config) + } +} + +impl Default for ShareConsumerConfig { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_match_kip() { + let config = ShareConsumerConfig::new(); + assert_eq!(config.acknowledgement_mode, AcknowledgementMode::Implicit); + assert_eq!(config.auto_offset_reset, AutoOffsetReset::Latest); + assert_eq!(config.isolation_level, IsolationLevel::ReadUncommitted); + assert_eq!(config.delivery_attempt_limit, 5); + assert_eq!(config.record_lock_duration_ms, 30_000); + assert!(config.heartbeat_interval_ms.is_none()); + assert!(config.session_timeout_ms.is_none()); + } + + #[test] + fn into_client_config_requires_group_id() { + assert!(ShareConsumerConfig::new().into_client_config().is_none()); + } + + #[test] + fn into_client_config_writes_canonical_keys() { + let mut config = ShareConsumerConfig::new(); + config + .group_id("shares-1") + .acknowledgement_mode(AcknowledgementMode::Explicit) + .auto_offset_reset(AutoOffsetReset::Earliest) + .isolation_level(IsolationLevel::ReadCommitted) + .delivery_attempt_limit(7) + .record_lock_duration_ms(45_000) + .heartbeat_interval_ms(5_000) + .session_timeout_ms(60_000) + .set("bootstrap.servers", "localhost:9092"); + + let client_config = config.into_client_config().expect("group.id set"); + let map = client_config.config_map(); + assert_eq!(map.get("group.id").copied(), Some("shares-1")); + assert_eq!( + map.get("share.acknowledgement.mode").copied(), + Some("explicit") + ); + assert_eq!( + map.get("share.auto.offset.reset").copied(), + Some("earliest") + ); + assert_eq!( + map.get("share.isolation.level").copied(), + Some("read_committed") + ); + assert_eq!( + map.get("group.share.delivery.attempt.limit").copied(), + Some("7") + ); + assert_eq!( + map.get("group.share.record.lock.duration.ms").copied(), + Some("45000") + ); + assert_eq!( + map.get("group.share.heartbeat.interval.ms").copied(), + Some("5000") + ); + assert_eq!( + map.get("group.share.session.timeout.ms").copied(), + Some("60000") + ); + assert_eq!( + map.get("bootstrap.servers").copied(), + Some("localhost:9092") + ); + } + + #[test] + fn enum_parse_round_trip() { + for mode in [AcknowledgementMode::Implicit, AcknowledgementMode::Explicit] { + assert_eq!( + mode.to_string().parse::().unwrap(), + mode + ); + } + for reset in [AutoOffsetReset::Earliest, AutoOffsetReset::Latest] { + assert_eq!(reset.to_string().parse::().unwrap(), reset); + } + for level in [ + IsolationLevel::ReadUncommitted, + IsolationLevel::ReadCommitted, + ] { + assert_eq!(level.to_string().parse::().unwrap(), level); + } + } + + #[test] + fn enum_parse_is_case_insensitive() { + assert_eq!( + "READ_COMMITTED".parse::().unwrap(), + IsolationLevel::ReadCommitted + ); + assert_eq!( + "Earliest".parse::().unwrap(), + AutoOffsetReset::Earliest + ); + } +} diff --git a/src/consumer/share/consumer.rs b/src/consumer/share/consumer.rs new file mode 100644 index 000000000..50d2f81ba --- /dev/null +++ b/src/consumer/share/consumer.rs @@ -0,0 +1,277 @@ +//! Share consumer trait and stub implementation. +//! +//! The runtime methods on [`BaseShareConsumer`] return +//! [`KafkaError::Unsupported`] until librdkafka exposes the share +//! consumer C API; see +//! . + +use std::fmt; +use std::sync::Arc; + +use log::trace; + +use crate::config::{ClientConfig, FromClientConfig, FromClientConfigAndContext}; +use crate::consumer::share::acknowledge::AcknowledgeType; +use crate::consumer::share::context::{DefaultShareConsumerContext, ShareConsumerContext}; +use crate::consumer::share::records::{ShareConsumerRecords, ShareRecord}; +use crate::error::{KafkaError, KafkaResult}; +use crate::util::Timeout; + +const UNSUPPORTED_REASON: &str = + "KIP-932 share consumer support is not yet available in librdkafka; \ + see https://github.com/confluentinc/librdkafka/issues/5441"; + +fn unsupported() -> KafkaResult { + Err(KafkaError::Unsupported(UNSUPPORTED_REASON)) +} + +/// Common trait for share consumers. +/// +/// Mirrors the [`KafkaShareConsumer`][java] surface from KIP-932, +/// adapted to idiomatic Rust. All methods are non-blocking apart from +/// [`Self::poll`] and [`Self::commit_sync`], which respect the supplied +/// timeout. +/// +/// # Stability +/// +/// `ShareConsumer` is gated behind the `kip-932` cargo feature and +/// tracks the KIP's `@InterfaceStability.Evolving` posture. Method +/// shapes will change to match librdkafka once it exposes the public C +/// API. +/// +/// [java]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka +pub trait ShareConsumer +where + C: ShareConsumerContext, +{ + /// Returns the consumer context. + fn context(&self) -> &Arc; + + /// Subscribes the consumer to a list of topics. The broker assigns + /// partitions automatically; explicit assignment is not supported + /// for share consumers. + fn subscribe(&self, topics: &[&str]) -> KafkaResult<()>; + + /// Unsubscribes the consumer from all topics. + fn unsubscribe(&self) -> KafkaResult<()>; + + /// Returns the current subscription list. + fn subscription(&self) -> KafkaResult>; + + /// Fetches the next batch of acquired records. + fn poll>(&self, timeout: T) -> KafkaResult>; + + /// Acknowledges a record with the given disposition. + fn acknowledge(&self, record: &ShareRecord<'_>, ack: AcknowledgeType) -> KafkaResult<()>; + + /// Acknowledges a record by (topic, partition, offset). Use when + /// the original [`ShareRecord`] is no longer in scope (for example + /// after a deserialization error). + fn acknowledge_offset( + &self, + topic: &str, + partition: i32, + offset: i64, + ack: AcknowledgeType, + ) -> KafkaResult<()>; + + /// Synchronously commits all pending acknowledgements. Blocks until + /// the broker responds or the timeout elapses. + fn commit_sync>(&self, timeout: T) -> KafkaResult<()>; + + /// Asynchronously commits all pending acknowledgements. Results are + /// reported through + /// [`ShareConsumerContext::acknowledgement_commit`][cb]. + /// + /// [cb]: crate::consumer::share::ShareConsumerContext::acknowledgement_commit + fn commit_async(&self) -> KafkaResult<()>; + + /// Interrupts an in-progress [`Self::poll`] from another thread. + /// + /// Safe to call from any thread; the rest of the consumer is not + /// thread-safe. + fn wakeup(&self); + + /// Releases acquired records, commits pending acknowledgements, and + /// leaves the share group. + fn close(&self) -> KafkaResult<()>; +} + +/// Low-level share consumer. +/// +/// Today this is a scaffolding stub: construction parses the +/// configuration and stores the context, but every runtime method +/// returns [`KafkaError::Unsupported`]. See the [module docs][m] for +/// the upstream tracking issue. +/// +/// [m]: crate::consumer::share +pub struct BaseShareConsumer +where + C: ShareConsumerContext, +{ + context: Arc, + group_id: String, +} + +impl fmt::Debug for BaseShareConsumer +where + C: ShareConsumerContext, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("BaseShareConsumer") + .field("group_id", &self.group_id) + .finish_non_exhaustive() + } +} + +impl BaseShareConsumer +where + C: ShareConsumerContext, +{ + fn new(config: &ClientConfig, context: C) -> KafkaResult { + let group_id = config + .get("group.id") + .ok_or_else(|| { + KafkaError::ClientCreation("share consumer requires group.id to be set".to_owned()) + })? + .to_owned(); + Ok(Self { + context: Arc::new(context), + group_id, + }) + } + + /// Returns the configured share group identifier. + pub fn group_id(&self) -> &str { + &self.group_id + } +} + +impl FromClientConfig for BaseShareConsumer { + fn from_config(config: &ClientConfig) -> KafkaResult { + BaseShareConsumer::new(config, DefaultShareConsumerContext) + } +} + +impl FromClientConfigAndContext for BaseShareConsumer +where + C: ShareConsumerContext, +{ + fn from_config_and_context(config: &ClientConfig, context: C) -> KafkaResult { + BaseShareConsumer::new(config, context) + } +} + +impl ShareConsumer for BaseShareConsumer +where + C: ShareConsumerContext, +{ + fn context(&self) -> &Arc { + &self.context + } + + fn subscribe(&self, _topics: &[&str]) -> KafkaResult<()> { + unsupported() + } + + fn unsubscribe(&self) -> KafkaResult<()> { + unsupported() + } + + fn subscription(&self) -> KafkaResult> { + unsupported() + } + + fn poll>(&self, _timeout: T) -> KafkaResult> { + unsupported() + } + + fn acknowledge(&self, _record: &ShareRecord<'_>, _ack: AcknowledgeType) -> KafkaResult<()> { + unsupported() + } + + fn acknowledge_offset( + &self, + _topic: &str, + _partition: i32, + _offset: i64, + _ack: AcknowledgeType, + ) -> KafkaResult<()> { + unsupported() + } + + fn commit_sync>(&self, _timeout: T) -> KafkaResult<()> { + unsupported() + } + + fn commit_async(&self) -> KafkaResult<()> { + unsupported() + } + + fn wakeup(&self) { + trace!("BaseShareConsumer::wakeup is a no-op until librdkafka exposes KIP-932"); + } + + fn close(&self) -> KafkaResult<()> { + trace!("BaseShareConsumer::close is a no-op until librdkafka exposes KIP-932"); + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::consumer::share::config::ShareConsumerConfig; + + fn make_config() -> ClientConfig { + let mut config = ShareConsumerConfig::new(); + config.group_id("share-test"); + config.into_client_config().expect("group.id set") + } + + #[test] + fn construction_requires_group_id() { + let config = ClientConfig::new(); + let err = BaseShareConsumer::::from_config(&config) + .expect_err("missing group.id should fail"); + match err { + KafkaError::ClientCreation(msg) => assert!(msg.contains("group.id")), + other => panic!("expected ClientCreation, got {other:?}"), + } + } + + #[test] + fn construction_succeeds_with_group_id() { + let config = make_config(); + let consumer = BaseShareConsumer::::from_config(&config) + .expect("config has group.id"); + assert_eq!(consumer.group_id(), "share-test"); + } + + fn assert_unsupported(result: KafkaResult) { + match result { + Err(KafkaError::Unsupported(_)) => {} + other => panic!("expected Unsupported, got {other:?}"), + } + } + + #[test] + fn runtime_methods_return_unsupported() { + let consumer = + BaseShareConsumer::::from_config(&make_config()) + .expect("construction"); + + assert_unsupported(consumer.subscribe(&["topic"])); + assert_unsupported(consumer.unsubscribe()); + assert_unsupported(consumer.subscription()); + assert_unsupported(consumer.poll(std::time::Duration::from_millis(0))); + assert_unsupported(consumer.acknowledge_offset("topic", 0, 42, AcknowledgeType::Accept)); + assert_unsupported(consumer.commit_sync(std::time::Duration::from_millis(0))); + assert_unsupported(consumer.commit_async()); + + // wakeup/close are not unsupported, but they are explicitly safe + // no-ops in the stub. + consumer.wakeup(); + consumer.close().expect("close stub is Ok"); + } +} diff --git a/src/consumer/share/context.rs b/src/consumer/share/context.rs new file mode 100644 index 000000000..485de1553 --- /dev/null +++ b/src/consumer/share/context.rs @@ -0,0 +1,36 @@ +//! Share consumer context trait. +//! +//! Mirrors [`ConsumerContext`][cc] for the KIP-932 surface: lets the +//! application observe asynchronous events without owning the consumer +//! poll loop. The single callback today, +//! [`ShareConsumerContext::acknowledgement_commit`], surfaces the result +//! of asynchronous acknowledgement commits. +//! +//! [cc]: crate::consumer::ConsumerContext + +use crate::client::ClientContext; +use crate::consumer::share::acknowledge::AcknowledgementCommitResult; + +/// Per-consumer callback surface for KIP-932 share consumers. +/// +/// The default implementations are intentionally empty so downstream +/// crates can override only what they need. +pub trait ShareConsumerContext: ClientContext + Sized { + /// Called when an asynchronous acknowledgement commit returns. + /// + /// The slice contains one entry per acknowledged record from the + /// associated [`commit_async`][ca] call. The callback runs on a + /// librdkafka-owned thread; do not call back into the consumer here + /// (other than `wakeup`). + /// + /// [ca]: crate::consumer::share::ShareConsumer::commit_async + #[allow(unused_variables)] + fn acknowledgement_commit(&self, results: &[AcknowledgementCommitResult]) {} +} + +/// Inert [`ShareConsumerContext`] for callers that need no callbacks. +#[derive(Clone, Debug, Default)] +pub struct DefaultShareConsumerContext; + +impl ClientContext for DefaultShareConsumerContext {} +impl ShareConsumerContext for DefaultShareConsumerContext {} diff --git a/src/consumer/share/mod.rs b/src/consumer/share/mod.rs new file mode 100644 index 000000000..c3631f42a --- /dev/null +++ b/src/consumer/share/mod.rs @@ -0,0 +1,40 @@ +//! KIP-932 share consumers (Queues for Kafka). +//! +//! This module hosts the public Rust surface for the share consumer API +//! introduced by [KIP-932][kip]. Share consumers cooperatively consume +//! records from a share group, with broker-tracked per-record locks and +//! explicit acknowledgement semantics. +//! +//! # Stability +//! +//! The KIP marks the Java API as `@InterfaceStability.Evolving`. The Rust +//! surface mirrors that posture: the entire `share` module is gated behind +//! the `kip-932` cargo feature and may change in any minor release while +//! the underlying protocol stabilises. +//! +//! # Runtime support +//! +//! librdkafka does not yet expose a public C API for share consumers. +//! Construction of a [`BaseShareConsumer`] succeeds so downstream code +//! can exercise the type surface, but every runtime method returns +//! [`KafkaError::Unsupported`][unsup]. Track librdkafka progress at +//! . +//! +//! [kip]: https://cwiki.apache.org/confluence/display/KAFKA/KIP-932%3A+Queues+for+Kafka +//! [unsup]: crate::error::KafkaError::Unsupported + +mod acknowledge; +mod config; +mod consumer; +mod context; +mod records; + +pub use self::acknowledge::{ + AcknowledgeType, AcknowledgementCommitResult, ParseAcknowledgeTypeError, +}; +pub use self::config::{ + AcknowledgementMode, AutoOffsetReset, IsolationLevel, ParseConfigEnumError, ShareConsumerConfig, +}; +pub use self::consumer::{BaseShareConsumer, ShareConsumer}; +pub use self::context::{DefaultShareConsumerContext, ShareConsumerContext}; +pub use self::records::{ShareConsumerRecords, ShareRecord}; diff --git a/src/consumer/share/records.rs b/src/consumer/share/records.rs new file mode 100644 index 000000000..6f2fadf73 --- /dev/null +++ b/src/consumer/share/records.rs @@ -0,0 +1,136 @@ +//! Borrowed record wrappers returned from share consumer polls. +//! +//! The types model the KIP-932 fetch result: a batch of records carrying +//! per-record metadata (offset, partition, delivery count) plus borrowed +//! key/payload bytes. Today the producer-side stub never yields records, +//! but the lifetime story is established so the real implementation can +//! slot in without breaking the API. + +use std::marker::PhantomData; +use std::slice; + +use crate::consumer::share::acknowledge::AcknowledgeType; +use crate::error::KafkaError; + +/// A batch of records returned from +/// [`ShareConsumer::poll`][poll]. +/// +/// Today the stub never produces a non-empty batch; once librdkafka +/// exposes a share fetch API the borrowed payload and key slices will +/// reference librdkafka-owned memory tied to the consumer's lifetime. +/// +/// [poll]: crate::consumer::share::ShareConsumer::poll +#[derive(Debug)] +pub struct ShareConsumerRecords<'a> { + records: Vec>, +} + +impl<'a> ShareConsumerRecords<'a> { + /// Returns an empty batch. Used by the stub implementation and by + /// tests that need a placeholder. + pub fn empty() -> Self { + Self { + records: Vec::new(), + } + } + + /// Returns `true` if the batch contains no records. + pub fn is_empty(&self) -> bool { + self.records.is_empty() + } + + /// Returns the number of records in the batch. + pub fn len(&self) -> usize { + self.records.len() + } + + /// Borrows the records in iteration order. + pub fn iter(&self) -> slice::Iter<'_, ShareRecord<'a>> { + self.records.iter() + } +} + +impl<'a, 'b> IntoIterator for &'b ShareConsumerRecords<'a> { + type Item = &'b ShareRecord<'a>; + type IntoIter = slice::Iter<'b, ShareRecord<'a>>; + + fn into_iter(self) -> Self::IntoIter { + self.iter() + } +} + +/// A single record returned from +/// [`ShareConsumer::poll`][poll]. +/// +/// Carries the KIP-932 per-record metadata (topic, partition, offset, +/// delivery count) plus borrowed key/payload bytes. +/// +/// [poll]: crate::consumer::share::ShareConsumer::poll +#[derive(Debug)] +pub struct ShareRecord<'a> { + topic: String, + partition: i32, + offset: i64, + delivery_count: u32, + key: Option<&'a [u8]>, + payload: Option<&'a [u8]>, + _phantom: PhantomData<&'a ()>, +} + +impl ShareRecord<'_> { + /// Topic of the record. + pub fn topic(&self) -> &str { + &self.topic + } + + /// Partition of the record. + pub fn partition(&self) -> i32 { + self.partition + } + + /// Offset of the record within its partition. + pub fn offset(&self) -> i64 { + self.offset + } + + /// Number of times the broker has delivered this record to a share + /// consumer in the group; `1` on the first delivery. + pub fn delivery_count(&self) -> u32 { + self.delivery_count + } + + /// Borrowed key bytes, if the record carries a key. + pub fn key(&self) -> Option<&[u8]> { + self.key + } + + /// Borrowed payload bytes, if the record carries a payload. + pub fn payload(&self) -> Option<&[u8]> { + self.payload + } + + /// Acknowledges this record with the given disposition. + /// + /// Returns [`KafkaError::Unsupported`] until librdkafka exposes the + /// share consumer C API. + #[allow(unused_variables)] + pub fn acknowledge(&self, ack: AcknowledgeType) -> Result<(), KafkaError> { + Err(KafkaError::Unsupported( + "KIP-932 share consumer support is not yet available in librdkafka; \ + see https://github.com/confluentinc/librdkafka/issues/5441", + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_batch_is_empty() { + let batch = ShareConsumerRecords::empty(); + assert!(batch.is_empty()); + assert_eq!(batch.len(), 0); + assert_eq!(batch.iter().count(), 0); + } +} diff --git a/src/error.rs b/src/error.rs index 634fe9eee..b853e6aac 100644 --- a/src/error.rs +++ b/src/error.rs @@ -187,6 +187,11 @@ pub enum KafkaError { Transaction(RDKafkaError), /// Mock Cluster error MockCluster(RDKafkaErrorCode), + /// An API surface exists in rust-rdkafka but is not yet backed by the + /// underlying librdkafka build. Returned today by the KIP-932 share + /// consumer stubs; see + /// . + Unsupported(&'static str), } impl fmt::Debug for KafkaError { @@ -248,6 +253,7 @@ impl fmt::Debug for KafkaError { } KafkaError::Transaction(err) => write!(f, "KafkaError (Transaction error: {})", err), KafkaError::MockCluster(err) => write!(f, "KafkaError (Mock cluster error: {})", err), + KafkaError::Unsupported(reason) => write!(f, "KafkaError (Unsupported: {})", reason), } } } @@ -289,6 +295,7 @@ impl fmt::Display for KafkaError { KafkaError::Subscription(err) => write!(f, "Subscription error: {}", err), KafkaError::Transaction(err) => write!(f, "Transaction error: {}", err), KafkaError::MockCluster(err) => write!(f, "Mock cluster error: {}", err), + KafkaError::Unsupported(reason) => write!(f, "Unsupported: {}", reason), } } } @@ -322,6 +329,7 @@ impl Error for KafkaError { KafkaError::Subscription(_) => None, KafkaError::Transaction(err) => Some(err), KafkaError::MockCluster(err) => Some(err), + KafkaError::Unsupported(_) => None, } } } @@ -363,6 +371,7 @@ impl KafkaError { KafkaError::Subscription(_) => None, KafkaError::Transaction(err) => Some(err.code()), KafkaError::MockCluster(err) => Some(*err), + KafkaError::Unsupported(_) => None, } } } diff --git a/src/lib.rs b/src/lib.rs index 21f79298e..ad97a430b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -33,6 +33,11 @@ //! - Access to producer and consumer metrics, errors and callbacks. //! - Exactly-once semantics (EOS) via idempotent and transactional producers //! and read-committed consumers. +//! - Scaffolding for the KIP-932 (Queues for Kafka) share consumer surface +//! behind the `kip-932` cargo feature (off by default). Runtime methods +//! return [`error::KafkaError::Unsupported`] until librdkafka exposes the public +//! share consumer C API; see +//! . //! //! ### One million messages per second //! diff --git a/tests/test_share_consumer.rs b/tests/test_share_consumer.rs new file mode 100644 index 000000000..55850394c --- /dev/null +++ b/tests/test_share_consumer.rs @@ -0,0 +1,122 @@ +//! Public-surface tests for the KIP-932 share consumer scaffolding. +//! +//! These tests deliberately do not require a broker: the +//! `BaseShareConsumer` is a stub until librdkafka exposes a public +//! share consumer C API. The goal is to lock in the shape of the public +//! API and prove that runtime methods uniformly return +//! `KafkaError::Unsupported`, so the future FFI swap is mechanical. + +#![cfg(feature = "kip-932")] + +use std::time::Duration; + +use rdkafka::config::{ClientConfig, FromClientConfig, FromClientConfigAndContext}; +use rdkafka::consumer::share::{ + AcknowledgeType, AcknowledgementMode, AutoOffsetReset, BaseShareConsumer, + DefaultShareConsumerContext, IsolationLevel, ShareConsumer, ShareConsumerConfig, +}; +use rdkafka::error::KafkaError; + +fn make_config() -> ClientConfig { + let mut config = ShareConsumerConfig::new(); + config + .group_id("share-consumer-smoke") + .acknowledgement_mode(AcknowledgementMode::Explicit) + .auto_offset_reset(AutoOffsetReset::Earliest) + .isolation_level(IsolationLevel::ReadCommitted) + .set("bootstrap.servers", "localhost:9092"); + config.into_client_config().expect("group.id is set") +} + +fn assert_unsupported(result: Result) { + match result { + Err(KafkaError::Unsupported(reason)) => { + assert!( + reason.contains("KIP-932"), + "expected KIP-932 marker in reason, got {reason:?}" + ); + } + other => panic!("expected Unsupported, got {other:?}"), + } +} + +#[test] +fn config_writes_canonical_keys() { + let config = make_config(); + let map = config.config_map(); + assert_eq!(map.get("group.id").copied(), Some("share-consumer-smoke")); + assert_eq!( + map.get("share.acknowledgement.mode").copied(), + Some("explicit") + ); + assert_eq!( + map.get("share.auto.offset.reset").copied(), + Some("earliest") + ); + assert_eq!( + map.get("share.isolation.level").copied(), + Some("read_committed") + ); + assert_eq!( + map.get("group.share.delivery.attempt.limit").copied(), + Some("5") + ); + assert_eq!( + map.get("group.share.record.lock.duration.ms").copied(), + Some("30000") + ); + assert_eq!( + map.get("bootstrap.servers").copied(), + Some("localhost:9092") + ); +} + +#[test] +fn construction_round_trip() { + let config = make_config(); + let consumer = BaseShareConsumer::::from_config(&config).unwrap(); + assert_eq!(consumer.group_id(), "share-consumer-smoke"); +} + +#[test] +fn construction_with_context() { + let config = make_config(); + let consumer = + BaseShareConsumer::from_config_and_context(&config, DefaultShareConsumerContext).unwrap(); + assert_eq!(consumer.group_id(), "share-consumer-smoke"); +} + +#[test] +fn runtime_methods_return_unsupported() { + let config = make_config(); + let consumer = BaseShareConsumer::::from_config(&config).unwrap(); + + assert_unsupported(consumer.subscribe(&["topic-a", "topic-b"])); + assert_unsupported(consumer.unsubscribe()); + assert_unsupported(consumer.subscription()); + assert_unsupported(consumer.poll(Duration::from_millis(10))); + assert_unsupported(consumer.acknowledge_offset("topic-a", 0, 42, AcknowledgeType::Accept)); + assert_unsupported(consumer.commit_sync(Duration::from_millis(10))); + assert_unsupported(consumer.commit_async()); +} + +#[test] +fn close_and_wakeup_are_safe_no_ops() { + let config = make_config(); + let consumer = BaseShareConsumer::::from_config(&config).unwrap(); + consumer.wakeup(); + consumer.close().expect("close stub returns Ok"); +} + +#[test] +fn acknowledge_type_string_round_trip() { + for ack in [ + AcknowledgeType::Accept, + AcknowledgeType::Release, + AcknowledgeType::Reject, + ] { + let s = ack.to_string(); + let parsed: AcknowledgeType = s.parse().unwrap(); + assert_eq!(parsed, ack); + } +}