You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: error.rs
    path of file: ./repos/rust-rdkafka/src
    
    the code of the file until where you have to start completion: //! Error manipulations.

use std::error::Error;
use std::ffi::{self, CStr};
use std::fmt;
use std::ptr;
use std::sync::Arc;

use rdkafka_sys as rdsys;
use rdkafka_sys::types::*;

use crate::util::{KafkaDrop, NativePtr};

// Re-export rdkafka error code
pub use rdsys::types::RDKafkaErrorCode;

/// Kafka result.
pub type KafkaResult<T> = Result<T, KafkaError>;

/// Verify if the value represents an error condition.
///
/// Some librdkafka codes are informational, rather than true errors.
pub trait IsError {
    /// Reports whether the value represents an error.
    fn is_error(&self) -> bool;
}

impl IsError for RDKafkaRespErr {
    fn is_error(&self) -> bool {
        *self != RDKafkaRespErr::RD_KAFKA_RESP_ERR_NO_ERROR
    }
}

impl IsError for RDKafkaConfRes {
    fn is_error(&self) -> bool {
        *self != RDKafkaConfRes::RD_KAFKA_CONF_OK
    }
}

impl IsError for RDKafkaError {
    fn is_error(&self) -> bool {
        self.0.is_some()
    }
}

/// Native rdkafka error.
#[derive(Clone)]
pub struct RDKafkaError(Option<Arc<NativePtr<rdsys::rd_kafka_error_t>>>);

unsafe impl KafkaDrop for rdsys::rd_kafka_error_t {
    const TYPE: &'static str = "error";
    const DROP: unsafe extern "C" fn(*mut Self) = rdsys::rd_kafka_error_destroy;
}

unsafe impl Send for RDKafkaError {}
unsafe impl Sync for RDKafkaError {}

impl RDKafkaError {
    pub(crate) unsafe fn from_ptr(ptr: *mut rdsys::rd_kafka_error_t) -> RDKafkaError {
        RDKafkaError(NativePtr::from_ptr(ptr).map(Arc::new))
    }

    fn ptr(&self) -> *const rdsys::rd_kafka_error_t {
        match &self.0 {
            None => ptr::null(),
            Some(p) => p.ptr(),
        }
    }

    /// Returns the error code or [`RDKafkaErrorCode::NoError`] if the error is
    /// null.
    pub fn code(&self) -> RDKafkaErrorCode {
        unsafe { rdsys::rd_kafka_error_code(self.ptr()).into() }
    }

    /// Returns the error code name, e.g., "ERR_UNKNOWN_MEMBER_ID" or an empty
    /// string if the e
    
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            KafkaError::AdminOp(err) => write!(f, "Admin operation error: {}", err),
            KafkaError::AdminOpCreation(ref err) => {
                write!(f, "Admin operation creation error: {}", err)
            }
            KafkaError::Canceled => write!(f, "KafkaError (Client dropped)"),
            KafkaError::ClientConfig(_, ref desc, ref key, ref value) => {
                write!(f, "Client config error: {} {} {}", desc, key, value)
            }
            KafkaError::ClientCreation(ref err) => write!(f, "Client creation error: {}", err),
            KafkaError::ConsumerCommit(err) => write!(f, "Consumer commit error: {}", err),
            KafkaError::ConsumerQueueClose(err) => write!(f, "Consumer queue close error: {}", err),
            KafkaError::Flush(err) => write!(f, "Flush error: {}", err),
            KafkaError::Global(err) => write!(f, "Global error: {}", err),
            KafkaError::GroupListFetch(err) => write!(f, "Group list fetch error: {}", err),
            KafkaError::MessageConsumption(err) => write!(f, "Message consumption error: {}", err),
            KafkaError::MessageConsumptionFatal(err) => {
                write!(f, "(Fatal) Message consumption error: {}", err)
            }
            KafkaError::MessageProduction(err) => write!(f, "Message production error: {}", err),
            KafkaError::MetadataFetch(err) => write!(f, "Meta data fetch error: {}", err),
            KafkaError::NoMessageReceived => {
                write!(f, "No message received within the given poll interval")
            }
            KafkaError::Nul(_) => write!(f, "FFI nul error"),
            KafkaError::OffsetFetch(err) => write!(f, "Offset fetch error: {}", err),
            KafkaError::PartitionEOF(part_n) => write!(f, "Partition EOF: {}", part_n),
            KafkaError::PauseResume(ref err) => write!(f, "Pause/resume error: {}", err),
            KafkaError::Rebalance(ref err) => write!(f, "Rebalance error: {}", err),
            KafkaError::Seek(ref err) => write!(f, "Seek error: {}", err),
            KafkaError::SetPartitionOffset(err) => write!(f, "Set partition offset error: {}", err),
            KafkaError::StoreOffset(err) => write!(f, "Store offset error: {}", err),
            KafkaError::Subscription(ref err) => write!(f, "Subscription error: {}", err),
            KafkaError::Transaction(err) => write!(f, "Transaction error: {}", err),
            KafkaError::MockCluster(err) => write!(f, "Mock cluster error: {}", err),
        }
    }