diff --git a/openraft/src/base/batch/display.rs b/openraft/src/base/batch/display.rs
index 9342bca8a..c827c50e1 100644
--- a/openraft/src/base/batch/display.rs
+++ b/openraft/src/base/batch/display.rs
@@ -3,40 +3,46 @@
 use std::fmt;
 use std::fmt::Formatter;
 
-use super::Batch;
+use super::RaftBatch;
+use crate::OptionalSend;
 
-impl<T: fmt::Display> Batch<T> {
-    /// Returns a display helper that shows all elements.
-    pub fn display(&self) -> impl fmt::Display + '_ {
-        BatchDisplay {
-            elements: self,
-            max: None,
-        }
-    }
+/// Display helper for types implementing `RaftBatch`.
+pub struct DisplayBatch<'a, T, B>
+where
+    T: fmt::Display + OptionalSend + 'static + fmt::Debug,
+    B: RaftBatch<T>,
+{
+    pub(super) elements: &'a B,
+    pub(super) max: Option<usize>,
+    pub(super) _phantom: std::marker::PhantomData<T>,
+}
 
-    /// Returns a display helper that shows at most `max` elements.
-    pub fn display_n(&self, max: usize) -> impl fmt::Display + '_ {
-        BatchDisplay {
-            elements: self,
-            max: Some(max),
+impl<'a, T, B> DisplayBatch<'a, T, B>
+where
+    T: fmt::Display + OptionalSend + 'static + fmt::Debug,
+    B: RaftBatch<T>,
+{
+    pub(super) fn new(elements: &'a B, max: Option<usize>) -> Self {
+        Self {
+            elements,
+            max,
+            _phantom: std::marker::PhantomData,
         }
     }
 }
 
-struct BatchDisplay<'a, T> {
-    elements: &'a Batch<T>,
-    max: Option<usize>,
-}
-
-impl<'a, T: fmt::Display> fmt::Display for BatchDisplay<'a, T> {
+impl<'a, T, B> fmt::Display for DisplayBatch<'a, T, B>
+where
+    T: fmt::Display + OptionalSend + 'static + fmt::Debug,
+    B: RaftBatch<T>,
+{
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        let slice = self.elements.as_slice();
-        let max = self.max.unwrap_or(slice.len());
-        let len = slice.len();
+        let len = self.elements.len();
+        let max = self.max.unwrap_or(len);
         let shown = max.min(len);
 
         write!(f, "[")?;
-        for (i, e) in slice.iter().take(max).enumerate() {
+        for (i, e) in self.elements.iter().take(max).enumerate() {
             if i > 0 {
                 write!(f, ", ")?;
             }
@@ -55,25 +61,53 @@ impl<'a, T: fmt::Display> fmt::Display for BatchDisplay<'a, T> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::impls::Batch;
 
     #[test]
     fn test_display() {
-        assert_eq!(format!("{}", Batch::Single(42).display()), "[42]");
-        assert_eq!(format!("{}", Batch::Vec(vec![1, 2]).display()), "[1, 2]");
-        assert_eq!(format!("{}", Batch::<i32>::Vec(vec![]).display()), "[]");
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display(&Batch::Single(42))),
+            "[42]"
+        );
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display(&Batch::Vec(vec![1, 2]))),
+            "[1, 2]"
+        );
+        assert_eq!(
+            format!(
+                "{}",
+                <Batch<i32> as RaftBatch<i32>>::display(&Batch::<i32>::Vec(vec![]))
+            ),
+            "[]"
+        );
     }
 
     #[test]
     fn test_display_n() {
         let v: Batch<i32> = [1, 2, 3, 4, 5].into();
 
-        assert_eq!(format!("{}", v.display_n(3)), "[1, 2, 3, ... 2 more]");
-        assert_eq!(format!("{}", v.display_n(5)), "[1, 2, 3, 4, 5]");
-        assert_eq!(format!("{}", v.display_n(10)), "[1, 2, 3, 4, 5]");
-        assert_eq!(format!("{}", v.display_n(0)), "[... 5 more]");
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v, 3)),
+            "[1, 2, 3, ... 2 more]"
+        );
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v, 5)),
+            "[1, 2, 3, 4, 5]"
+        );
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v, 10)),
+            "[1, 2, 3, 4, 5]"
+        );
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v, 0)),
+            "[... 5 more]"
+        );
 
         let v2: Batch<i32> = 42.into();
