Enum Request 

pub enum Request {
    Authenticate {
        username: String,
        password: String,
    },
    SaslAuthenticate {
        mechanism: Bytes,
        auth_bytes: Bytes,
    },
    ScramClientFirst {
        message: Bytes,
    },
    ScramClientFinal {
        message: Bytes,
    },
    Publish {
        topic: String,
        partition: Option<u32>,
        key: Option<Bytes>,
        value: Bytes,
    },
    Consume {
        topic: String,
        partition: u32,
        offset: u64,
        max_messages: usize,
    },
    CreateTopic {
        name: String,
        partitions: Option<u32>,
    },
    ListTopics,
    DeleteTopic {
        name: String,
    },
    CommitOffset {
        consumer_group: String,
        topic: String,
        partition: u32,
        offset: u64,
    },
    GetOffset {
        consumer_group: String,
        topic: String,
        partition: u32,
    },
    GetMetadata {
        topic: String,
    },
    GetClusterMetadata {
        topics: Vec<String>,
    },
    Ping,
    RegisterSchema {
        subject: String,
        schema: String,
    },
    GetSchema {
        id: i32,
    },
    GetOffsetBounds {
        topic: String,
        partition: u32,
    },
    ListGroups,
    DescribeGroup {
        consumer_group: String,
    },
    DeleteGroup {
        consumer_group: String,
    },
    GetOffsetForTimestamp {
        topic: String,
        partition: u32,
        timestamp_ms: i64,
    },
    InitProducerId {
        producer_id: Option<u64>,
    },
    IdempotentPublish {
        topic: String,
        partition: Option<u32>,
        key: Option<Bytes>,
        value: Bytes,
        producer_id: u64,
        producer_epoch: u16,
        sequence: i32,
    },
    BeginTransaction {
        txn_id: String,
        producer_id: u64,
        producer_epoch: u16,
        timeout_ms: Option<u64>,
    },
    AddPartitionsToTxn {
        txn_id: String,
        producer_id: u64,
        producer_epoch: u16,
        partitions: Vec<(String, u32)>,
    },
    TransactionalPublish {
        txn_id: String,
        topic: String,
        partition: Option<u32>,
        key: Option<Bytes>,
        value: Bytes,
        producer_id: u64,
        producer_epoch: u16,
        sequence: i32,
    },
    AddOffsetsToTxn {
        txn_id: String,
        producer_id: u64,
        producer_epoch: u16,
        group_id: String,
        offsets: Vec<(String, u32, i64)>,
    },
    CommitTransaction {
        txn_id: String,
        producer_id: u64,
        producer_epoch: u16,
    },
    AbortTransaction {
        txn_id: String,
        producer_id: u64,
        producer_epoch: u16,
    },
    DescribeQuotas {
        entities: Vec<(String, Option<String>)>,
    },
    AlterQuotas {
        alterations: Vec<QuotaAlteration>,
    },
    AlterTopicConfig {
        topic: String,
        configs: Vec<TopicConfigEntry>,
    },
    CreatePartitions {
        topic: String,
        new_partition_count: u32,
        assignments: Vec<Vec<String>>,
    },
    DeleteRecords {
        topic: String,
        partition_offsets: Vec<(u32, u64)>,
    },
    DescribeTopicConfigs {
        topics: Vec<String>,
    },
}

Protocol request messages

Stability

WARNING: Variant order must remain stable for postcard serialization compatibility. Adding new variants should only be done at the end of the enum.

Variants

Authenticate

Authenticate with username/password (SASL/PLAIN compatible)

Fields

username: String

password: String

SaslAuthenticate

Authenticate with SASL bytes (for Kafka client compatibility)

Fields

mechanism: Bytes

auth_bytes: Bytes

ScramClientFirst

SCRAM-SHA-256: Client-first message

Fields

message: Bytes

Client-first-message bytes (n,,n=<user>,r=<nonce>)

ScramClientFinal

SCRAM-SHA-256: Client-final message

Fields

message: Bytes

Client-final-message bytes (c=<binding>,r=<nonce>,p=<proof>)

Publish

Publish a message to a topic

Fields

topic: String

partition: Option<u32>

key: Option<Bytes>

value: Bytes

Consume

Consume messages from a topic

Fields

topic: String

partition: u32

offset: u64

max_messages: usize

CreateTopic

Create a new topic

Fields

name: String

partitions: Option<u32>

ListTopics

List all topics

DeleteTopic

Delete a topic

Fields

name: String

CommitOffset

Commit consumer offset

Fields

consumer_group: String

topic: String

partition: u32

offset: u64

GetOffset

Get consumer offset

Fields

consumer_group: String

topic: String

partition: u32

GetMetadata

Get topic metadata

Fields

topic: String

GetClusterMetadata

Get cluster metadata (all topics or specific ones)

Fields

topics: Vec<String>

Topics to get metadata for (empty = all topics)

Ping

Ping

RegisterSchema

Register a schema

