This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch server_perf_cut_extra_work
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/server_perf_cut_extra_work by
this push:
new a64a966c5 fix(cluster): bound polling and repair resource use
a64a966c5 is described below
commit a64a966c576e786e3e747bd60bfc9644a4290c29
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Mon Sep 14 21:02:58 2026 +0200
fix(cluster): bound polling and repair resource use
Oversized batches widened later reads, WAL cleanup could fall behind,
and out-of-order repair eviction discarded newer operations. Cap read
growth, drain each reclaim snapshot, and index retained repairs by op.
Separate client queues from replica capacity, back off repeated route
refusals, and reject empty Go rosters. Add regression coverage and SDK
upgrade guidance while retaining session fencing.
---
.../src/requests/consumer_offsets/mod.rs | 62 ++++++++++++
core/configs/src/server_config/defaults.rs | 1 +
core/configs/src/server_config/message_bus.rs | 38 ++++++-
core/consensus/src/client_table.rs | 2 +
core/journal/src/partition_journal.rs | 50 ++++++---
core/message_bus/src/config.rs | 6 +-
core/message_bus/src/installer/tcp.rs | 4 +-
core/partitions/src/journal.rs | 112 +++++++++++++++------
core/partitions/src/poll_plan.rs | 108 +++++++++++++++-----
core/sdk/Cargo.toml | 1 +
core/sdk/src/poll_routing.rs | 64 +++++++++++-
core/sdk/src/quic/quic_client.rs | 6 +-
core/sdk/src/tcp/tcp_client.rs | 6 +-
core/sdk/src/websocket/websocket_client.rs | 6 +-
core/server/config.toml | 6 +-
core/server/src/dispatch/mod.rs | 5 +-
core/server/src/http/forward.rs | 2 +-
foreign/csharp/README.md | 7 ++
foreign/go/README.md | 6 +-
foreign/go/client/tcp/cluster.go | 5 +
foreign/go/client/tcp/tcp_poll_routing_test.go | 65 +++++++-----
foreign/java/README.md | 8 ++
.../iggy/client/async/tcp/MessagesTcpClient.java | 5 +
foreign/node/README.md | 7 ++
24 files changed, 458 insertions(+), 124 deletions(-)
diff --git a/core/binary_protocol/src/requests/consumer_offsets/mod.rs
b/core/binary_protocol/src/requests/consumer_offsets/mod.rs
index bde4d02b5..cda0a9ceb 100644
--- a/core/binary_protocol/src/requests/consumer_offsets/mod.rs
+++ b/core/binary_protocol/src/requests/consumer_offsets/mod.rs
@@ -22,3 +22,65 @@ pub mod store_consumer_offset;
pub use delete_consumer_offset::DeleteConsumerOffsetRequest;
pub use get_consumer_offset::GetConsumerOffsetRequest;
pub use store_consumer_offset::StoreConsumerOffsetRequest;
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::WireIdentifier;
+ use crate::codec::{WireDecode, WireEncode};
+ use crate::primitives::ack_level::AckLevel;
+ use crate::primitives::consumer::WireConsumer;
+
+ #[test]
+ fn offset_writes_share_the_routing_request_prefix() {
+ let identifiers = [
+ WireIdentifier::numeric(7),
+ WireIdentifier::named("named-resource").unwrap(),
+ ];
+ for consumer_id in &identifiers {
+ for consumer in [
+ WireConsumer::consumer(consumer_id.clone()),
+ WireConsumer::consumer_group(consumer_id.clone()),
+ ] {
+ for stream_id in &identifiers {
+ for topic_id in &identifiers {
+ for partition_id in [None, Some(0), Some(u32::MAX)] {
+ let route = GetConsumerOffsetRequest {
+ consumer: consumer.clone(),
+ stream_id: stream_id.clone(),
+ topic_id: topic_id.clone(),
+ partition_id,
+ };
+ for ack in [AckLevel::NoAck, AckLevel::Quorum] {
+ let store = StoreConsumerOffsetRequest {
+ consumer: consumer.clone(),
+ stream_id: stream_id.clone(),
+ topic_id: topic_id.clone(),
+ partition_id,
+ offset: u64::MAX,
+ ack,
+ }
+ .to_bytes();
+ let delete = DeleteConsumerOffsetRequest {
+ consumer: consumer.clone(),
+ stream_id: stream_id.clone(),
+ topic_id: topic_id.clone(),
+ partition_id,
+ ack,
+ }
+ .to_bytes();
+ for payload in [store, delete] {
+ let (decoded, consumed) =
+
GetConsumerOffsetRequest::decode(&payload).unwrap();
+ assert_eq!(decoded, route);
+ assert_eq!(consumed, route.encoded_size());
+ assert_eq!(&payload[..consumed],
route.to_bytes().as_ref());
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/core/configs/src/server_config/defaults.rs
b/core/configs/src/server_config/defaults.rs
index feb83d5f5..6019a8b0d 100644
--- a/core/configs/src/server_config/defaults.rs
+++ b/core/configs/src/server_config/defaults.rs
@@ -297,6 +297,7 @@ impl Default for MessageBusConfig {
max_batch: bus.max_batch as usize,
max_message_size: bus.max_message_size.parse().unwrap(),
peer_queue_capacity: bus.peer_queue_capacity as usize,
+ client_queue_capacity: bus.client_queue_capacity as usize,
reconnect_period: bus.reconnect_period.parse().unwrap(),
close_peer_timeout: bus.close_peer_timeout.parse().unwrap(),
close_grace: bus.close_grace.parse().unwrap(),
diff --git a/core/configs/src/server_config/message_bus.rs
b/core/configs/src/server_config/message_bus.rs
index 77c2bbf9d..b37c17dd0 100644
--- a/core/configs/src/server_config/message_bus.rs
+++ b/core/configs/src/server_config/message_bus.rs
@@ -70,6 +70,8 @@ use serde_with::{DisplayFromStr, serde_as};
/// failure until both are reconciled.
pub const IOV_MAX_LIMIT: usize = 512;
+const DEFAULT_CLIENT_QUEUE_CAPACITY: usize = 256;
+
/// Tunables for the message bus that ships consensus traffic between
/// replicas and SDK-client traffic between shards.
#[serde_as]
@@ -86,11 +88,16 @@ pub struct MessageBusConfig {
#[config_env(leaf)]
pub max_message_size: IggyByteSize,
- /// Bound on the per-peer mpsc queue. Writer task drains; the
+ /// Bound on each replica peer's mpsc queue. Writer task drains; the
/// `send_to_*` path enqueues. Too small drops under burst; too
/// large delays backpressure signalling.
pub peer_queue_capacity: usize,
+ /// Bound on each SDK connection's inbound and outbound queues. Separate
+ /// from replica queues so repair bursts do not enlarge every client
allocation.
+ #[serde(default = "default_client_queue_capacity")]
+ pub client_queue_capacity: usize,
+
/// Interval between outbound reconnect attempts to peers with
/// `peer_id > self_id`.
#[config_env(leaf)]
@@ -141,6 +148,10 @@ impl Validatable<ConfigurationError> for MessageBusConfig {
eprintln!("{COMPONENT} message_bus.peer_queue_capacity must be >
0");
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.client_queue_capacity == 0 {
+ eprintln!("{COMPONENT} message_bus.client_queue_capacity must be >
0");
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
if self.max_message_size.as_bytes_u64() == 0 {
eprintln!("{COMPONENT} message_bus.max_message_size must be > 0");
return Err(ConfigurationError::InvalidConfigurationValue);
@@ -175,6 +186,10 @@ impl Validatable<ConfigurationError> for MessageBusConfig {
}
}
+const fn default_client_queue_capacity() -> usize {
+ DEFAULT_CLIENT_QUEUE_CAPACITY
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -216,6 +231,27 @@ mod tests {
assert!(c.validate().is_err());
}
+ #[test]
+ fn missing_client_queue_capacity_keeps_independent_default() {
+ let mut config = serde_json::to_value(baseline()).unwrap();
+ config
+ .as_object_mut()
+ .unwrap()
+ .remove("client_queue_capacity");
+ config["peer_queue_capacity"] = serde_json::json!(8192);
+ let decoded: MessageBusConfig =
serde_json::from_value(config).unwrap();
+ assert_eq!(decoded.client_queue_capacity,
DEFAULT_CLIENT_QUEUE_CAPACITY);
+ assert_eq!(decoded.peer_queue_capacity, 8192);
+ decoded.validate().unwrap();
+ }
+
+ #[test]
+ fn rejects_zero_client_queue_capacity() {
+ let mut config = baseline();
+ config.client_queue_capacity = 0;
+ assert!(config.validate().is_err());
+ }
+
#[test]
fn rejects_zero_max_message_size() {
let mut c = baseline();
diff --git a/core/consensus/src/client_table.rs
b/core/consensus/src/client_table.rs
index b78bc8914..c72ef8e34 100644
--- a/core/consensus/src/client_table.rs
+++ b/core/consensus/src/client_table.rs
@@ -1676,6 +1676,8 @@ impl ClientTable {
/// Attach only after the caller has authenticated `user_id` and waited
/// for the local metadata frontier to cover the requested session.
+ /// The registered user owns the session, matching authenticated login
+ /// resume. Client ids and epochs are identifiers, not authentication
secrets.
pub fn attach_session(
&mut self,
client_id: u128,
diff --git a/core/journal/src/partition_journal.rs
b/core/journal/src/partition_journal.rs
index ece6dc0c9..5f783df54 100644
--- a/core/journal/src/partition_journal.rs
+++ b/core/journal/src/partition_journal.rs
@@ -1043,18 +1043,11 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
Ok(())
}
- /// Drop every file the recovered history does not retain, repeating while
- /// the queue shrinks: `cleanup_obsolete` removes a bounded batch per call
- /// and re-queues what it could not remove.
+ /// Drop every file the recovered history does not retain.
async fn remove_obsolete_history(&mut self) -> io::Result<()> {
self.discover_obsolete().await?;
- loop {
- let remaining = self.obsolete.len();
- self.cleanup_obsolete().await;
- if self.obsolete.is_empty() || self.obsolete.len() == remaining {
- return Ok(());
- }
- }
+ self.cleanup_obsolete().await;
+ Ok(())
}
async fn discover_obsolete(&mut self) -> io::Result<()> {
@@ -1079,8 +1072,8 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
Ok(())
}
- /// Remove a bounded batch of the files no published generation retains,
- /// and retry a directory barrier a previous batch could not complete.
+ /// Attempt every queued obsolete file once, and retry a directory barrier
+ /// a previous pass could not complete. Failed unlinks stay queued.
/// Does nothing when the queue is empty and no barrier is owed.
///
/// Off the append path, so the unlinks and the barrier do not sit inside a
@@ -1090,7 +1083,7 @@ impl<S: DurableStorage> PartitionPrepareJournal<S> {
/// still reachable by a reader or a recovery. The writer owns the journal,
/// so this runs between mutations and never beside one.
pub async fn cleanup_obsolete(&mut self) {
- let count = self.obsolete.len().min(16);
+ let count = self.obsolete.len();
for _ in 0..count {
let Some(path) = self.obsolete.pop_front() else {
break;
@@ -1873,6 +1866,37 @@ mod tests {
assert!(!stale.exists());
}
+ #[compio::test]
+ async fn cleanup_reclaims_all_generations_queued_between_writer_passes() {
+ const GENERATIONS: u64 = 64;
+ let partition = tempdir().unwrap();
+ let directory = partition.path().join("prepares-7");
+ let mut journal = PartitionPrepareJournal::open(&directory, 42, 7)
+ .await
+ .unwrap();
+ let mut parent = 0;
+ for op in 1..=GENERATIONS {
+ let entry = prepare(op, parent);
+ parent = entry.header().checksum;
+ journal.append(entry.into_frozen()).await.unwrap();
+ journal.checkpoint(op).await.unwrap();
+ }
+ let retired: Vec<_> = journal.obsolete.iter().cloned().collect();
+ assert_eq!(retired.len(), usize::try_from(GENERATIONS).unwrap());
+ journal.cleanup_obsolete().await;
+ assert!(
+ journal.obsolete.is_empty(),
+ "one reclaim pass must keep up with a writer's accumulated
generations"
+ );
+ assert!(retired.iter().all(|path| !path.exists()));
+ drop(journal);
+ let journal = PartitionPrepareJournal::open(&directory, 42, 7)
+ .await
+ .unwrap();
+ assert_eq!(journal.checkpoint_op(), GENERATIONS);
+ assert_eq!(journal.head(), GENERATIONS);
+ }
+
#[compio::test]
async fn a_rewrite_reclaims_the_wal_path_it_reuses() {
let partition = tempdir().unwrap();
diff --git a/core/message_bus/src/config.rs b/core/message_bus/src/config.rs
index be060b333..19731b4f7 100644
--- a/core/message_bus/src/config.rs
+++ b/core/message_bus/src/config.rs
@@ -138,11 +138,14 @@ pub struct MessageBusConfig {
/// validator; undersize or oversize frames are rejected.
pub max_message_size: usize,
- /// Bound on the per-peer mpsc queue. The writer task drains; the
+ /// Bound on each replica peer's mpsc queue. The writer task drains; the
/// `send_to_*` path enqueues. Too small drops under burst; too
/// large delays backpressure signalling.
pub peer_queue_capacity: usize,
+ /// Bound on each SDK connection's inbound and outbound queues.
+ pub client_queue_capacity: usize,
+
/// Interval between outbound reconnect attempts to peers with
/// `peer_id > self_id`.
pub reconnect_period: Duration,
@@ -214,6 +217,7 @@ impl From<&ServerConfig> for MessageBusConfig {
max_message_size:
usize::try_from(bus.max_message_size.as_bytes_u64())
.expect("message_bus.max_message_size fits usize on supported
targets"),
peer_queue_capacity: bus.peer_queue_capacity,
+ client_queue_capacity: bus.client_queue_capacity,
reconnect_period: bus.reconnect_period.get_duration(),
mesh_expected_peers: if cfg.cluster.enabled {
cfg.cluster.nodes.len().saturating_sub(1)
diff --git a/core/message_bus/src/installer/tcp.rs
b/core/message_bus/src/installer/tcp.rs
index 12cfffbde..67bfc091e 100644
--- a/core/message_bus/src/installer/tcp.rs
+++ b/core/message_bus/src/installer/tcp.rs
@@ -69,9 +69,9 @@ pub fn install_client_conn<C: TransportConn>(
on_request: RequestHandler,
) {
let client_id = meta.client_id;
- let (tx, rx) = async_channel::bounded(bus.peer_queue_capacity());
+ let (tx, rx) = async_channel::bounded(bus.config().client_queue_capacity);
let (in_tx, in_rx) =
-
async_channel::bounded::<Message<GenericHeader>>(bus.peer_queue_capacity());
+
async_channel::bounded::<Message<GenericHeader>>(bus.config().client_queue_capacity);
// If the registry insert below loses a race for `client_id`, the
// losing reader must NOT invoke `on_request` (it would route
diff --git a/core/partitions/src/journal.rs b/core/partitions/src/journal.rs
index b277a7c4d..96c1600c8 100644
--- a/core/partitions/src/journal.rs
+++ b/core/partitions/src/journal.rs
@@ -24,7 +24,7 @@ use server_common::{
use std::io;
use std::{
cell::{Cell, UnsafeCell},
- collections::{BTreeMap, HashMap, VecDeque},
+ collections::{BTreeMap, HashMap},
ops::RangeInclusive,
};
use tracing::warn;
@@ -185,7 +185,7 @@ where
/// entries left the resident journal at flush. Bounded by
/// [`EVICTED_RING_CAPACITY`]; requests older than the ring answer
/// `RangeEvicted` honestly.
- evicted_ring: UnsafeCell<VecDeque<(u64, JournalBuffer)>>,
+ evicted_ring: UnsafeCell<BTreeMap<u64, JournalBuffer>>,
/// Running byte total of the buffers held by `evicted_ring`.
evicted_ring_bytes: Cell<u64>,
/// Entry-count ceiling for `evicted_ring`. Defaults to
@@ -231,7 +231,7 @@ where
inner: UnsafeCell::new(JournalInner {
storage: S::default(),
}),
- evicted_ring: UnsafeCell::new(VecDeque::new()),
+ evicted_ring: UnsafeCell::new(BTreeMap::new()),
evicted_ring_bytes: Cell::new(0),
evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY),
evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX),
@@ -362,7 +362,7 @@ impl PartitionJournal<PartitionJournalMemStorage> {
let ring = unsafe { &mut *self.evicted_ring.get() };
debug_assert!(ring.is_empty());
self.evicted_ring_bytes.set(prepare.len() as u64);
- ring.push_back((op, prepare));
+ ring.insert(op, prepare);
}
/// Entry bytes for `op`, from the resident journal or the evicted ring.
@@ -377,9 +377,7 @@ impl PartitionJournal<PartitionJournalMemStorage> {
}
}
let ring = unsafe { &*self.evicted_ring.get() };
- ring.iter()
- .find(|(ring_op, _)| *ring_op == op)
- .map(|(_, entry)| entry.clone())
+ ring.get(&op).cloned()
}
/// The header at `op`, over exactly the range [`Self::repair_entry`]
serves.
@@ -393,32 +391,32 @@ impl PartitionJournal<PartitionJournalMemStorage> {
/// The entry is still servable from the evicted ring, which is what makes
/// the blank wrong rather than merely pessimistic.
///
- /// The ring drops from the front, so the highest evicted op -- the commit
- /// point of the last flush -- is the last thing it forgets.
+ /// Retention drops the lowest op, including when repair backfills arrive
+ /// out of order, so the commit point is the last thing it forgets.
pub fn repair_header(&self, op: u64) -> Option<PrepareHeader> {
if let Some(header) = self.header_by_op(op) {
return Some(header);
}
let ring = unsafe { &*self.evicted_ring.get() };
- let (_, entry) = ring.iter().find(|(ring_op, _)| *ring_op == op)?;
+ let entry = ring.get(&op)?;
let header_bytes = entry.as_slice().get(..PREPARE_HEADER_SIZE)?;
bytemuck::checked::try_from_bytes::<PrepareHeader>(header_bytes)
.ok()
.copied()
}
- /// Every repairable header with an op in `ops`, in ONE pass over the
resident
- /// headers and ONE over the evicted ring.
+ /// Every repairable header with an op in `ops`, scanning resident headers
+ /// once and seeking the retained repair range by op.
///
- /// [`Self::repair_header`] is two linear scans, so probing it per op costs
- /// O(window x (headers + ring)), and the `DoViewChange` suffix build does
- /// exactly that, up to `DVC_HEADERS_MAX` probes, on every SVC/DVC arrival
and
- /// non-Normal tick, on the pump. Result size is bounded by what the
journal
- /// holds, not by the width of `ops`. Resident wins over ring, as
`repair_header`
- /// probes.
+ /// Avoid repeating the resident header scan for each op in a view-change
+ /// suffix. Result size is bounded by what the journal holds, not by the
+ /// width of `ops`. Resident entries take precedence over retained repairs.
#[must_use]
pub fn repair_headers_in(&self, ops: RangeInclusive<u64>) -> BTreeMap<u64,
PrepareHeader> {
let mut found = BTreeMap::new();
+ if ops.is_empty() {
+ return found;
+ }
{
let headers = unsafe { &*self.headers.get() };
for header in headers.iter().filter(|header|
ops.contains(&header.op)) {
@@ -426,7 +424,7 @@ impl PartitionJournal<PartitionJournalMemStorage> {
}
}
let ring = unsafe { &*self.evicted_ring.get() };
- for (op, entry) in ring.iter().filter(|(op, _)| ops.contains(op)) {
+ for (op, entry) in ring.range(ops) {
if found.contains_key(op) {
continue;
}
@@ -440,17 +438,16 @@ impl PartitionJournal<PartitionJournalMemStorage> {
found
}
- /// Oldest op this journal can still serve for repair (ring front, else
- /// resident head), or `None` when it holds nothing at all.
+ /// Oldest op this journal can still serve for repair, including resident
+ /// backfills older than the retained repair range.
pub fn repair_retained_from(&self) -> Option<u64> {
- {
- let ring = unsafe { &*self.evicted_ring.get() };
- if let Some((op, _)) = ring.front() {
- return Some(*op);
- }
- }
- let headers = unsafe { &*self.headers.get() };
- headers.first().map(|header| header.op)
+ let ring = unsafe { &*self.evicted_ring.get() };
+ let resident = unsafe { &*self.op_to_storage_offset.get() };
+ ring.first_key_value()
+ .map(|(op, _)| *op)
+ .into_iter()
+ .chain(resident.first_key_value().map(|(op, _)| *op))
+ .min()
}
/// Synchronous resident-range poll read. Never awaits (mem storage reads
@@ -602,11 +599,13 @@ impl PartitionJournal<PartitionJournalMemStorage> {
break;
};
ring_bytes += entry.len() as u64;
- ring.push_back((op, entry));
+ if let Some(previous) = ring.insert(op, entry) {
+ ring_bytes -= previous.len() as u64;
+ }
while ring.len() > self.evicted_ring_capacity.get()
|| (ring_bytes > self.evicted_ring_bytes_max.get() &&
ring.len() > 1)
{
- if let Some((_, dropped)) = ring.pop_front() {
+ if let Some((_, dropped)) = ring.pop_first() {
ring_bytes -= dropped.len() as u64;
}
}
@@ -765,7 +764,7 @@ where
timestamp_to_op: UnsafeCell::new(BTreeMap::new()),
headers: UnsafeCell::new(Vec::new()),
inner: UnsafeCell::new(JournalInner { storage }),
- evicted_ring: UnsafeCell::new(VecDeque::new()),
+ evicted_ring: UnsafeCell::new(BTreeMap::new()),
evicted_ring_bytes: Cell::new(0),
evicted_ring_capacity: Cell::new(EVICTED_RING_CAPACITY),
evicted_ring_bytes_max: Cell::new(EVICTED_RING_BYTES_MAX),
@@ -1495,6 +1494,55 @@ mod tests {
assert!(!shape.holds_messages);
}
+ #[compio::test]
+ async fn
repair_retention_keeps_the_newest_ops_after_out_of_order_backfill() {
+ const CAPACITY: usize = 4;
+ let journal =
PartitionJournal::<PartitionJournalMemStorage>::default();
+ journal.set_ring_caps(CAPACITY, u64::MAX);
+ for op in [4, 2, 8, 6, 3, 1, 7, 5] {
+ journal
+ .append(build_prepare(op, HEADER_SIZE + 16).into_frozen())
+ .await
+ .unwrap();
+ }
+ journal.evict_prefix(8).await;
+ assert_eq!(journal.evicted_ring_occupancy().0, CAPACITY);
+ assert_eq!(journal.repair_retained_from(), Some(5));
+ for op in 1..=4 {
+ assert!(journal.repair_entry(op).is_none());
+ assert!(journal.repair_header(op).is_none());
+ }
+ for op in 5..=8 {
+ assert_eq!(journal.repair_header(op).unwrap().op, op);
+ assert_eq!(
+ journal.repair_entry(op).unwrap().as_slice(),
+ build_prepare(op, HEADER_SIZE + 16).as_slice()
+ );
+ }
+ assert_eq!(
+ journal
+ .repair_headers_in(6..=7)
+ .into_keys()
+ .collect::<Vec<_>>(),
+ vec![6, 7]
+ );
+ let empty_start = 7;
+ assert!(journal.repair_headers_in(empty_start..=6).is_empty());
+ for op in [3, 1] {
+ journal
+ .append(build_prepare(op, HEADER_SIZE + 16).into_frozen())
+ .await
+ .unwrap();
+ }
+ assert_eq!(journal.repair_retained_from(), Some(1));
+ journal.evict_prefix(8).await;
+ assert_eq!(journal.repair_retained_from(), Some(5));
+ assert_eq!(journal.evicted_ring_occupancy().0, CAPACITY);
+ journal.clear_all();
+ assert_eq!(journal.evicted_ring_occupancy(), (0, 0));
+ assert!(journal.repair_entry(8).is_none());
+ }
+
#[compio::test]
async fn repair_headers_in_serves_the_commit_point_from_the_evicted_ring()
{
// Blank AT the commit point is the one slot a merge can neither adopt
nor
diff --git a/core/partitions/src/poll_plan.rs b/core/partitions/src/poll_plan.rs
index 5348477cb..a4397f371 100644
--- a/core/partitions/src/poll_plan.rs
+++ b/core/partitions/src/poll_plan.rs
@@ -395,10 +395,9 @@ pub enum DiskReadOutcome {
Faulted,
}
-/// Largest first read of a disk poll, and the size every poll used to read
-/// whatever it asked for. A batch wider than this still grows past it through
-/// the re-read path below; this bounds only where a walk starts.
-pub const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
+/// Ceiling for ordinary disk reads. An incomplete batch may require one
+/// larger re-read, without widening subsequent chunks or segments.
+const DISK_POLL_CHUNK_MAX: u64 = 1 << 20;
/// Smallest first read of a disk poll. Below this the syscall and the segment
/// walk cost more than the bytes the smaller read saves, and a poll for a
@@ -426,7 +425,7 @@ struct DiskWalk {
matched: u32,
fragments: PollFragments<4096>,
last_matching_offset: Option<u64>,
- /// Batch width learned from an incomplete read, retained across segments.
+ /// Batch width learned from an incomplete read, capped at the chunk
ceiling.
batch_read_floor: u64,
#[cfg(feature = "poll-diagnostics")]
requested_bytes: u64,
@@ -691,7 +690,9 @@ impl DiskReadPlan {
return SegmentWalk::Faulted;
}
chunk_len = if needed > len {
- walk.batch_read_floor = walk.batch_read_floor.max(needed
as u64);
+ walk.batch_read_floor = walk
+ .batch_read_floor
+ .max((needed as u64).min(DISK_POLL_CHUNK_MAX));
needed as u64
} else {
chunk_len.saturating_mul(4)
@@ -1154,25 +1155,10 @@ mod tests {
async fn incomplete_batch_reread_keeps_its_exact_length_for_the_walk() {
const BATCH_COUNT: u32 = 4;
let directory = tempfile::tempdir().unwrap();
- let mut messages = IggyMessages::with_capacity(1);
- messages.push(IggyMessage {
- header: IggyMessageHeader {
- payload_length: 128 << 10,
- ..Default::default()
- },
- payload: Bytes::from(vec![1; 128 << 10]),
- user_headers: None,
- });
- let batch =
- SendMessagesOwned::from_messages(IggyNamespace::new(1, 1, 0),
&messages).unwrap();
- let length = batch.header.total_size();
- let mut records = vec![0; length * BATCH_COUNT as usize];
- for (offset, record) in records.chunks_exact_mut(length).enumerate() {
- let mut header = batch.header;
- header.base_offset = offset as u64;
- header.batch_checksum = header.checksum_for_blob(&batch.blob);
- header.encode_into(record);
- record[COMMAND_HEADER_SIZE..].copy_from_slice(&batch.blob);
+ let length = disk_batch(128 << 10, 0).len();
+ let mut records = Vec::with_capacity(length * BATCH_COUNT as usize);
+ for offset in 0..BATCH_COUNT {
+ records.extend_from_slice(&disk_batch(128 << 10,
u64::from(offset)));
}
let mut file =
compio::fs::File::create(directory.path().join("batches.log"))
.await
@@ -1207,6 +1193,78 @@ mod tests {
);
}
+ #[cfg(feature = "poll-diagnostics")]
+ #[compio::test]
+ async fn oversized_batch_does_not_widen_later_reads_or_the_next_segment() {
+ const SMALL_BATCHES: u64 = 16;
+ for separate_segments in [false, true] {
+ let directory = tempfile::tempdir().unwrap();
+ let mut wide = disk_batch(3 << 20, 0);
+ let wide_length = wide.len() as u64;
+ let mut tail = Vec::new();
+ for offset in 1..=SMALL_BATCHES {
+ tail.extend_from_slice(&disk_batch(128 << 10, offset));
+ }
+ let segments = if separate_segments {
+ vec![wide, tail]
+ } else {
+ wide.extend_from_slice(&tail);
+ vec![wide]
+ };
+ let plan = sizing_plan(Some(1), 0);
+ let mut walk = DiskWalk::starting_at(0, 0);
+ for (index, records) in segments.into_iter().enumerate() {
+ let path = directory.path().join(format!("{index}.log"));
+ std::fs::write(&path, &records).unwrap();
+ let file = compio::fs::File::open(path).await.unwrap();
+ walk.position = 0;
+ assert!(matches!(
+ plan.walk_segment(
+ &file,
+ MessageLookup::Offset {
+ offset: 0,
+ count: 2,
+ ceiling: u64::MAX
+ },
+ 2,
+ records.len() as u64,
+ &mut walk,
+ )
+ .await,
+ SegmentWalk::Done
+ ));
+ }
+ assert_eq!(walk.matched, 2);
+ assert_eq!(walk.chunk_reads, 3);
+ assert_eq!(
+ walk.requested_bytes,
+ DISK_POLL_CHUNK_MIN + wide_length + DISK_POLL_CHUNK_MAX,
+ "one oversized batch must not raise subsequent reads above the
chunk ceiling"
+ );
+ }
+ }
+
+ #[cfg(feature = "poll-diagnostics")]
+ fn disk_batch(payload_length: u32, offset: u64) -> Vec<u8> {
+ let mut messages = IggyMessages::with_capacity(1);
+ messages.push(IggyMessage {
+ header: IggyMessageHeader {
+ payload_length,
+ ..Default::default()
+ },
+ payload: Bytes::from(vec![1;
usize::try_from(payload_length).unwrap()]),
+ user_headers: None,
+ });
+ let mut batch =
+ SendMessagesOwned::from_messages(IggyNamespace::new(1, 1, 0),
&messages).unwrap();
+ batch.header.base_offset = offset;
+ batch.header.batch_checksum =
batch.header.checksum_for_blob(&batch.blob);
+ let mut record = vec![0; batch.header.total_size()];
+ batch.header.encode_into(&mut record);
+ record[COMMAND_HEADER_SIZE..].copy_from_slice(&batch.blob);
+ record
+ }
+
fn offset_query(offset: u64) -> MessageLookup {
MessageLookup::Offset {
offset,
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index 76fff1935..7cf5cb62d 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -60,6 +60,7 @@ webpki-roots = { workspace = true }
[dev-dependencies]
mockall = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
[lints]
workspace = true
diff --git a/core/sdk/src/poll_routing.rs b/core/sdk/src/poll_routing.rs
index 612b16d0d..5056e775e 100644
--- a/core/sdk/src/poll_routing.rs
+++ b/core/sdk/src/poll_routing.rs
@@ -50,6 +50,7 @@ const MAX_CACHED_ROUTES: usize = 4096;
const MAX_DATA_CONNECTIONS: usize = 256;
const POLL_TIMEOUT: Duration = Duration::from_secs(30);
const ROUTING_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const ROUTING_RETRY_MAX_INTERVAL: Duration = Duration::from_secs(1);
pub(crate) const ROSTER_READ_TIMEOUT: Duration = Duration::from_secs(5);
pub(crate) const fn is_poll_routing_code(code: u32) -> bool {
@@ -224,6 +225,9 @@ impl<T: PollTransport> PollRouter<T> {
code: u32,
payload: Bytes,
) -> Result<Bytes, IggyError> {
+ if !self.is_clustered(coordinator).await? {
+ return coordinator.send_raw_with_response(code, payload).await;
+ }
let (_, route_size) =
GetConsumerOffsetRequest::decode(&payload).map_err(|_|
IggyError::InvalidCommand)?;
let key = (
@@ -278,6 +282,7 @@ impl<T: PollTransport> PollRouter<T> {
.send_poll_control(PING_CODE, Bytes::new())
.await?;
}
+ let mut retry_interval = ROUTING_RETRY_INTERVAL;
loop {
let result = self.poll_once(coordinator, code, &key,
&payload).await;
if !matches!(result, Err(IggyError::TransientNotAccepted)) {
@@ -287,10 +292,11 @@ impl<T: PollTransport> PollRouter<T> {
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(&key);
- if Instant::now() + ROUTING_RETRY_INTERVAL >= deadline {
+ if Instant::now() + retry_interval >= deadline {
return Err(IggyError::TransientNotAccepted);
}
- sleep(ROUTING_RETRY_INTERVAL).await;
+ sleep(retry_interval).await;
+ retry_interval = (retry_interval *
2).min(ROUTING_RETRY_MAX_INTERVAL);
}
})
.await;
@@ -551,6 +557,7 @@ mod tests {
#[derive(Debug, Default)]
struct Script {
exchanges: Mutex<VecDeque<Exchange>>,
+ route_queries: AtomicUsize,
attachments: Mutex<Vec<AttachConsumerSessionRequest>>,
connections: AtomicUsize,
pause: Mutex<Option<Arc<Pause>>>,
@@ -594,6 +601,12 @@ mod tests {
.pop_front()
.expect("unexpected request");
assert_eq!((channel, code), (expected_channel, expected_code));
+ if matches!(
+ code,
+ GET_POLL_ROUTING_CODE | GET_CONSUMER_OFFSET_ROUTING_CODE
+ ) {
+ self.script.route_queries.fetch_add(1, Ordering::Relaxed);
+ }
if code == ATTACH_CONSUMER_SESSION_CODE {
self.script
.attachments
@@ -1054,6 +1067,53 @@ mod tests {
assert!(coordinator.script.exchanges.lock().unwrap().is_empty());
}
+ #[tokio::test(start_paused = true)]
+ async fn
sustained_refusals_bound_route_queries_and_keep_the_non_admission_result() {
+ const SCRIPTED_ATTEMPTS: usize = 1000;
+ const MAX_ROUTE_QUERIES: usize = 40;
+ for refuse_routing in [true, false] {
+ let exchanges = (0..SCRIPTED_ATTEMPTS).flat_map(|_| {
+ if refuse_routing {
+ return vec![(
+ Channel::Coordinator,
+ GET_POLL_ROUTING_CODE,
+ Err(IggyError::TransientNotAccepted),
+ )];
+ }
+ vec![
+ (Channel::Coordinator, GET_POLL_ROUTING_CODE,
Ok(routing())),
+ (
+ Channel::Data,
+ ATTACH_CONSUMER_SESSION_CODE,
+ Ok(Bytes::new()),
+ ),
+ (
+ Channel::Data,
+ POLL_MESSAGES_ON_PRIMARY_CODE,
+ Err(IggyError::TransientNotAccepted),
+ ),
+ ]
+ });
+ let (router, coordinator, request) = fixture(exchanges);
+ let started = Instant::now();
+ assert!(matches!(
+ router.poll(&coordinator, &request).await,
+ Err(IggyError::TransientNotAccepted)
+ ));
+ let queries =
coordinator.script.route_queries.load(Ordering::Relaxed);
+ assert!(
+ queries <= MAX_ROUTE_QUERIES,
+ "sustained refusal overloaded the coordinator with {queries}
route queries"
+ );
+ assert!(started.elapsed() <= POLL_TIMEOUT);
+ assert!(started.elapsed() >= POLL_TIMEOUT -
Duration::from_secs(1));
+ assert_eq!(
+ coordinator.script.connections.load(Ordering::Relaxed),
+ usize::from(!refuse_routing)
+ );
+ }
+ }
+
#[tokio::test]
async fn complete_error_replies_preserve_the_attached_connection() {
let (router, coordinator, request) = fixture([
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 1ccb61405..1110e7291 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -142,11 +142,7 @@ impl BinaryTransport for QuicClient {
code: u32,
payload: Bytes,
) -> Result<Bytes, IggyError> {
- if self.poll_router.is_clustered(self).await? {
- self.poll_router.write_offset(self, code, payload).await
- } else {
- self.send_raw_with_response(code, payload).await
- }
+ self.poll_router.write_offset(self, code, payload).await
}
async fn send_poll_with_response(
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index da1ee621d..8cf7154bf 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -196,11 +196,7 @@ impl BinaryTransport for TcpClient {
code: u32,
payload: Bytes,
) -> Result<Bytes, IggyError> {
- if self.poll_router.is_clustered(self).await? {
- self.poll_router.write_offset(self, code, payload).await
- } else {
- self.send_raw_with_response(code, payload).await
- }
+ self.poll_router.write_offset(self, code, payload).await
}
async fn send_poll_with_response(
diff --git a/core/sdk/src/websocket/websocket_client.rs
b/core/sdk/src/websocket/websocket_client.rs
index c3b22a1f3..f18a4ad53 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -136,11 +136,7 @@ impl BinaryTransport for WebSocketClient {
code: u32,
payload: Bytes,
) -> Result<Bytes, IggyError> {
- if self.poll_router.is_clustered(self).await? {
- self.poll_router.write_offset(self, code, payload).await
- } else {
- self.send_raw_with_response(code, payload).await
- }
+ self.poll_router.write_offset(self, code, payload).await
}
async fn send_poll_with_response(
diff --git a/core/server/config.toml b/core/server/config.toml
index 0fc6a470f..628757c78 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -1064,10 +1064,14 @@ max_batch = 256
# also breaks Go clients, whose frame cap is a hard 64 MiB constant.
max_message_size = "64 MiB"
-# Bound on the per-peer mpsc queue. The writer task drains; the
+# Bound on each replica peer's mpsc queue. The writer task drains; the
# send_to_* path enqueues.
peer_queue_capacity = 4096
+# Each SDK connection allocates separate inbound and outbound queues.
+# Keep their capacity independent of replica repair bursts.
+client_queue_capacity = 256
+
# Interval between outbound reconnect attempts to peers with peer_id > self_id.
reconnect_period = "5 s"
diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs
index f383030f0..8692bb630 100644
--- a/core/server/src/dispatch/mod.rs
+++ b/core/server/src/dispatch/mod.rs
@@ -225,9 +225,8 @@ where
// manager, so BOTH planes ran as the original registrant;
// - the pair carries far less entropy than "client-generated random
// u128" implies: HTTP mints `client_id` from the shard-0 sequential
-// counter (`mint_shard_zero_client_id`, seeded at 1 per process) and no
-// live path ever bumps an epoch past 1, so the token was `client=N,
-// session=1` for small N;
+// counter (`mint_shard_zero_client_id`) and the epoch is a metadata commit
+// position, so neither value is an authentication secret;
// - `ClientEntry` carries no transport or plane tag, so a raw TCP peer
// could bind an HTTP-originated session;
// - `bind_session` demotes the evicted holder to `Connected`, the one
diff --git a/core/server/src/http/forward.rs b/core/server/src/http/forward.rs
index 758deb8ce..b39074909 100644
--- a/core/server/src/http/forward.rs
+++ b/core/server/src/http/forward.rs
@@ -707,7 +707,7 @@ async fn classify_local_partition_reply(response: Response)
-> AttemptOutcome {
return AttemptOutcome::Relay(bad_gateway());
}
};
- if parts.status == StatusCode::SERVICE_UNAVAILABLE &&
is_transient_not_accepted_body(&body) {
+ if is_transient_not_accepted_body(&body) {
return AttemptOutcome::Retry;
}
AttemptOutcome::Relay(Response::from_parts(parts, Body::from(body)))
diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md
index 30f831f91..33f9bd049 100644
--- a/foreign/csharp/README.md
+++ b/foreign/csharp/README.md
@@ -28,6 +28,13 @@ The examples target server 0.9.0. For source builds, use the
server and SDK from
The SDK targets .NET 8 and .NET 10; repository examples require .NET 10.
`0.9.0-edge.9`
includes the independent message and consumer-offset durability options.
+Cluster auto-commit polling over TCP/TLS keeps group membership on the
coordinator
+and uses separate connections to partition primaries. It requires server
support
+for binary commands 14, 103 and 104. Pause binary auto-commit consumers for the
+whole upgrade: upgrade every server first, then the SDKs, and restart consumers
+so they rejoin their groups. Older SDKs can lose membership when a backup
refuses
+an offset commit; the new SDK does not fall back to legacy polling.
+
### Supported Protocols
The SDK supports two transport protocols:
diff --git a/foreign/go/README.md b/foreign/go/README.md
index 1474bd88c..877434e50 100644
--- a/foreign/go/README.md
+++ b/foreign/go/README.md
@@ -61,7 +61,11 @@ but does not by itself prove that new messages were appended.
In a cluster, auto-commit polls use persistent connections to partition
primaries while the coordinator keeps the consumer's group membership.
-Servers must support primary poll routing and consumer-session attachment.
+Servers must support primary poll routing and consumer-session attachment
+(binary commands 14, 103 and 104). Pause binary auto-commit consumers for the
+whole upgrade: upgrade every server first, then the SDKs, and restart consumers
+so they rejoin their groups. Older SDKs can lose membership when a backup
+refuses an offset commit; the new SDK does not fall back to legacy polling.
Only a poll refused before admission is retried. `ErrTransientNotCommitted`
or cancellation after sending a poll can mean its offset advanced without
a reply; the SDK does not replay that poll automatically.
diff --git a/foreign/go/client/tcp/cluster.go b/foreign/go/client/tcp/cluster.go
index 63de81b46..0f040278c 100644
--- a/foreign/go/client/tcp/cluster.go
+++ b/foreign/go/client/tcp/cluster.go
@@ -19,8 +19,10 @@ package tcp
import (
"context"
+ "fmt"
iggcon "github.com/apache/iggy/foreign/go/contracts"
+ ierror "github.com/apache/iggy/foreign/go/errors"
"github.com/apache/iggy/foreign/go/internal/command"
)
@@ -34,6 +36,9 @@ func (c *IggyTcpClient) GetClusterMetadata(ctx
context.Context) (*iggcon.Cluster
if err != nil {
return nil, err
}
+ if len(metadata.Nodes) == 0 {
+ return nil, fmt.Errorf("%w: empty cluster roster",
ierror.ErrTransientNotAccepted)
+ }
c.clustered.Store(len(metadata.Nodes) > 1)
c.topologyKnown.Store(true)
return &metadata, nil
diff --git a/foreign/go/client/tcp/tcp_poll_routing_test.go
b/foreign/go/client/tcp/tcp_poll_routing_test.go
index 6e7fe7554..49c0f6fcb 100644
--- a/foreign/go/client/tcp/tcp_poll_routing_test.go
+++ b/foreign/go/client/tcp/tcp_poll_routing_test.go
@@ -540,34 +540,45 @@ func
TestPrimaryPoll_CoordinatorTrafficDoesNotStarveColdRouting(t *testing.T) {
}
func TestPrimaryPoll_UnknownTopologyCannotSelectLegacyPolling(t *testing.T) {
- var failRoster atomic.Bool
- failRoster.Store(true)
- fixture := newPrimaryPollFixture(t, nil, func(_ int, read request)
([]byte, bool) {
- if read.code() == uint32(command.GetClusterMetadataCode) &&
failRoster.Load() {
- return statusReplyFrame(vsr.OperationNonReplicated,
uint32(ierror.ErrFeatureUnavailable.Code()), nil), true
- }
- return nil, false
- })
- require.False(t, fixture.client.topologyKnown.Load(), "the login-time
roster read failed")
- parent := fixture.client.session.ClientID()
- _, err := pollPrimaryPartition(context.Background(), fixture.client, 0)
- require.ErrorIs(t, err, ierror.ErrFeatureUnavailable)
- assert.Zero(t, requestCount(fixture.coordinator.recorded(),
command.PollMessagesCode))
- assert.Zero(t, requestCount(fixture.coordinator.recorded(),
command.GetPollRoutingCode))
+ for _, test := range []struct {
+ name string
+ reply []byte
+ wantErr error
+ }{
+ {"unsupported", statusReplyFrame(vsr.OperationNonReplicated,
uint32(ierror.ErrFeatureUnavailable.Code()), nil),
ierror.ErrFeatureUnavailable},
+ {"empty", clusterMetadataFrame(t, 0),
ierror.ErrTransientNotAccepted},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ var failRoster atomic.Bool
+ failRoster.Store(true)
+ fixture := newPrimaryPollFixture(t, nil, func(_ int,
read request) ([]byte, bool) {
+ if read.code() ==
uint32(command.GetClusterMetadataCode) && failRoster.Load() {
+ return append([]byte(nil),
test.reply...), true
+ }
+ return nil, false
+ })
+ require.False(t, fixture.client.topologyKnown.Load(),
"the login-time roster read failed")
+ parent := fixture.client.session.ClientID()
+ _, err := pollPrimaryPartition(context.Background(),
fixture.client, 0)
+ require.ErrorIs(t, err, test.wantErr)
+ assert.Zero(t,
requestCount(fixture.coordinator.recorded(), command.PollMessagesCode))
+ assert.Zero(t,
requestCount(fixture.coordinator.recorded(), command.GetPollRoutingCode))
- failRoster.Store(false)
- _, err = pollPrimaryPartition(context.Background(), fixture.client, 0)
- require.NoError(t, err)
- require.True(t, fixture.client.topologyKnown.Load())
- failRoster.Store(true)
- _, err = fixture.client.GetClusterMetadata(context.Background())
- require.ErrorIs(t, err, ierror.ErrFeatureUnavailable)
- _, err = pollPrimaryPartition(context.Background(), fixture.client, 0)
- require.NoError(t, err, "a failed refresh must preserve known topology
and warm routes")
- assert.Equal(t, 1, fixture.primaries[0].connections())
- assert.Equal(t, 1, fixture.coordinator.connections())
- assert.Equal(t, parent, fixture.client.session.ClientID())
- assert.Zero(t, requestCount(fixture.coordinator.recorded(),
command.PollMessagesCode))
+ failRoster.Store(false)
+ _, err = pollPrimaryPartition(context.Background(),
fixture.client, 0)
+ require.NoError(t, err)
+ require.True(t, fixture.client.topologyKnown.Load())
+ failRoster.Store(true)
+ _, err =
fixture.client.GetClusterMetadata(context.Background())
+ require.ErrorIs(t, err, test.wantErr)
+ _, err = pollPrimaryPartition(context.Background(),
fixture.client, 0)
+ require.NoError(t, err, "a failed refresh must preserve
known topology and warm routes")
+ assert.Equal(t, 1, fixture.primaries[0].connections())
+ assert.Equal(t, 1, fixture.coordinator.connections())
+ assert.Equal(t, parent,
fixture.client.session.ClientID())
+ assert.Zero(t,
requestCount(fixture.coordinator.recorded(), command.PollMessagesCode))
+ })
+ }
}
func TestPrimaryPoll_UnknownTopologyCanRecoverAsStandalone(t *testing.T) {
diff --git a/foreign/java/README.md b/foreign/java/README.md
index f6fcb4f9a..f64f75615 100644
--- a/foreign/java/README.md
+++ b/foreign/java/README.md
@@ -79,6 +79,14 @@ dependencies {
## Quick Start
+Cluster auto-commit polling over TCP/TLS keeps group membership on the
+coordinator and uses separate connections to partition primaries. It requires
+server support for binary commands 14, 103 and 104. Pause binary auto-commit
+consumers for the whole upgrade: upgrade every server first, then the SDKs,
+and restart consumers so they rejoin their groups. Older SDKs can lose
membership
+when a backup refuses an offset commit; the new SDK does not fall back to
legacy
+polling. Use `Iggy.tcpClientBuilder()` to get this routing and session
management.
+
Start the server with the [example
prerequisites](../../examples/java/#running-examples) and matching credentials.
The following snippets show alternative clients. Close a blocking client with
`close()` or an async client with `close().join()` when finished.
### TCP Client (Blocking)
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
index 0f63cb575..6c87f0a8d 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/MessagesTcpClient.java
@@ -88,6 +88,11 @@ public class MessagesTcpClient implements MessagesClient {
private final PollRouter pollRouter;
private final Supplier<CompletableFuture<Boolean>> clustered;
+ /**
+ * Creates a low-level client on the supplied connection without primary
routing.
+ * Use {@code Iggy.tcpClientBuilder()} for clustered auto-commit polling
so the
+ * coordinator retains group membership while data connections reach
primaries.
+ */
public MessagesTcpClient(Supplier<AsyncTcpConnection> connectionSupplier) {
this(connectionSupplier, new ClientRoutingState());
}
diff --git a/foreign/node/README.md b/foreign/node/README.md
index 19b4d3793..58461d3ad 100644
--- a/foreign/node/README.md
+++ b/foreign/node/README.md
@@ -146,6 +146,13 @@ Durations accept the same expressions as the Rust SDK, for
example `500ms`,
## use sources
+Cluster auto-commit polling over TCP/TLS keeps group membership on the
coordinator
+and uses separate connections to partition primaries. It requires server
support
+for binary commands 14, 103 and 104. Pause binary auto-commit consumers for the
+whole upgrade: upgrade every server first, then the SDKs, and restart consumers
+so they rejoin their groups. Older SDKs can lose membership when a backup
refuses
+an offset commit; the new SDK does not fall back to legacy polling.
+
### Install
```bash
