Enum RocketMQError 

pub enum RocketMQError {
    Network(NetworkError),
    Serialization(SerializationError),
    Protocol(ProtocolError),
    Rpc(RpcClientError),
    Authentication(AuthError),
    Controller(ControllerError),
    InvalidProperty(String),
    BrokerNotFound {
        name: String,
    },
    BrokerRegistrationFailed {
        name: String,
        reason: String,
    },
    BrokerOperationFailed {
        operation: &'static str,
        code: i32,
        message: String,
        broker_addr: Option<String>,
    },
    TopicNotExist {
        topic: String,
    },
    QueueNotExist {
        topic: String,
        queue_id: i32,
    },
    SubscriptionGroupNotExist {
        group: String,
    },
    QueueIdOutOfRange {
        topic: String,
        queue_id: i32,
        max: i32,
    },
    MessageTooLarge {
        actual: usize,
        limit: usize,
    },
    MessageValidationFailed {
        reason: String,
    },
    RetryLimitExceeded {
        group: String,
        current: i32,
        max: i32,
    },
    TransactionRejected,
    BrokerPermissionDenied {
        operation: String,
    },
    NotMasterBroker {
        master_address: String,
    },
    MessageLookupFailed {
        offset: i64,
    },
    TopicSendingForbidden {
        topic: String,
    },
    BrokerAsyncTaskFailed {
        task: &'static str,
        context: String,
        source: Box<dyn Error + Send + Sync>,
    },
    RequestBodyInvalid {
        operation: &'static str,
        reason: String,
    },
    RequestHeaderError(String),
    ResponseProcessFailed {
        operation: &'static str,
        reason: String,
    },
    RouteNotFound {
        topic: String,
    },
    RouteInconsistent {
        topic: String,
        reason: String,
    },
    RouteRegistrationConflict {
        broker_name: String,
        reason: String,
    },
    RouteVersionConflict {
        expected: u64,
        actual: u64,
    },
    ClusterNotFound {
        cluster: String,
    },
    ClientNotStarted,
    ClientAlreadyStarted,
    ClientShuttingDown,
    ClientInvalidState {
        expected: &'static str,
        actual: String,
    },
    ProducerNotAvailable,
    ConsumerNotAvailable,
    Tools(ToolsError),
    Filter(FilterError),
    StorageReadFailed {
        path: String,
        reason: String,
    },
    StorageWriteFailed {
        path: String,
        reason: String,
    },
    StorageCorrupted {
        path: String,
    },
    StorageOutOfSpace {
        path: String,
    },
    StorageLockFailed {
        path: String,
    },
    ConfigParseFailed {
        key: &'static str,
        reason: String,
    },
    ConfigMissing {
        key: &'static str,
    },
    ConfigInvalidValue {
        key: &'static str,
        value: String,
        reason: String,
    },
    ControllerNotLeader {
        leader_id: Option<u64>,
    },
    ControllerRaftError {
        reason: String,
    },
    ControllerConsensusTimeout {
        operation: &'static str,
        timeout_ms: u64,
    },
    ControllerSnapshotFailed {
        reason: String,
    },
    IO(Error),
    IllegalArgument(String),
    Timeout {
        operation: &'static str,
        timeout_ms: u64,
    },
    Internal(String),
    Service(ServiceError),
    InvalidVersionOrdinal(u32),
    Legacy(String),
    NotInitialized(String),
    MissingRequiredMessageProperty {
        property: &'static str,
    },
}

Main error type for all RocketMQ operations

This enum provides a unified error system across all RocketMQ crates. Each variant represents a logical category of errors with rich context information.

Design Principles

• Semantic: Each error clearly expresses what went wrong
• Performance: Minimal heap allocations, use of &’static str where possible
• Debuggability: Rich context for production debugging
• Ergonomics: Automatic conversions via From trait

Examples

use rocketmq_error::RocketMQError;
use rocketmq_error::RocketMQResult;

fn send_message(addr: &str) -> RocketMQResult<()> {
    // Create a network error
    if addr.is_empty() {
        return Err(RocketMQError::network_connection_failed(
            "localhost:9876",
            "empty address",
        ));
    }
    Ok(())
}