-        assert_eq!(format!("{}", v2.display_n(0)), "[... 1 more]");
-        assert_eq!(format!("{}", v2.display_n(1)), "[42]");
+        assert_eq!(
+            format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v2, 0)),
+            "[... 1 more]"
+        );
+        assert_eq!(format!("{}", <Batch<i32> as RaftBatch<i32>>::display_n(&v2, 1)), "[42]");
     }
 }
diff --git a/openraft/src/base/batch/iter.rs b/openraft/src/base/batch/iter.rs
index 4b13aecee..bf4b9e232 100644
--- a/openraft/src/base/batch/iter.rs
+++ b/openraft/src/base/batch/iter.rs
@@ -44,6 +44,9 @@ impl<T> Iterator for BatchIter<T> {
 
 impl<T> ExactSizeIterator for BatchIter<T> {}
 
+// Safety: BatchIter<T> is Send when T is Send because both Option<T> and vec::IntoIter<T> are Send
+unsafe impl<T: Send> Send for BatchIter<T> {}
+
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/openraft/src/base/batch/mod.rs b/openraft/src/base/batch/mod.rs
index 0085f03f3..d81407d65 100644
--- a/openraft/src/base/batch/mod.rs
+++ b/openraft/src/base/batch/mod.rs
@@ -2,17 +2,20 @@
 
 mod display;
 mod iter;
+mod raft_batch;
 
 use std::ops::Index;
 use std::slice;
 
+pub use raft_batch::RaftBatch;
+
 /// A container that stores elements efficiently by avoiding heap allocation for single elements.
 ///
 /// This type uses an enum with two variants:
 /// - `Single`: stores exactly one element inline (no heap allocation)
 /// - `Vec`: stores zero or more elements using a `Vec`
 #[derive(Debug, Clone, Eq)]
-pub(crate) enum Batch<T> {
+pub enum Batch<T> {
     /// A single element stored inline without heap allocation.
     Single(T),
     /// Multiple elements stored in a Vec.
@@ -135,6 +138,66 @@ impl<T> Batch<T> {
     }
 }
 
+// Implement RaftBatch trait for Batch
+impl<T> RaftBatch<T> for Batch<T>
+where T: crate::OptionalSend + 'static + std::fmt::Debug
+{
+    type Iter<'a>
+        = slice::Iter<'a, T>
+    where T: 'a;
+    type IterMut<'a>
+        = slice::IterMut<'a, T>
+    where T: 'a;
+    type IntoIter = iter::BatchIter<T>;
+
+    fn from_item(item: T) -> Self {
+        Batch::Single(item)
+    }
+
+    fn from_vec(vec: Vec<T>) -> Self {
+        Batch::from(vec)
+    }
+
+    fn from_exact_iter<I>(iter: I) -> Self
+    where I: ExactSizeIterator<Item = T> {
+        match iter.len() {
+            0 => Batch::Vec(Vec::new()),
+            1 => Batch::Single(iter.into_iter().next().unwrap()),
+            _ => Batch::Vec(iter.collect()),
+        }
+    }
+
+    fn len(&self) -> usize {
+        Batch::len(self)
+    }
+
+    fn first(&self) -> Option<&T> {
+        Batch::first(self)
+    }
+
+    fn last(&self) -> Option<&T> {
+        Batch::last(self)
+    }
+
+    fn iter(&self) -> Self::Iter<'_> {
+        self.as_slice().iter()
+    }
+
+    fn iter_mut(&mut self) -> Self::IterMut<'_> {
+        self.as_mut_slice().iter_mut()
+    }
+
+    fn into_iter(self) -> Self::IntoIter {
+        // Use the existing IntoIterator impl
+        IntoIterator::into_iter(self)
+    }
+
+    fn extend(&mut self, other: Self) {
+        Batch::extend(self, other)
+    }
+}
+
+// Index
 impl<T> Index<usize> for Batch<T> {
     type Output = T;
 
diff --git a/openraft/src/base/batch/raft_batch.rs b/openraft/src/base/batch/raft_batch.rs
new file mode 100644
index 000000000..7bbc0f309
--- /dev/null
+++ b/openraft/src/base/batch/raft_batch.rs
@@ -0,0 +1,105 @@
+//! Trait definition for customizable batch containers in Raft.
+
+use std::fmt::Debug;
+use std::fmt::Display;
+
+use super::display::DisplayBatch;
+use crate::OptionalSend;
+
+/// Trait for batch containers used throughout Raft for efficient element grouping.
+///
+/// This trait abstracts over different batch container implementations while
+/// preserving performance characteristics required by Raft.
+///
+/// Implementations may optimize for:
+/// - Avoiding heap allocation for small batches
+/// - Arena or pool-backed allocation
+/// - Cache locality
+///
+/// # Design Notes
+///
+/// - The trait is `Sized` to enable static dispatch and full monomorphization.
+/// - Iteration is explicit via `iter()` / `iter_mut()` to preserve lifetime and `ExactSizeIterator`
+///   guarantees.
+/// - Consuming iteration via `into_iter()` method with explicit `IntoIter` associated type.
+pub trait RaftBatch<T>: OptionalSend + Sized + Default + Debug + 'static
+where T: OptionalSend + Debug + 'static
+{
+    /// Iterator type for immutable element references.
+    ///
+    /// Must implement `ExactSizeIterator` to allow efficient size queries.
+    type Iter<'a>: Iterator<Item = &'a T> + ExactSizeIterator
+    where
+        T: 'a,
+        Self: 'a;
+
+    /// Iterator type for mutable element references.
+    ///
+    /// Must implement `ExactSizeIterator` to allow efficient size queries.
+    type IterMut<'a>: Iterator<Item = &'a mut T> + ExactSizeIterator
+    where
+        T: 'a,
+        Self: 'a;
+
+    /// Iterator type for consuming iteration.
+    ///
+    /// Must implement `ExactSizeIterator` to allow efficient size queries.
+    type IntoIter: Iterator<Item = T> + ExactSizeIterator + OptionalSend;
+
+    /// Creates a batch containing a single item.
+    ///
+    /// Implementations may optimize this case to avoid heap allocation.
+    fn from_item(item: T) -> Self;
+
+    /// Creates a batch from a `Vec`.
+    ///
+    /// Implementations may optimize for the single-element case.
+    fn from_vec(vec: Vec<T>) -> Self;
+
+    /// Creates a batch from an exact-size iterator.
+    ///
+    /// The exact size allows implementations to pre-allocate or select
+    /// efficient internal representations.
+    fn from_exact_iter<I>(iter: I) -> Self
+    where I: ExactSizeIterator<Item = T>;
+
+    /// Returns the number of elements in the batch.
+    fn len(&self) -> usize;
+
+    /// Returns `true` if the batch contains no elements.
+    fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    /// Returns a reference to the first element, or `None` if empty.
+    fn first(&self) -> Option<&T>;
+
+    /// Returns a reference to the last element, or `None` if empty.
+    fn last(&self) -> Option<&T>;
+
+    /// Returns an iterator over immutable element references.
+    fn iter(&self) -> Self::Iter<'_>;
+
+    /// Returns an iterator over mutable element references.
+    fn iter_mut(&mut self) -> Self::IterMut<'_>;
+
+    /// Consumes the batch and returns an iterator over the elements.
+    fn into_iter(self) -> Self::IntoIter;
+
+    /// Appends all elements from another batch to this batch.
+    ///
+    /// This method consumes `other` to allow efficient transfer of ownership.
+    fn extend(&mut self, other: Self);
+
+    /// Returns a display helper that formats all elements.
+    fn display(&self) -> DisplayBatch<'_, T, Self>
+    where T: Display {
+        DisplayBatch::new(self, None)
+    }
+
+    /// Returns a display helper that formats at most `max` elements.
+    fn display_n(&self, max: usize) -> DisplayBatch<'_, T, Self>
+    where T: Display {
+        DisplayBatch::new(self, Some(max))
+    }
+}
diff --git a/openraft/src/base/mod.rs b/openraft/src/base/mod.rs
index dadd148aa..b47cb73e6 100644
--- a/openraft/src/base/mod.rs
+++ b/openraft/src/base/mod.rs
@@ -32,7 +32,7 @@ pub(crate) mod finalized;
 pub(crate) mod histogram;
 pub(crate) mod shared_id_generator;
 
-pub(crate) use batch::Batch;
+pub use batch::RaftBatch;
 pub use openraft_rt::BoxAny;
 pub use openraft_rt::BoxAsyncOnceMut;
 pub use openraft_rt::BoxFuture;
diff --git a/openraft/src/core/merged_raft_msg_receiver.rs b/openraft/src/core/merged_raft_msg_receiver.rs
index 397a12ae8..fcc9dd4bd 100644
--- a/openraft/src/core/merged_raft_msg_receiver.rs
+++ b/openraft/src/core/merged_raft_msg_receiver.rs
@@ -8,6 +8,7 @@
 use crate::RaftTypeConfig;
 use crate::async_runtime::MpscReceiver;
 use crate::async_runtime::TryRecvError;
+use crate::base::RaftBatch;
 use crate::core::raft_msg::RaftMsg;
 use crate::error::Fatal;
 use crate::type_config::alias::MpscReceiverOf;
@@ -196,7 +197,6 @@ where C: RaftTypeConfig
 mod tests {
     use super::*;
     use crate::async_runtime::MpscSender;
-    use crate::base::Batch;
     use crate::engine::testing::UTConfig;
     use crate::engine::testing::log_id;
     use crate::entry::EntryPayload;
@@ -211,15 +211,14 @@ mod tests {
 
     fn client_write(data: u64, leader: Option<CommittedLeaderIdOf<C>>) -> RaftMsg<C> {
         RaftMsg::ClientWrite {
-            payloads: Batch::Single(EntryPayload::Normal(data)),
-            responders: Batch::Single(None),
+            payloads: C::Batch::from_item(EntryPayload::Normal(data)),
+            responders: C::Batch::from_item(None),
             expected_leader: leader,
         }
     }
 
-    fn extract_payload_data(payloads: &Batch<EntryPayload<C>>) -> Vec<u64> {
+    fn extract_payload_data(payloads: &C::Batch<EntryPayload<C>>) -> Vec<u64> {
         payloads
-            .as_slice()
             .iter()
             .map(|p| match p {
                 EntryPayload::Normal(d) => *d,
diff --git a/openraft/src/core/raft_core.rs b/openraft/src/core/raft_core.rs
index 0d7ac0d6e..852c965f9 100644
--- a/openraft/src/core/raft_core.rs
+++ b/openraft/src/core/raft_core.rs
@@ -23,6 +23,7 @@ use crate::async_runtime::MpscReceiver;
 use crate::async_runtime::OneshotSender;
 use crate::async_runtime::TryRecvError;
 use crate::async_runtime::watch::WatchSender;
+use crate::base::RaftBatch;
 use crate::config::Config;
 use crate::config::RuntimeConfig;
 use crate::core::ClientResponderQueue;
@@ -1539,7 +1540,7 @@ where
                     }
                 }
                 self.runtime_stats.write_batch.record(payloads.len() as u64);
-                self.write_entries(payloads, responders);
+                self.write_entries(payloads.into_iter(), responders.into_iter());
             }
             RaftMsg::Initialize { members, tx } => {
                 tracing::info!("received RaftMsg::Initialize: {}, members: {:?}", func_name!(), members);
@@ -2079,7 +2080,7 @@ where
                 self.engine.state.log_progress_mut().submit(io_id.clone());
 
                 // Submit IO request, do not wait for the response.
-                self.log_store.append(entries, callback).await.sto_write_logs()?;
+                self.log_store.append(entries.into_iter(), callback).await.sto_write_logs()?;
             }
             Command::SaveVote { vote } => {
                 let io_id = IOId::new(&vote);
diff --git a/openraft/src/core/raft_msg/mod.rs b/openraft/src/core/raft_msg/mod.rs
index 3d9a78895..7ae0d240f 100644
--- a/openraft/src/core/raft_msg/mod.rs
+++ b/openraft/src/core/raft_msg/mod.rs
@@ -6,7 +6,6 @@ use super::RuntimeStats;
 use crate::ChangeMembers;
 use crate::RaftState;
 use crate::RaftTypeConfig;
-use crate::base::Batch;
 use crate::base::BoxOnce;
 use crate::core::raft_msg::external_command::ExternalCommand;
 use crate::display_ext::DisplayBTreeMapDebugValueExt;
@@ -81,8 +80,8 @@ where C: RaftTypeConfig
     },
 
     ClientWrite {
-        payloads: Batch<EntryPayload<C>>,
-        responders: Batch<Option<CoreResponder<C>>>,
+        payloads: C::Batch<EntryPayload<C>>,
+        responders: C::Batch<Option<CoreResponder<C>>>,
         expected_leader: Option<CommittedLeaderIdOf<C>>,
     },
 
diff --git a/openraft/src/engine/command.rs b/openraft/src/engine/command.rs
index b051c44b3..e30628cb2 100644
--- a/openraft/src/engine/command.rs
+++ b/openraft/src/engine/command.rs
@@ -4,7 +4,7 @@ use std::fmt::Debug;
 use crate::OptionalSend;
 use crate::RaftTypeConfig;
 use crate::async_runtime::OneshotSender;
-use crate::base::batch::Batch;
+use crate::base::RaftBatch;
 use crate::core::sm;
 use crate::display_ext::DisplayOptionExt;
 use crate::display_ext::DisplayResultExt;
@@ -64,7 +64,7 @@ where C: RaftTypeConfig
         /// [`LogIOId`]: crate::raft_state::io_state::io_id::IOId
         committed_vote: CommittedVote<C>,
 
-        entries: Batch<C::Entry>,
+        entries: C::Batch<C::Entry>,
     },
 
     /// Replicate the committed log id to other nodes
@@ -231,6 +231,7 @@ impl<C> PartialEq for Command<C>
 where
     C: RaftTypeConfig,
     C::Entry: PartialEq,
+    C::Batch<C::Entry>: PartialEq,
 {
     #[rustfmt::skip]
     fn eq(&self, other: &Self) -> bool {
diff --git a/openraft/src/engine/command_scheduler.rs b/openraft/src/engine/command_scheduler.rs
index f39a5561e..3b77e30a2 100644
--- a/openraft/src/engine/command_scheduler.rs
+++ b/openraft/src/engine/command_scheduler.rs
@@ -1,4 +1,5 @@
 use crate::RaftTypeConfig;
+use crate::base::RaftBatch;
 use crate::config::Config;
 use crate::engine::Command;
 use crate::engine::engine_output::EngineOutput;
diff --git a/openraft/src/engine/handler/following_handler/mod.rs b/openraft/src/engine/handler/following_handler/mod.rs
index 197ac43cc..e718b1353 100644
--- a/openraft/src/engine/handler/following_handler/mod.rs
+++ b/openraft/src/engine/handler/following_handler/mod.rs
@@ -4,6 +4,7 @@ use crate::EffectiveMembership;
 use crate::RaftState;
 use crate::RaftTypeConfig;
 use crate::StoredMembership;
+use crate::base::RaftBatch;
 use crate::core::sm;
 use crate::display_ext::DisplayOption;
 use crate::display_ext::DisplayOptionExt;
@@ -153,7 +154,7 @@ where C: RaftTypeConfig
         self.output.push_command(Command::AppendEntries {
             // A follower should always use the node's vote.
             committed_vote: self.leader_vote.clone(),
-            entries: entries.into(),
+            entries: C::Batch::from_vec(entries),
         });
     }
 
diff --git a/openraft/src/engine/handler/leader_handler/mod.rs b/openraft/src/engine/handler/leader_handler/mod.rs
index 4d6881624..91b0f3a1b 100644
--- a/openraft/src/engine/handler/leader_handler/mod.rs
+++ b/openraft/src/engine/handler/leader_handler/mod.rs
@@ -1,6 +1,6 @@
 use crate::RaftState;
 use crate::RaftTypeConfig;
-use crate::base::batch::Batch;
+use crate::base::RaftBatch;
 use crate::engine::Command;
 use crate::engine::EngineConfig;
 use crate::engine::EngineOutput;
@@ -61,7 +61,7 @@ where C: RaftTypeConfig
         self.state.extend_log_ids_from_same_leader(log_ids.clone());
 
         let mut membership_entry = None;
-        let entries = Batch::from_iter(payloads.zip(log_ids.clone()).map(|(payload, log_id)| {
+        let entries = C::Batch::from_exact_iter(payloads.zip(log_ids.clone()).map(|(payload, log_id)| {
             tracing::debug!("assign log id: {}", log_id);
             let entry = C::Entry::new(log_id, payload);
             if let Some(m) = entry.get_membership() {
diff --git a/openraft/src/impls/mod.rs b/openraft/src/impls/mod.rs
index 4b0a28d27..4abfab21b 100644
--- a/openraft/src/impls/mod.rs
+++ b/openraft/src/impls/mod.rs
@@ -5,6 +5,7 @@
 //!
 //! ## Key Types
 //!
+//! - [`Batch`] - Default batch container implementation
 //! - [`Entry`] - Default log entry implementation
 //! - [`LogId`] - Default log identifier
 //! - [`Vote`] - Default vote implementation
@@ -31,6 +32,8 @@ pub use boxed_error_source::BoxedErrorSource;
 #[deprecated(since = "0.10.0", note = "use `openraft_rt_tokio::TokioRuntime` directly")]
 pub use openraft_rt_tokio::TokioRuntime;
 
+// Re-export Batch from base as a default implementation
+pub use crate::base::batch::Batch;
 pub use crate::entry::Entry;
 pub use crate::node::BasicNode;
 pub use crate::node::EmptyNode;
diff --git a/openraft/src/lib.rs b/openraft/src/lib.rs
index 99efa2691..bb259a872 100644
--- a/openraft/src/lib.rs
+++ b/openraft/src/lib.rs
@@ -106,6 +106,7 @@ use crate::base::OptionalFeatures;
 pub use crate::base::OptionalSend;
 pub use crate::base::OptionalSerde;
 pub use crate::base::OptionalSync;
+pub use crate::base::RaftBatch;
 pub use crate::change_members::ChangeMembers;
 pub use crate::config::Config;
 pub use crate::config::ConfigError;
diff --git a/openraft/src/raft/api/app.rs b/openraft/src/raft/api/app.rs
index a9d0ef1fd..386ee4ca8 100644
--- a/openraft/src/raft/api/app.rs
+++ b/openraft/src/raft/api/app.rs
@@ -4,8 +4,8 @@ use openraft_macros::since;
 
 use crate::RaftTypeConfig;
 use crate::ReadPolicy;
-use crate::base::Batch;
 use crate::base::BoxStream;
+use crate::base::RaftBatch;
 use crate::core::raft_msg::RaftMsg;
 use crate::entry::EntryPayload;
 use crate::error::ClientWriteError;
@@ -60,8 +60,8 @@ where C: RaftTypeConfig
         let (responder, _commit_rx, complete_rx) = ProgressResponder::new();
 
         self.do_client_write_ff(
-            Batch::from(payload),
-            Batch::from(Some(CoreResponder::Progress(responder))),
+            C::Batch::from_item(payload),
+            C::Batch::from_item(Some(CoreResponder::Progress(responder))),
         )
         .await?;
 
@@ -78,8 +78,8 @@ where C: RaftTypeConfig
         responder: Option<WriteResponderOf<C>>,
     ) -> Result<(), Fatal<C>> {
         self.do_client_write_ff(
-            Batch::from(payload),
-            Batch::from(responder.map(|r| CoreResponder::UserDefined(r))),
+            C::Batch::from_item(payload),
+            C::Batch::from_item(responder.map(|r| CoreResponder::UserDefined(r))),
         )
         .await
     }
@@ -88,8 +88,8 @@ where C: RaftTypeConfig
     #[since(version = "0.10.0")]
     async fn do_client_write_ff(
         &self,
-        payloads: Batch<EntryPayload<C>>,
-        responders: Batch<Option<CoreResponder<C>>>,
+        payloads: C::Batch<EntryPayload<C>>,
+        responders: C::Batch<Option<CoreResponder<C>>>,
     ) -> Result<(), Fatal<C>> {
         self.inner
             .send_msg(RaftMsg::ClientWrite {
@@ -126,7 +126,7 @@ where C: RaftTypeConfig
             receivers.push(complete_rx);
         }
 
-        self.do_client_write_ff(Batch::from(payloads), Batch::from(responders)).await?;
+        self.do_client_write_ff(C::Batch::from_vec(payloads), C::Batch::from_vec(responders)).await?;
 
         let stream = futures_util::stream::unfold(Some(receivers.into_iter()), |opt_iter| async move {
             let mut iter = opt_iter?;
diff --git a/openraft/src/raft/message/write_request.rs b/openraft/src/raft/message/write_request.rs
index db44991a3..f465f6634 100644
--- a/openraft/src/raft/message/write_request.rs
+++ b/openraft/src/raft/message/write_request.rs
@@ -3,8 +3,8 @@ use std::future::IntoFuture;
 use openraft_macros::since;
 
 use crate::RaftTypeConfig;
-use crate::base::Batch;
 use crate::base::BoxFuture;
+use crate::base::RaftBatch;
 use crate::core::raft_msg::RaftMsg;
 use crate::entry::EntryPayload;
 use crate::error::Fatal;
@@ -138,8 +138,8 @@ where C: RaftTypeConfig
         Box::pin(async move {
             self.inner
                 .send_msg(RaftMsg::ClientWrite {
-                    payloads: Batch::from(EntryPayload::Normal(self.app_data)),
-                    responders: Batch::from(self.responder),
+                    payloads: C::Batch::from_item(EntryPayload::Normal(self.app_data)),
+                    responders: C::Batch::from_item(self.responder),
                     expected_leader: self.expected_leader,
                 })
                 .await
diff --git a/openraft/src/raft/mod.rs b/openraft/src/raft/mod.rs
index 51de33bc5..3a64b3024 100644
--- a/openraft/src/raft/mod.rs
+++ b/openraft/src/raft/mod.rs
@@ -224,6 +224,7 @@ macro_rules! declare_raft_types {
                 (Vote           , , $crate::impls::Vote<Self>                    ),
                 (Entry          , , $crate::impls::Entry<Self>                   ),
                 (SnapshotData   , , std::io::Cursor<Vec<u8>>                     ),
+                (Batch<T>       , , $crate::impls::Batch<T> where T: $crate::OptionalSend + 'static + std::fmt::Debug ),
                 (Responder<T>   , , $crate::impls::ProgressResponder<Self, T> where T: $crate::OptionalSend + 'static     ),
                 (AsyncRuntime   , , $crate::impls::TokioRuntime                  ),
                 (ErrorSource    , , $crate::impls::BoxedErrorSource               ),
diff --git a/openraft/src/raft/responder/core_responder.rs b/openraft/src/raft/responder/core_responder.rs
index b7617e2f9..f23bce08e 100644
--- a/openraft/src/raft/responder/core_responder.rs
+++ b/openraft/src/raft/responder/core_responder.rs
@@ -16,6 +16,17 @@ where C: RaftTypeConfig
     UserDefined(WriteResponderOf<C>),
 }
 
+impl<C> std::fmt::Debug for CoreResponder<C>
+where C: RaftTypeConfig
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Progress(_) => f.debug_tuple("Progress").field(&"<responder>").finish(),
+            Self::UserDefined(_) => f.debug_tuple("UserDefined").field(&"<responder>").finish(),
+        }
+    }
+}
+
 impl<C> Responder<C, ClientWriteResult<C>> for CoreResponder<C>
 where C: RaftTypeConfig
 {
diff --git a/openraft/src/type_config.rs b/openraft/src/type_config.rs
index 2a85f45d7..d8c1be1ae 100644
--- a/openraft/src/type_config.rs
+++ b/openraft/src/type_config.rs
@@ -123,6 +123,49 @@ pub trait RaftTypeConfig:
     /// [sto]: crate::docs::getting_started#3-implement-raftlogstorage-and-raftstatemachine
     type SnapshotData: OptionalSend + 'static;
 
+    /// Batch container type for grouping multiple items efficiently.
+    ///
+    /// The batch type is used throughout Raft to store collections of entries, data items,
+    /// and responders efficiently. Implementations can optimize for different use cases:
+    ///
+    /// - Memory allocation strategies (arena allocators, pre-allocated pools)
+    /// - Cache locality optimization
+    /// - Minimizing heap allocations for small batches
+    ///
+    /// The default implementation ([`Batch`](crate::impls::Batch)) uses an enum that stores
+    /// single elements inline without heap allocation, and multiple elements in a `Vec`.
+    ///
+    /// # Examples
+    ///
+    /// Using the default batch implementation:
+    ///
+    /// ```ignore
+    /// openraft::declare_raft_types!(
+    ///     pub MyTypeConfig:
+    ///         D = String,
+    ///         R = String,
+    /// );
+    /// // Batch<T> defaults to openraft::impls::Batch<T>
+    /// ```
+    ///
+    /// Providing a custom batch implementation:
+    ///
+    /// ```ignore
+    /// openraft::declare_raft_types!(
+    ///     pub MyTypeConfig:
+    ///         D = String,
+    ///         R = String,
+    ///         Batch<T> = MyCustomBatch<T> where T: OptionalSend + 'static,
+    /// );
+    /// ```
+    ///
+    /// # See Also
+    ///
+    /// - [`RaftBatch`](crate::base::RaftBatch) - The trait that batch implementations must satisfy
+    /// - [`Batch`](crate::impls::Batch) - The default implementation
+    type Batch<T>: crate::base::RaftBatch<T>
+    where T: OptionalSend + 'static + Debug;
+
     /// Asynchronous runtime type.
     type AsyncRuntime: AsyncRuntime;
 
@@ -183,6 +226,7 @@ pub mod alias {
     pub type VoteOf<C> = <C as RaftTypeConfig>::Vote;
     pub type EntryOf<C> = <C as RaftTypeConfig>::Entry;
     pub type SnapshotDataOf<C> = <C as RaftTypeConfig>::SnapshotData;
+    pub type BatchOf<C, T> = <C as RaftTypeConfig>::Batch<T>;
     pub type AsyncRuntimeOf<C> = <C as RaftTypeConfig>::AsyncRuntime;
     pub type ResponderOf<C, T> = <C as RaftTypeConfig>::Responder<T>;
     pub type ErrorSourceOf<C> = <C as RaftTypeConfig>::ErrorSource;