Crate krabka_client_core

Expand description

Connection management and request dispatch for Apache Kafka in Rust.

This crate is the first I/O-doing layer of Krabka. It wraps krabka-protocol’s typed request/response messages in a tokio-based TCP client that:

Opens one connection per broker and multiplexes requests by correlation ID.
Negotiates API versions on connect.
Manages a BrokerPool keyed on broker id with lazy connect.
Resolves bootstrap addresses in the builder.

§Quick start

use krabka_client_core::Client;
use krabka_protocol::owned::api_versions_request::ApiVersionsRequest;

let client = Client::builder()
    .bootstrap("localhost:9092")
    .client_id("my-app")
    .build()
    .await?;

let resp = client.send(ApiVersionsRequest::default()).await?;
println!("broker supports {} APIs", resp.api_keys.len());

client.close();

§Scope and boundaries

This crate is the shared transport and request-dispatch layer. It supplies bootstrap resolution, API-version negotiation, broker-id connection pooling, typed request/response dispatch, low-level fetch helpers, and client-side TLS/SASL negotiation. Higher-level semantics live in the producer, consumer, and admin crates built on top of this one. Those semantics are batching, idempotence, consumer-group heartbeats, commits, admin retries, and transactions.

TLS / SASL: a client-side security surface lives in security and sasl. Set ConnectionOptions::security (or the Client builder’s .security(...)) to negotiate TLS then SASL before the API-versions bootstrap. None (the default) is plaintext.

Sockets: transport::dial opens every outbound TCP connection. An embedder that supplies its own sockets installs a transport::Connector.

§WebAssembly

The crate builds for wasm32-wasip1 with --cfg tokio_unstable, which tokio needs for its net feature on WebAssembly. It runs on a current-thread tokio runtime. On that target:

WASI preview 1 has no connect call, so the embedder installs a transport::Connector before the first connection.
WASI has no name resolution, so broker addresses are IP address literals, such as 10.0.0.1:9092.
client.dns.lookup=resolve_canonical_bootstrap_servers_only does no reverse lookup, so each bootstrap address keeps its IP address text.
A TLS connection needs a configured trust store, because there is no platform trust store.
SASL/GSSAPI fails with SaslAuthenticationError::UnsupportedMechanism, with or without the gssapi feature.

§Cargo features

mock — exposes MockBroker beyond #[cfg(test)] for downstream testing.
gssapi (off by default) — the SASL/GSSAPI (Kerberos) client, through krabka-security/gssapi. It stays out of default because the Kerberos stack depends on rsa 0.10, which carries RUSTSEC-2023-0071 and has no patched release; a build without the feature compiles no rsa. Without it, a SaslCredentials::Gssapi credential fails ConnectionOptions::validate, and so client construction, with ClientError::InvalidConfig; a connection that skips validation fails the handshake with SaslAuthenticationError::UnsupportedMechanism. The admin, consumer and producer crates forward the feature.

Re-exports§

pub use sasl::OAuthBearerTokenSource;
pub use sasl::OutboundSaslError;
pub use sasl::SaslAuthenticationError;
pub use sasl::SaslCredentials;
pub use sasl::SaslSession;
pub use sasl::outbound_sasl;
pub use security::ClientSecurity;
pub use security::KeyStore;
pub use security::Password;
pub use security::TlsConfigError;
pub use security::TlsConnectorConfig;
pub use security::TlsVersion;
pub use security::TrustStore;
pub use security::connection_target_host;

Modules§

oauth: OAUTHBEARER tokens: the token provider seam, an OAuth 2 client credentials provider and a jwt-bearer client assertion provider (KIP-768), and a background refresh task for either.
sasl: Transport-agnostic outbound SASL handshake for the client.
security: Client-side TLS/SASL security surface for crate::Client.
telemetry: Client metrics push (KIP-714).
transport: Outbound TCP connections and name resolution.

Structs§

ApiVersionTable
BrokerHandle: A handle to a specific broker within a Client’s pool.
BrokerInfo: Information about a single Kafka broker, as reported by a MetadataResponse.
BrokerPool: Pool of Arc<Connection> keyed by broker id.
Client: A Kafka client backed by a BrokerPool.
ClientDnsTimeout: Positive, whole-millisecond DNS lookup deadline.
ClientFrameMax: Positive whole-byte accepted-frame limit bounded by the fixed security ceiling.
Connection: A connection to a single Kafka broker.
ConnectionDispatchQueueCapacity: Positive capacity of one connection’s pending request dispatch queue.
ConnectionOptions: Connect-time + per-request configuration knobs.
CoordinatorEndpoint: The broker that coordinates one consumer group or one transactional id.
EpochEndOffset: One leader-epoch end-offset answer for a partition.
ExponentialBackoff: Kafka’s ExponentialBackoff: initial * multiplier^attempts, times a random factor in [1 - jitter, 1 + jitter), and never above max.
FetchMinBytes: Positive whole-byte minimum response size for a single-partition fetch.
FetchPartitionResult: Records returned by a fetch together with the next safe partition offset.
FetchedRecord: One record decoded from a single-partition fetch.
FinalizedFeatures: The finalized feature levels of a broker (KIP-584), from ApiVersions v3 and later, as Kafka’s ApiVersions.FinalizedFeaturesInfo holds them.
IsolatedFetch: Parameters for a single-partition fetch with an explicit isolation level.
MetadataRecoveryRebootstrapTrigger: Non-negative, whole-millisecond KIP-1102 rebootstrap trigger.
MetadataTopics: The topics of a client’s metadata requests, with the last time that each topic was used.
VersionBoundedRequest: A protocol request with a narrower version range than its wire type.