Fields

subject: String

schema: String

GetSchema

Get a schema

Fields

id: i32

GetOffsetBounds

Get offset bounds for a partition

Fields

topic: String

partition: u32

ListGroups

List all consumer groups

DescribeGroup

Describe a consumer group (get all offsets)

Fields

consumer_group: String

DeleteGroup

Delete a consumer group

Fields

consumer_group: String

GetOffsetForTimestamp

Find offset for a timestamp

Fields

topic: String

partition: u32

timestamp_ms: i64

Timestamp in milliseconds since epoch

InitProducerId

Initialize idempotent producer (request producer ID and epoch)

Call this before sending idempotent produce requests. If reconnecting, provide the previous producer_id to bump epoch.

Fields

producer_id: Option<u64>

Previous producer ID (None for new producers)

IdempotentPublish

Publish with idempotent semantics (exactly-once delivery)

Requires InitProducerId to have been called first.

Fields

topic: String

partition: Option<u32>

key: Option<Bytes>

value: Bytes

producer_id: u64

Producer ID from InitProducerId response

producer_epoch: u16

Producer epoch from InitProducerId response

sequence: i32

Sequence number (starts at 0, increments per partition)

BeginTransaction

Begin a new transaction

Fields

txn_id: String

Transaction ID (unique per producer)

producer_id: u64

Producer ID from InitProducerId

producer_epoch: u16

Producer epoch

timeout_ms: Option<u64>

Transaction timeout in milliseconds (optional, defaults to 60s)

AddPartitionsToTxn

Add partitions to an active transaction

Fields

txn_id: String

Transaction ID

producer_id: u64

Producer ID

producer_epoch: u16

Producer epoch

partitions: Vec<(String, u32)>

Partitions to add (topic, partition pairs)

TransactionalPublish

Publish within a transaction (combines IdempotentPublish + transaction tracking)

Fields

txn_id: String

Transaction ID

topic: String

partition: Option<u32>

key: Option<Bytes>

value: Bytes

producer_id: u64

Producer ID

producer_epoch: u16

Producer epoch

sequence: i32

Sequence number

AddOffsetsToTxn

Add consumer offsets to transaction (for exactly-once consume-transform-produce)

Fields

txn_id: String

Transaction ID

producer_id: u64

Producer ID

producer_epoch: u16

Producer epoch

group_id: String

Consumer group ID

offsets: Vec<(String, u32, i64)>

Offsets to commit (topic, partition, offset triples)

CommitTransaction

Commit a transaction

Fields

txn_id: String

Transaction ID

producer_id: u64

Producer ID

producer_epoch: u16

Producer epoch

AbortTransaction

Abort a transaction

Fields

txn_id: String

Transaction ID

producer_id: u64

Producer ID

producer_epoch: u16

Producer epoch

DescribeQuotas

Describe quotas for entities

Fields

entities: Vec<(String, Option<String>)>

Entities to describe (empty = all) Format: Vec<(entity_type, entity_name)> entity_type: “user”, “client-id”, “consumer-group”, “default” entity_name: None for defaults, Some for specific entities

AlterQuotas

Alter quotas for entities

Fields

alterations: Vec<QuotaAlteration>

Quota alterations to apply Each item: (entity_type, entity_name, quota_key, quota_value) quota_key: “produce_bytes_rate”, “consume_bytes_rate”, “request_rate” quota_value: None to remove, Some(value) to set

AlterTopicConfig

Alter topic configuration

Fields

topic: String

Topic name

configs: Vec<TopicConfigEntry>

Configuration changes to apply

CreatePartitions

Create additional partitions for an existing topic

Fields

topic: String

Topic name

new_partition_count: u32

New total partition count (must be > current count)

assignments: Vec<Vec<String>>

Optional assignment of new partitions to brokers If empty, broker will auto-assign

DeleteRecords

Delete records before a given offset (log truncation)

Fields

topic: String

Topic name

partition_offsets: Vec<(u32, u64)>

Partition-offset pairs: delete all records before these offsets

DescribeTopicConfigs

Describe topic configurations

Fields

topics: Vec<String>

Topics to describe (empty = all)

Implementations

impl Request

pub fn to_bytes(&self) -> Result<Vec<u8>, ProtocolError>

Serialize request to bytes

pub fn from_bytes(data: &[u8]) -> Result<Request, ProtocolError>

Deserialize request from bytes

Trait Implementations

impl Clone for Request

fn clone(&self) -> Request

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Request

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl<'de> Deserialize<'de> for Request

fn deserialize<__D>( __deserializer: __D, ) -> Result<Request, <__D as Deserializer<'de>>::Error>

where __D: Deserializer<'de>,

Deserialize this value from the given Serde deserializer.

impl Serialize for Request

fn serialize<__S>( &self, __serializer: __S, ) -> Result<<__S as Serializer>::Ok, <__S as Serializer>::Error>

where __S: Serializer,

Serialize this value into the given Serde serializer.