krabka_client_core/
lib.rs

1//! Connection management and request dispatch for Apache Kafka in Rust.
2//!
3//! This crate is the first I/O-doing layer of Krabka. It wraps
4//! `krabka-protocol`'s typed request/response messages in a `tokio`-based
5//! TCP client that:
6//!
7//! - Opens one connection per broker and multiplexes requests by
8//!   correlation ID.
9//! - Negotiates API versions on connect.
10//! - Manages a [`BrokerPool`] keyed on broker id with lazy connect.
11//! - Resolves bootstrap addresses in the builder.
12//!
13//! ## Quick start
14//!
15//! ```no_run
16//! use krabka_client_core::Client;
17//! use krabka_protocol::owned::api_versions_request::ApiVersionsRequest;
18//!
19//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
20//! let client = Client::builder()
21//!     .bootstrap("localhost:9092")
22//!     .client_id("my-app")
23//!     .build()
24//!     .await?;
25//!
26//! let resp = client.send(ApiVersionsRequest::default()).await?;
27//! println!("broker supports {} APIs", resp.api_keys.len());
28//!
29//! client.close();
30//! # Ok(())
31//! # }
32//! ```
33//!
34//! ## Scope and boundaries
35//!
36//! This crate is the shared transport and request-dispatch layer. It supplies
37//! bootstrap resolution, API-version negotiation, broker-id connection pooling,
38//! typed request/response dispatch, low-level fetch helpers, and client-side
39//! TLS/SASL negotiation. Higher-level semantics live in the producer, consumer,
40//! and admin crates built on top of this one. Those semantics are batching,
41//! idempotence, consumer-group heartbeats, commits, admin retries, and
42//! transactions.
43//!
44//! TLS / SASL: a client-side security surface lives in [`security`] and
45//! [`sasl`]. Set [`ConnectionOptions::security`] (or the `Client`
46//! builder's `.security(...)`) to negotiate TLS then SASL before the
47//! API-versions bootstrap. `None` (the default) is plaintext.
48//!
49//! Sockets: [`transport::dial`] opens every outbound TCP connection. An
50//! embedder that supplies its own sockets installs a
51//! [`transport::Connector`].
52//!
53//! ## WebAssembly
54//!
55//! The crate builds for `wasm32-wasip1` with `--cfg tokio_unstable`, which
56//! tokio needs for its `net` feature on WebAssembly. It runs on a
57//! current-thread tokio runtime. On that target:
58//!
59//! - WASI preview 1 has no `connect` call, so the embedder installs a
60//!   [`transport::Connector`] before the first connection.
61//! - WASI has no name resolution, so broker addresses are IP address
62//!   literals, such as `10.0.0.1:9092`.
63//! - `client.dns.lookup=resolve_canonical_bootstrap_servers_only` does no
64//!   reverse lookup, so each bootstrap address keeps its IP address text.
65//! - A TLS connection needs a configured trust store, because there is no
66//!   platform trust store.
67//! - SASL/GSSAPI fails with
68//!   [`SaslAuthenticationError::UnsupportedMechanism`], with or without the
69//!   `gssapi` feature.
70//!
71//! ## Cargo features
72//!
73//! - `mock` — exposes `MockBroker` beyond `#[cfg(test)]` for downstream
74//!   testing.
75//! - `gssapi` (off by default) — the SASL/GSSAPI (Kerberos) client, through
76//!   `krabka-security/gssapi`. It stays out of `default` because the Kerberos
77//!   stack depends on `rsa` 0.10, which carries RUSTSEC-2023-0071 and has no
78//!   patched release; a build without the feature compiles no `rsa`. Without
79//!   it, a [`SaslCredentials::Gssapi`] credential fails
80//!   [`ConnectionOptions::validate`], and so client construction, with
81//!   [`ClientError::InvalidConfig`]; a connection that skips validation fails
82//!   the handshake with [`SaslAuthenticationError::UnsupportedMechanism`]. The
83//!   admin, consumer and producer crates forward the feature.
84
85mod backoff;
86mod bootstrap;
87mod client;
88mod connection;
89mod coordinator;
90mod error;
91mod fetch;
92mod framing;
93mod metadata_topics;
94pub mod oauth;
95mod offset_for_leader_epoch;
96mod pool;
97mod reauth;
98mod request;
99pub mod sasl;
100pub mod security;
101pub mod telemetry;
102pub mod transport;
103mod version;
104
105#[cfg(any(test, feature = "mock"))]
106mod mock;
107#[cfg(any(test, feature = "mock"))]
108pub mod mock_test_support;
109
110pub use backoff::{ExponentialBackoff, KAFKA_BACKOFF_JITTER, KAFKA_BACKOFF_MULTIPLIER};
111pub use client::{
112    BrokerHandle, Client, DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER,
113    MetadataRecoveryRebootstrapTrigger, MetadataRecoveryStrategy,
114};
115pub use connection::{
116    ClientDnsLookup, ClientDnsTimeout, ClientDuplex, ClientFrameMax, Connection,
117    ConnectionDispatchQueueCapacity, ConnectionOptions, DEFAULT_CLIENT_DNS_TIMEOUT,
118    DEFAULT_CLIENT_FRAME_MAX, DEFAULT_CLIENT_REQUEST_TIMEOUT,
119    DEFAULT_CONNECTION_DISPATCH_QUEUE_CAPACITY, DEFAULT_CONNECTIONS_MAX_IDLE,
120    DEFAULT_RECEIVE_BUFFER, DEFAULT_RECONNECT_BACKOFF, DEFAULT_RECONNECT_BACKOFF_MAX,
121    DEFAULT_SEND_BUFFER, DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT,
122    DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX, MAX_CLIENT_FRAME_BYTES,
123};
124pub use coordinator::{
125    CoordinatorEndpoint, CoordinatorKeyType, build_find_coordinator, coordinator_endpoint,
126    find_coordinator,
127};
128pub use error::{AuthenticationError, ClientError};
129pub use fetch::{
130    DEFAULT_FETCH_MIN, DEFAULT_FETCH_RESPONSE_MAX, FetchMinBytes, FetchPartitionResult,
131    FetchedHeader, FetchedRecord, IsolatedFetch, fetch_partition, fetch_partition_with_isolation,
132    fetch_partition_with_isolation_progress,
133};
134pub use metadata_topics::{MetadataScope, MetadataTopics, all_topics_request, topics_request};
135#[cfg(any(test, feature = "mock"))]
136pub use mock::{MockBroker, MockReply, MockSaslAnswer};
137pub use offset_for_leader_epoch::{EpochEndOffset, offset_for_leader_epoch};
138pub use pool::{BrokerInfo, BrokerPool};
139pub use request::{
140    AllTopicsOffsetFetch, MembersOnlyLeaveGroup, ProtocolRequest, TopicNameOffsetCommit,
141    TopicNameOffsetFetch, VersionBoundedRequest,
142};
143pub use sasl::{
144    OAuthBearerTokenSource, OutboundSaslError, SaslAuthenticationError, SaslCredentials,
145    SaslSession, outbound_sasl,
146};
147pub use security::{
148    ClientSecurity, KeyStore, Password, TlsConfigError, TlsConnectorConfig, TlsVersion, TrustStore,
149    connection_target_host,
150};
151pub use version::{ApiVersionTable, FinalizedFeatures, UNKNOWN_FINALIZED_FEATURES_EPOCH};
krabka_client_core/lib.rs

krabka_client_core/
lib.rs