fn authenticate_user(username: &str) -> RocketMQResult<()> {
    // Create an authentication error
    if username.is_empty() {
        return Err(RocketMQError::user_not_found(""));
    }
    Ok(())
}

Variants

Network(NetworkError)

Network operation errors (connection, timeout, send/receive failures)

Serialization(SerializationError)

Serialization/deserialization errors (encoding, decoding, format validation)

Protocol(ProtocolError)

RocketMQ protocol errors (invalid commands, version mismatch, etc.)

Rpc(RpcClientError)

RPC client specific errors (broker lookup, request failures, etc.)

Authentication(AuthError)

Authentication/authorization errors (credential validation, access control, etc.)

Controller(ControllerError)

Controller operation errors (Raft consensus, leader election, broker management, etc.)

InvalidProperty(String)

Invalid message property

BrokerNotFound

Broker not found

Fields

name: String

BrokerRegistrationFailed

Broker registration failed

Fields

name: String

reason: String

BrokerOperationFailed

Broker operation failed with error code

Fields

operation: &'static str

code: i32

message: String

broker_addr: Option<String>

TopicNotExist

Topic does not exist

Fields

topic: String

QueueNotExist

Queue does not exist

Fields

topic: String

queue_id: i32

SubscriptionGroupNotExist

Subscription group not found

Fields

group: String

QueueIdOutOfRange

Queue ID out of range

Fields

topic: String

queue_id: i32

max: i32

MessageTooLarge

Message body too large

Fields

actual: usize

limit: usize

MessageValidationFailed

Message validation failed

Fields

reason: String

RetryLimitExceeded

Retry limit exceeded

Fields

group: String

current: i32

max: i32

TransactionRejected

Transaction message rejected

BrokerPermissionDenied

Broker permission denied

Fields

operation: String

NotMasterBroker

Not master broker

Fields

master_address: String

MessageLookupFailed

Message lookup failed

Fields

offset: i64

TopicSendingForbidden

Topic sending forbidden

Fields

topic: String

BrokerAsyncTaskFailed

Async task failed

Fields

task: &'static str

context: String

source: Box<dyn Error + Send + Sync>

RequestBodyInvalid

Request body missing or invalid

Fields

operation: &'static str

reason: String

RequestHeaderError(String)

Request header missing or invalid

ResponseProcessFailed

Response encoding/decoding failed

Fields

operation: &'static str

reason: String

RouteNotFound

Route information not found

Fields

topic: String

RouteInconsistent

Route data inconsistency detected

Fields

topic: String

reason: String

RouteRegistrationConflict

Broker registration conflict

Fields

broker_name: String

reason: String

RouteVersionConflict

Route state version conflict

Fields

expected: u64

actual: u64

ClusterNotFound

Cluster not found

Fields

cluster: String

ClientNotStarted

Client not started

ClientAlreadyStarted

Client already started

ClientShuttingDown

Client is shutting down

ClientInvalidState

Invalid client state

Fields

expected: &'static str

actual: String

ProducerNotAvailable

Producer not available

ConsumerNotAvailable

Consumer not available

Tools(ToolsError)

Tools and admin operation errors

Filter(FilterError)

Bloom filter and bit array operation errors

StorageReadFailed

Storage read failed

Fields

path: String

reason: String

StorageWriteFailed

Storage write failed

Fields

path: String

reason: String

StorageCorrupted

Data corruption detected

Fields

path: String

StorageOutOfSpace

Out of storage space

Fields

path: String

StorageLockFailed

Storage lock failed

Fields

path: String

ConfigParseFailed

Configuration parsing failed

Fields

key: &'static str

reason: String

ConfigMissing

Required configuration missing

Fields

key: &'static str

ConfigInvalidValue

Invalid configuration value

Fields

key: &'static str

value: String

reason: String

ControllerNotLeader

Not the Raft leader

Fields

leader_id: Option<u64>

ControllerRaftError

Raft consensus error

Fields

reason: String

ControllerConsensusTimeout

Consensus operation timeout

Fields

operation: &'static str

timeout_ms: u64

ControllerSnapshotFailed

Snapshot operation failed

