hubcio commented on code in PR #4169:
URL: https://github.com/apache/iggy/pull/4169#discussion_r4009082110
##########
core/partitions/src/persistence.rs:
##########
@@ -1140,49 +1155,69 @@ impl<S: DurableStorage> PartitionPersistence<S> {
break;
}
if epoch == self.epoch.get() && !self.retired.get() {
- let mut references = self.segment_references.borrow_mut();
- if rebuild_references {
- references.clear();
- references.extend(journal.written_segment_references(0));
- } else if let Some(from_op) =
self.written_head.get().checked_add(1) {
-
references.extend(journal.written_segment_references(from_op));
- }
- drop(references);
- self.disk_bytes.set(journal.size_bytes());
- self.retained_bytes.set(journal.retained_bytes());
- self.segment_checkpoint.set(journal.segment_checkpoint());
- let advanced = journal.durable_op() != self.durable_head.get()
- || journal.checkpoint_op() != self.checkpoint.get()
- || journal.certified_log_view() !=
self.certified_log_view.get()
- || (journal.segment_checkpoint().is_some()
- && journal.head() != self.written_head.get());
- self.certified_log_view.set(journal.certified_log_view());
- if self
- .requested_log_view
- .get()
- .is_some_and(|(view, _, _)| Some(view) ==
self.certified_log_view.get())
- {
- self.requested_log_view.set(None);
- }
- self.written_head.set(journal.head());
- self.durable_head.set(journal.durable_op());
- if journal.checkpoint_op() > self.checkpoint.get() {
- self.accepted
- .borrow_mut()
- .checkpoint(journal.checkpoint_op());
- }
- self.checkpoint.set(journal.checkpoint_op());
- self.checkpoint_checksum.set(journal.checkpoint_checksum());
- self.purge_generation.set(journal.purge_marker().0);
- self.purge_floor.set(journal.purge_marker().1);
- if advanced {
- self.notify();
- }
+ self.publish_mutation(journal, rebuild_references);
+ }
+ // Reclaim after publishing the mutation when the queue is empty.
+ // Appends arriving during reclaim still wait for its unlinks and
+ // directory barrier.
+ // A partition that never drains would then never reclaim, so force
+ // a pass every `RECLAIM_MUTATIONS_MAX` mutations and accept that
+ // one group's latency.
+ mutations_since_reclaim += 1;
+ let idle = self.queue.borrow().is_empty();
+ if idle || mutations_since_reclaim >= RECLAIM_MUTATIONS_MAX {
Review Comment:
The 16-path limit was removed in a64a966c5. `cleanup_obsolete` snapshots
`self.obsolete.len()` and attempts every path from that pass. The 64-mutation
constant only schedules cleanup; it does not limit how many paths are
removed.
Failed unlinks are queued for the next pass. The checkpoint-generation
regression verifies that one cleanup pass drains the backlog and reopening
preserves the published history.
##########
core/consensus/src/client_table.rs:
##########
@@ -1653,6 +1674,27 @@ impl ClientTable {
self.slots[slot_idx].as_ref().map(|entry| entry.epoch)
}
+ /// 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
Review Comment:
Agreed that the request documentation was ambiguous. It now states the actual
contract: the authenticated user must match the parent session's registered
user, the epoch must match, and attaching does not extend the parent's
lifetime.
Connections using the same account can share its group membership. This is
consistent with login/register, which already allows that user to rebind its
client id (`core/metadata/src/impls/metadata.rs`,
`submit_register_in_process`).
This documents the existing user-level boundary; it does not introduce
isolation between processes sharing credentials. Providing that isolation
needs a session capability and consistent changes to both attachment and
login/resume, so I am keeping that redesign separate.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+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 {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
Review Comment:
Keeping the known roster across a session reset because it describes cluster
topology, not group membership. `clear_session` advances the generation and
clears routes and data connections; in-flight replies and attachments are
fenced against that generation. Topology discovery updates the cached roster
when it obtains a valid response. Clearing it on every session invalidation
adds discovery I/O without strengthening the membership fence. The reconnect
regressions cover late route replies and pending data-connection users.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+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 {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
Review Comment:
This is asserted in a64a966c5 by
`offset_writes_share_the_routing_request_prefix` in
`core/binary_protocol/src/requests/consumer_offsets/mod.rs`. It independently
encodes store and delete requests, decodes their prefix as the routing
request,
and checks both bytes and consumed length. The combinations cover both
consumer kinds, numeric/named identifiers, absent/zero/maximum partition ids,
and both acknowledgement levels. No additional wire change is needed.
##########
core/server/config.toml:
##########
@@ -993,12 +997,16 @@ offset_reservation_lease = 65536
# served from the ring before falling back to bulk sync, at the cost of pinned
# memory per partition. Must be > 0 and <= 65536. Single-replica partitions
# retain nothing regardless.
-evicted_ring_capacity = 4096
+evicted_ring_capacity = 65536
# Byte ceiling for the evicted ring per partition; whichever ring cap (this or
# evicted_ring_capacity) trips first evicts. Bounds the ring memory a burst of
# large batches can pin. Must be > 0 and <= "256 MiB".
-evicted_ring_bytes_max = "16 MiB"
+# PER PARTITION and per replica, so the node-wide ceiling is this times the
partitions it
+# hosts, pinned as anonymous memory beside the same bytes in page cache.
Raising it on a
+# node with many partitions is a large memory decision;
partition_repair_ring_bytes on the
+# metrics endpoint reports what the rings on each shard actually hold.
+evicted_ring_bytes_max = "64 MiB"
Review Comment:
Correct: the default remains 64 MiB per partition per replica, and there is
no
node-wide budget. It is also a payload-byte budget, not an allocation or RSS
ceiling: spare buffer capacity, alignment and map overhead are additional,
and the newest entry is retained even if it exceeds the byte budget alone.
I corrected the config comments, occupancy documentation and metric help to
make those limits explicit. The configured defaults are intentional and
remain
unchanged. Allocation-aware accounting and a node-wide eviction policy need
separate work; the BTreeMap change addresses lookup cost, not those concerns.
##########
core/server/config.toml:
##########
@@ -726,7 +726,8 @@ pin_cores = true
# ~ the prepare queue depth of the planes the shard hosts ([metadata] on
# shard 0, [partition] elsewhere) times replica_count times directions.
# Both depths are tunable, so raising either raises the capacity needed here.
-inbox_capacity = 1024
+# Slots are allocated at startup per shard, even before traffic arrives.
+inbox_capacity = 65536
Review Comment:
The 65,536-slot default is intentional. I removed the misleading per-plane
formula: sizing must account for the combined traffic of all consensus groups
hosted by the shard, replica fan-out and burst headroom. The comment still
states that slots are preallocated per shard. A single group's prepare depth
does not derive this default.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+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 {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
+
+ pub(crate) async fn write_offset(
+ &self,
+ coordinator: &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 = (
+ GET_CONSUMER_OFFSET_ROUTING_CODE,
+ payload.slice(..route_size),
+ );
+ self.send_routed(coordinator, code, key, payload).await
+ }
+
+ pub(crate) async fn is_clustered(&self, coordinator: &T) -> Result<bool,
IggyError> {
+ if self.roster_size.load(Ordering::Acquire) == 0 {
+ let response = timeout(
+ ROSTER_READ_TIMEOUT,
+ coordinator.send_poll_control(GET_CLUSTER_METADATA_CODE,
Bytes::new()),
+ )
+ .await
+ .map_err(|_| IggyError::TransientNotAccepted)??;
+ let metadata = ClusterMetadataResponse::decode_from(&response)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if metadata.nodes.is_empty() {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ self.roster_size
+ .store(metadata.nodes.len(), Ordering::Release);
+ }
+ Ok(self.roster_size.load(Ordering::Acquire) > 1)
+ }
+
+ async fn send_routed(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: RouteKey,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let now = Instant::now();
+ let deadline = now + POLL_TIMEOUT;
+ let result = timeout_at(deadline, async {
+ let heartbeat_due = {
+ let mut next = self
+ .next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let due = next.is_some_and(|next| now >= next);
+ if next.is_none() || due {
+ *next = Some(now +
coordinator.get_heartbeat_interval().get_duration());
+ }
+ due
+ };
+ if heartbeat_due {
+ coordinator
+ .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)) {
+ return result;
+ }
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ if Instant::now() + retry_interval >= deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ sleep(retry_interval).await;
+ retry_interval = (retry_interval *
2).min(ROUTING_RETRY_MAX_INTERVAL);
Review Comment:
Keeping route invalidation here. Code 58 guarantees non-admission but does
not
identify the cause: leadership, membership/metadata fencing and capacity can
produce the same response. Reusing the route on an assumed capacity refusal
could keep hitting an obsolete primary or attachment for the entire retry
budget. Backoff bounds discovery traffic while preserving recovery. Avoiding
discovery safely needs a distinct refusal reason or a trustworthy route hint;
that protocol extension is outside this change.
##########
core/configs/src/server_config/partition.rs:
##########
@@ -115,7 +115,7 @@ fn default_offset_reservation_lease() -> NonZeroU32 {
}
/// Mirrors `partitions::EVICTED_RING_CAPACITY`.
-pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 4096;
+pub const DEFAULT_EVICTED_RING_CAPACITY: usize = 65536;
Review Comment:
Keeping both at 65,536. The default is an intentional tuning choice; the
validation maximum remains a guard against still larger allocations. Equality
does not disable either the entry cap or byte cap. Increasing the maximum
just
to leave headroom would allow a larger per-partition footprint without a
requirement for it; operators can still configure a lower capacity.
##########
core/sdk/src/poll_routing.rs:
##########
@@ -0,0 +1,1284 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Auto-commit polls and offset writes use persistent data connections while
+//! the coordinator retains group membership. Only explicit non-admission
permits
+//! rerouting. Replicated writes keep the data session's own deduplication
identity.
+//! Routes and attachments are fenced by the coordinator's session generation.
+//! Cluster routing requires servers supporting the routing and attachment
commands.
+
+use crate::leader_aware::{node_address, transport_port};
+use async_trait::async_trait;
+use bytes::Bytes;
+use iggy_binary_protocol::codes::{
+ ATTACH_CONSUMER_SESSION_CODE, GET_CLUSTER_METADATA_CODE,
GET_CONSUMER_OFFSET_ROUTING_CODE,
+ GET_POLL_ROUTING_CODE, PING_CODE, POLL_MESSAGES_CODE,
POLL_MESSAGES_ON_PRIMARY_CODE,
+};
+use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
+use iggy_binary_protocol::requests::system::AttachConsumerSessionRequest;
+use iggy_binary_protocol::responses::messages::PollRoutingResponse;
+use
iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetadataResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode};
+use iggy_common::{
+ BinaryClient, ClusterNode, Credentials, IdKind, Identifier, IggyError,
TransportProtocol,
+};
+use secrecy::SecretString;
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::Mutex as AsyncMutex;
+use tokio::time::{Instant, sleep, timeout, timeout_at};
+use tracing::error;
+
+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 {
+ matches!(
+ code,
+ ATTACH_CONSUMER_SESSION_CODE
+ | GET_POLL_ROUTING_CODE
+ | POLL_MESSAGES_ON_PRIMARY_CODE
+ | GET_CONSUMER_OFFSET_ROUTING_CODE
+ )
+}
+
+#[async_trait]
+pub(crate) trait PollTransport: BinaryClient + Send + Sync + Sized {
+ const PROTOCOL: TransportProtocol;
+
+ async fn connect_poll_client(&self, endpoint: &str) -> Result<Self,
IggyError>;
+
+ /// One exchange on this connection, with no node movement or automatic
+ /// replay of an ambiguous outcome, including replicated offset writes.
+ async fn send_poll_request(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError>;
+
+ async fn send_poll_control(&self, code: u32, payload: Bytes) ->
Result<Bytes, IggyError> {
+ let result = self.send_poll_request(code, payload.clone()).await;
+ if result.as_ref().is_err_and(poll_connection_failed) {
+ self.send_raw_with_response(PING_CODE, Bytes::new()).await?;
+ return self.send_poll_request(code, payload).await;
+ }
+ result
+ }
+}
+
+#[derive(Debug)]
+struct PollRoute {
+ generation: u64,
+ endpoint: String,
+ consumer_session: AttachConsumerSessionRequest,
+}
+
+#[derive(Debug)]
+struct PollConnection<T> {
+ client: T,
+ consumer_session: Option<AttachConsumerSessionRequest>,
+ usable: bool,
+}
+
+type ConnectionSlot<T> = Arc<AsyncMutex<Option<PollConnection<T>>>>;
+type RouteKey = (u32, Bytes);
+
+#[derive(Debug)]
+pub(crate) struct PollRouter<T> {
+ pub(crate) metadata_watermark: Arc<AtomicU64>,
+ /// Zero means no successful topology read, not a standalone server.
+ pub(crate) roster_size: AtomicUsize,
+ session_generation: AtomicU64,
+ routes: Mutex<HashMap<RouteKey, Arc<PollRoute>>>,
+ connections: Mutex<HashMap<String, ConnectionSlot<T>>>,
+ credentials: Mutex<Option<(Credentials, u32)>>,
+ next_heartbeat: Mutex<Option<Instant>>,
+}
+
+impl<T> Default for PollRouter<T> {
+ fn default() -> Self {
+ Self {
+ metadata_watermark: Arc::default(),
+ roster_size: AtomicUsize::new(0),
+ session_generation: AtomicU64::new(0),
+ routes: Mutex::default(),
+ connections: Mutex::default(),
+ credentials: Mutex::default(),
+ next_heartbeat: Mutex::default(),
+ }
+ }
+}
+
+impl<T> PollRouter<T> {
+ pub(crate) fn clear_session(&self) {
+ self.next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ let mut routes = self
+ .routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ // Serialize invalidation with route and connection publication.
+ self.session_generation.fetch_add(1, Ordering::AcqRel);
+ routes.clear();
+ connections.clear();
+ }
+
+ pub(crate) fn remember_credentials(&self, credentials: Credentials,
user_id: u32) {
+ *self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) =
Some((credentials, user_id));
+ }
+
+ pub(crate) fn forget_credentials(&self) {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .take();
+ }
+
+ pub(crate) fn credentials(&self) -> Option<Credentials> {
+ self.credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .as_ref()
+ .map(|(credentials, _)| credentials.clone())
+ }
+
+ pub(crate) fn refresh_password(&self, user: &Identifier, new_password:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, password), user_id))
=
+ credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *password = SecretString::from(new_password.to_owned());
+ }
+ }
+
+ pub(crate) fn refresh_username(&self, user: &Identifier, new_username:
&str) {
+ let mut credentials = self
+ .credentials
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let Some((Credentials::UsernamePassword(username, _), user_id)) =
credentials.as_mut()
+ else {
+ return;
+ };
+ if matches_session_user(user, *user_id, username) {
+ *username = new_username.to_owned();
+ }
+ }
+}
+
+impl<T: PollTransport> PollRouter<T> {
+ pub(crate) async fn poll(
+ &self,
+ coordinator: &T,
+ request: &PollMessagesRequest,
+ ) -> Result<Bytes, IggyError> {
+ if !request.auto_commit || !self.is_clustered(coordinator).await? {
+ return coordinator
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await;
+ }
+ let payload = request.to_bytes();
+ let parameters_size = request.strategy.encoded_size() +
size_of::<u32>() + size_of::<u8>();
+ let key = (
+ GET_POLL_ROUTING_CODE,
+ payload.slice(..payload.len() - parameters_size),
+ );
+ self.send_routed(coordinator, POLL_MESSAGES_ON_PRIMARY_CODE, key,
payload)
+ .await
+ }
+
+ pub(crate) async fn write_offset(
+ &self,
+ coordinator: &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 = (
+ GET_CONSUMER_OFFSET_ROUTING_CODE,
+ payload.slice(..route_size),
+ );
+ self.send_routed(coordinator, code, key, payload).await
+ }
+
+ pub(crate) async fn is_clustered(&self, coordinator: &T) -> Result<bool,
IggyError> {
+ if self.roster_size.load(Ordering::Acquire) == 0 {
+ let response = timeout(
+ ROSTER_READ_TIMEOUT,
+ coordinator.send_poll_control(GET_CLUSTER_METADATA_CODE,
Bytes::new()),
+ )
+ .await
+ .map_err(|_| IggyError::TransientNotAccepted)??;
+ let metadata = ClusterMetadataResponse::decode_from(&response)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ if metadata.nodes.is_empty() {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ self.roster_size
+ .store(metadata.nodes.len(), Ordering::Release);
+ }
+ Ok(self.roster_size.load(Ordering::Acquire) > 1)
+ }
+
+ async fn send_routed(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: RouteKey,
+ payload: Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let now = Instant::now();
+ let deadline = now + POLL_TIMEOUT;
+ let result = timeout_at(deadline, async {
+ let heartbeat_due = {
+ let mut next = self
+ .next_heartbeat
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let due = next.is_some_and(|next| now >= next);
+ if next.is_none() || due {
+ *next = Some(now +
coordinator.get_heartbeat_interval().get_duration());
+ }
+ due
+ };
+ if heartbeat_due {
+ coordinator
+ .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)) {
+ return result;
+ }
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ if Instant::now() + retry_interval >= deadline {
+ return Err(IggyError::TransientNotAccepted);
+ }
+ sleep(retry_interval).await;
+ retry_interval = (retry_interval *
2).min(ROUTING_RETRY_MAX_INTERVAL);
+ }
+ })
+ .await;
+ match result {
+ Ok(result) => result,
+ Err(_) => {
+ self.routes
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(&key);
+ Err(IggyError::TransientNotCommitted)
+ }
+ }
+ }
+
+ async fn poll_once(
+ &self,
+ coordinator: &T,
+ code: u32,
+ key: &RouteKey,
+ payload: &Bytes,
+ ) -> Result<Bytes, IggyError> {
+ let route = self.route(coordinator, key, payload).await?;
+ let slot = {
+ let mut connections = self
+ .connections
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ self.validate_route(&route)?;
+ if let Some(slot) = connections.get(&route.endpoint) {
+ Arc::clone(slot)
+ } else {
+ if connections.len() >= MAX_DATA_CONNECTIONS {
+ error!(endpoint = route.endpoint, protocol = %T::PROTOCOL,
limit = MAX_DATA_CONNECTIONS, "primary poll connection pool is full");
+ return Err(IggyError::InvalidConfiguration);
+ }
+ let slot = Arc::default();
+ connections.insert(route.endpoint.clone(), Arc::clone(&slot));
+ slot
+ }
+ };
+ let mut connection = slot.lock().await;
Review Comment:
Correct: one router has one serialized data connection per endpoint. The lock
protects attachment plus the following request, and cancellation marks an
unfinished exchange unusable before another caller can reuse it. Simply
releasing the lock would break that ownership guarantee; the underlying
transports also serialize exchanges. A bounded pool of independent sessions
could increase concurrency, but needs measured workload requirements and its
own attachment/cancellation lifecycle. Keeping that pool design separate from
this membership fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