Enums§

AuthenticationError: The reason that a peer rejected authentication. Each variant is a subclass of Kafka’s AuthenticationException.
ClientDnsLookup: How the client resolves broker host names (Kafka client.dns.lookup).
ClientError: Errors returned by Client, Connection, and the broker pool.
CoordinatorKeyType: The key space that a FindCoordinator lookup asks about.
MetadataRecoveryStrategy: Kafka client behavior when its last-known broker metadata can no longer be refreshed.
MetadataScope: Which topics crate::Client::refresh_metadata asks for.

Constants§

DEFAULT_CLIENT_DNS_TIMEOUT: Default deadline for one client DNS lookup.
DEFAULT_CLIENT_FRAME_MAX: Default maximum accepted client frame size.
DEFAULT_CLIENT_REQUEST_TIMEOUT: Default deadline for one client request.
DEFAULT_CONNECTIONS_MAX_IDLE: Kafka’s producer and consumer connections.max.idle.ms default (9 minutes, below the broker’s 10 minutes).
DEFAULT_CONNECTION_DISPATCH_QUEUE_CAPACITY: Default capacity of one connection’s pending request dispatch queue.
DEFAULT_FETCH_MIN: Default minimum response size for a single-partition fetch.
DEFAULT_FETCH_RESPONSE_MAX: Default whole-response byte limit for single-partition fetch helpers.
DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER: KIP-1102’s default interval from the first unsuccessful metadata attempt until the client repeats bootstrap discovery.
DEFAULT_RECEIVE_BUFFER: Kafka’s consumer and admin receive.buffer.bytes default. The producer default is 32 KiB.
DEFAULT_RECONNECT_BACKOFF: Kafka’s reconnect.backoff.ms default.
DEFAULT_RECONNECT_BACKOFF_MAX: Kafka’s reconnect.backoff.max.ms default.
DEFAULT_SEND_BUFFER: Kafka’s producer, consumer and admin send.buffer.bytes default.
DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT: Kafka’s socket.connection.setup.timeout.ms default: the TCP connect deadline of the first attempt to a broker.
DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX: Kafka’s socket.connection.setup.timeout.max.ms default: the connect deadline grows exponentially per failed attempt up to this value.
KAFKA_BACKOFF_JITTER: The jitter of Kafka’s connection and retry backoffs.
KAFKA_BACKOFF_MULTIPLIER: The multiplier of Kafka’s connection and retry backoffs.
MAX_CLIENT_FRAME_BYTES: Fixed security ceiling for accepted client frames.
UNKNOWN_FINALIZED_FEATURES_EPOCH: Kafka’s ApiVersionsResponse.UNKNOWN_FINALIZED_FEATURES_EPOCH.

Traits§

ClientDuplex: Trait alias for the duplex stream types Connection::from_stream accepts, such as TcpStream and tokio_rustls::client::TlsStream.
ProtocolRequest: Implemented by every generated Request struct in krabka-protocol.

Functions§

all_topics_request: The Metadata request for every topic of the cluster (MetadataRequest.Builder.allTopics): a null topic list.
build_find_coordinator: Build a FindCoordinator request for key that works at any negotiated version.
coordinator_endpoint: Read the coordinator endpoint for key out of a FindCoordinator response.
fetch_partition: Fetch up to partition_max from (topic, partition) at fetch_offset.
fetch_partition_with_isolation: Like fetch_partition, but lets the caller set the Kafka Fetch.isolation_level (0 = READ_UNCOMMITTED, 1 = READ_COMMITTED).
fetch_partition_with_isolation_progress: Fetch a partition with isolation and return both visible records and cursor progress across filtered transactional and control batches.
find_coordinator: Ask the broker behind conn which broker coordinates key.
offset_for_leader_epoch: Send a single-partition OffsetForLeaderEpoch request.
topics_request: A Metadata request that names topics (MetadataRequest.Builder forTopicNames). An empty list asks for no topic.

Type Aliases§

AllTopicsOffsetFetch: An admin OffsetFetch request for all topics, which needs v2 through v9.
FetchedHeader: One Kafka record header decoded from a fetched record.
MembersOnlyLeaveGroup: A LeaveGroup request that carries its member list, needing v3 or newer.
TopicNameOffsetCommit: An OffsetCommit request that names its topics, capped at v9.
TopicNameOffsetFetch: An OffsetFetch request that names its topics, capped at v9.