Fields

reason: String

IO(Error)

IO error from std::io

IllegalArgument(String)

Illegal argument

Timeout

Operation timeout

Fields

operation: &'static str

timeout_ms: u64

Internal(String)

Internal error (should be rare)

Service(ServiceError)

Service lifecycle error

InvalidVersionOrdinal(u32)

Invalid RocketMQ version ordinal value

Legacy(String)

👎Deprecated since 0.7.0:

Use specific error types instead

Legacy error - use specific error types instead

NotInitialized(String)

MissingRequiredMessageProperty

Fields

property: &'static str

Implementations

impl RocketMQError

pub fn network_connection_failed( addr: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a network connection failed error

pub fn network_timeout(addr: impl Into<String>, timeout: Duration) -> Self

Create a network timeout error

pub fn network_request_failed( addr: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a network request failed error

pub fn deserialization_failed( format: &'static str, reason: impl Into<String>, ) -> Self

Create a deserialization failed error

pub fn validation_failed( field: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a validation failed error

pub fn broker_operation_failed( operation: &'static str, code: i32, message: impl Into<String>, ) -> Self

Create a broker operation failed error

pub fn storage_read_failed( path: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a storage read failed error

pub fn storage_write_failed( path: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a storage write failed error

pub fn illegal_argument(message: impl Into<String>) -> Self

Create an illegal argument error

pub fn route_not_found(topic: impl Into<String>) -> Self

Create a route not found error

pub fn route_registration_conflict( broker_name: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a route registration conflict error

pub fn cluster_not_found(cluster: impl Into<String>) -> Self

Create a cluster not found error

pub fn request_body_invalid( operation: &'static str, reason: impl Into<String>, ) -> Self

Create a request body invalid error

pub fn request_header_error(message: impl Into<String>) -> Self

Create a request header error

pub fn response_process_failed( operation: &'static str, reason: impl Into<String>, ) -> Self

Create a response process failed error

pub fn with_broker_addr(self, addr: impl Into<String>) -> Self

Add broker address context to broker operation error

pub fn validation_error( field: impl Into<String>, reason: impl Into<String>, ) -> Self

Create a validation error

pub fn topic_not_found(topic: impl Into<String>) -> Self

Create a topic not found error (alias for TopicNotExist)

pub fn topic_already_exists(topic: impl Into<String>) -> Self

Create a topic already exists error

pub fn nameserver_unreachable(addr: impl Into<String>) -> Self

Create a nameserver unreachable error

pub fn nameserver_config_invalid(reason: impl Into<String>) -> Self

Create a nameserver config invalid error

pub fn not_initialized(reason: impl Into<String>) -> Self

Create a not initialized error

pub fn authentication_failed(reason: impl Into<String>) -> Self

Create an authentication failed error

pub fn invalid_credential(reason: impl Into<String>) -> Self

Create an invalid credential error

pub fn user_not_found(username: impl Into<String>) -> Self

Create a user not found error

pub fn invalid_signature(reason: impl Into<String>) -> Self

Create an invalid signature error

pub fn controller_not_leader(leader_id: Option<u64>) -> Self

Create a controller not leader error

Examples found in repository

examples/controller_error_integration.rs (line 122)

fn main() {
    println!("=== ControllerError Integration Example ===\n");

    // Example 1: Using constructor methods
    println!("1. Creating controller errors using constructor methods:");
    let err1 = RocketMQError::controller_not_leader(Some(1));
    println!("   Not leader: {}", err1);

    let err2 = RocketMQError::controller_timeout(5000);
    println!("   Timeout: {}", err2);

    let err3 = RocketMQError::controller_invalid_request("missing broker_name");
    println!("   Invalid request: {}", err3);

    // Example 2: Automatic conversion
    println!("\n2. Automatic conversion from ControllerError:");
    let controller_err = ControllerError::Raft("consensus failed".to_string());
    let rocketmq_err: RocketMQError = controller_err.into();
    println!("   Converted error: {}", rocketmq_err);

    // Example 3: Error propagation
    println!("\n3. Error propagation in functions:");
    match simulated_operation() {
        Ok(_) => println!("   Operation succeeded"),
        Err(e) => println!("   Operation failed: {}", e),
    }

    // Example 4: Pattern matching
    println!("\n4. Pattern matching on errors:");
    let errors = vec![
        RocketMQError::controller_not_leader(Some(2)),
        RocketMQError::controller_timeout(3000),
        RocketMQError::controller_shutdown(),
    ];

    for err in errors {
        print_error_details(err);
    }

    println!("\n=== Example completed successfully ===");
}

// Simulated operation that returns controller error
fn simulated_operation() -> RocketMQResult<()> {
    // Simulate a not leader scenario
    Err(RocketMQError::controller_not_leader(None))
}

pub fn controller_raft_error(reason: impl Into<String>) -> Self

Create a controller Raft error

pub fn controller_metadata_not_found(key: impl Into<String>) -> Self

Create a controller metadata not found error

pub fn controller_invalid_request(reason: impl Into<String>) -> Self

Create a controller invalid request error

Examples found in repository

examples/controller_error_integration.rs (line 128)

fn main() {
    println!("=== ControllerError Integration Example ===\n");

    // Example 1: Using constructor methods
    println!("1. Creating controller errors using constructor methods:");
    let err1 = RocketMQError::controller_not_leader(Some(1));
    println!("   Not leader: {}", err1);

    let err2 = RocketMQError::controller_timeout(5000);
    println!("   Timeout: {}", err2);

    let err3 = RocketMQError::controller_invalid_request("missing broker_name");
    println!("   Invalid request: {}", err3);

    // Example 2: Automatic conversion
    println!("\n2. Automatic conversion from ControllerError:");
    let controller_err = ControllerError::Raft("consensus failed".to_string());
    let rocketmq_err: RocketMQError = controller_err.into();
    println!("   Converted error: {}", rocketmq_err);

    // Example 3: Error propagation
    println!("\n3. Error propagation in functions:");
    match simulated_operation() {
        Ok(_) => println!("   Operation succeeded"),
        Err(e) => println!("   Operation failed: {}", e),
    }

    // Example 4: Pattern matching
    println!("\n4. Pattern matching on errors:");
    let errors = vec![
        RocketMQError::controller_not_leader(Some(2)),
        RocketMQError::controller_timeout(3000),
        RocketMQError::controller_shutdown(),
    ];

    for err in errors {
        print_error_details(err);
    }

    println!("\n=== Example completed successfully ===");
}

pub fn controller_timeout(timeout_ms: u64) -> Self

Create a controller timeout error

Examples found in repository

examples/controller_error_integration.rs (line 125)

fn main() {
    println!("=== ControllerError Integration Example ===\n");

    // Example 1: Using constructor methods
    println!("1. Creating controller errors using constructor methods:");
    let err1 = RocketMQError::controller_not_leader(Some(1));
    println!("   Not leader: {}", err1);

    let err2 = RocketMQError::controller_timeout(5000);
    println!("   Timeout: {}", err2);

    let err3 = RocketMQError::controller_invalid_request("missing broker_name");
    println!("   Invalid request: {}", err3);

    // Example 2: Automatic conversion
    println!("\n2. Automatic conversion from ControllerError:");
    let controller_err = ControllerError::Raft("consensus failed".to_string());
    let rocketmq_err: RocketMQError = controller_err.into();
    println!("   Converted error: {}", rocketmq_err);

    // Example 3: Error propagation
    println!("\n3. Error propagation in functions:");
    match simulated_operation() {
        Ok(_) => println!("   Operation succeeded"),
        Err(e) => println!("   Operation failed: {}", e),
    }

    // Example 4: Pattern matching
    println!("\n4. Pattern matching on errors:");
    let errors = vec![
        RocketMQError::controller_not_leader(Some(2)),
        RocketMQError::controller_timeout(3000),
        RocketMQError::controller_shutdown(),
    ];

    for err in errors {
        print_error_details(err);
    }

    println!("\n=== Example completed successfully ===");
}

pub fn controller_shutdown() -> Self

Create a controller shutdown error

Examples found in repository

examples/controller_error_integration.rs (line 149)

fn main() {
    println!("=== ControllerError Integration Example ===\n");

    // Example 1: Using constructor methods
    println!("1. Creating controller errors using constructor methods:");
    let err1 = RocketMQError::controller_not_leader(Some(1));
    println!("   Not leader: {}", err1);

    let err2 = RocketMQError::controller_timeout(5000);
    println!("   Timeout: {}", err2);

    let err3 = RocketMQError::controller_invalid_request("missing broker_name");
    println!("   Invalid request: {}", err3);

    // Example 2: Automatic conversion
    println!("\n2. Automatic conversion from ControllerError:");
    let controller_err = ControllerError::Raft("consensus failed".to_string());
    let rocketmq_err: RocketMQError = controller_err.into();
    println!("   Converted error: {}", rocketmq_err);

    // Example 3: Error propagation
    println!("\n3. Error propagation in functions:");
    match simulated_operation() {
        Ok(_) => println!("   Operation succeeded"),
        Err(e) => println!("   Operation failed: {}", e),
    }

    // Example 4: Pattern matching
    println!("\n4. Pattern matching on errors:");
    let errors = vec![
        RocketMQError::controller_not_leader(Some(2)),
        RocketMQError::controller_timeout(3000),
        RocketMQError::controller_shutdown(),
    ];

    for err in errors {
        print_error_details(err);
    }

    println!("\n=== Example completed successfully ===");
}

pub fn filter_empty_bytes() -> Self

Create an empty bytes error

pub fn filter_invalid_bit_length() -> Self

Create an invalid bit length error

pub fn filter_bit_length_too_small() -> Self

Create a bit length too small error

pub fn filter_bit_position_out_of_bounds(pos: usize, max: usize) -> Self

Create a bit position out of bounds error

pub fn filter_byte_position_out_of_bounds(pos: usize, max: usize) -> Self

Create a byte position out of bounds error

pub fn filter_uninitialized() -> Self

Create an uninitialized error

Trait Implementations

impl Debug for RocketMQError

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

Formats the value using the given formatter.

impl Display for RocketMQError

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

Formats the value using the given formatter.

impl Error for RocketMQError

fn source(&self) -> Option<&(dyn Error + 'static)>

Returns the lower-level source of this error, if any.

fn description(&self) -> &str

👎Deprecated since 1.42.0:

use the Display impl or to_string()

fn cause(&self) -> Option<&dyn Error>

👎Deprecated since 1.33.0:

replaced by Error::source, which can support downcasting

fn provide<'a>(&'a self, request: &mut Request<'a>)

🔬This is a nightly-only experimental API. (error_generic_member_access)

Provides type-based access to context intended for error reports.

impl From<AuthError> for RocketMQError

fn from(source: AuthError) -> Self

Converts to this type from the input type.

impl From<ControllerError> for RocketMQError

fn from(source: ControllerError) -> Self

Converts to this type from the input type.

impl From<Error> for RocketMQError

fn from(source: Error) -> Self

Converts to this type from the input type.

impl From<FilterError> for RocketMQError

fn from(source: FilterError) -> Self

Converts to this type from the input type.

impl From<NetworkError> for RocketMQError

fn from(source: NetworkError) -> Self

Converts to this type from the input type.

impl From<ProtocolError> for RocketMQError

fn from(source: ProtocolError) -> Self

Converts to this type from the input type.

impl From<RocketmqError> for RocketMQError

Automatic conversion from legacy RocketmqError to unified RocketMQError

This allows legacy error-producing code to work with new error-expecting code

fn from(err: RocketmqError) -> Self

Converts to this type from the input type.

impl From<RpcClientError> for RocketMQError

fn from(source: RpcClientError) -> Self

Converts to this type from the input type.

impl From<SerializationError> for RocketMQError

fn from(source: SerializationError) -> Self

Converts to this type from the input type.

impl From<ServiceError> for RocketMQError

fn from(source: ServiceError) -> Self

Converts to this type from the input type.

impl From<ToolsError> for RocketMQError

fn from(source: ToolsError) -> Self

Converts to this type from the input type.

impl From<Utf8Error> for RocketMQError

fn from(e: Utf8Error) -> Self

Converts to this type from the input type.