This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 40d1e356e fix(server-ng): legacy-parity consumer group errors, roster
HTTP bind (#3754)
40d1e356e is described below
commit 40d1e356eeba768a3e2ce30fd65f37bd358531fd
Author: Hubert Gruszecki <[email protected]>
AuthorDate: Mon Jul 27 20:47:10 2026 +0200
fix(server-ng): legacy-parity consumer group errors, roster HTTP bind
(#3754)
Consumer group join/leave/delete on the ng metadata plane
committed silent OK no-ops (or collapsed every miss into one
code) when the stream, topic, group or membership was missing,
so clients saw success where the legacy server reports an
error. All three ops now resolve level by level and commit the
legacy codes (1009/2010/5000, and 5006 for leaving as an absent
member), pinned end to end by an integration test across TCP,
QUIC and WebSocket; HTTP cannot run the join/leave ladder since
its stateless sessions carry no member identity. Consumer group
ids become 0-based to match the legacy server, asserted on the
wire-visible create response id; server-ng has no stable-format
users yet.
The HTTP listener bound from the top-level [http].address even
in cluster mode, so one-host clusters collided on the shared
port. Cluster listeners now take their port from the node's
roster entry while keeping the bind interface from each
transport's own address, so a loopback-only or 0.0.0.0 host
survives cluster mode and the roster ip stays purely the
advertised address. Roster ports are mandatory for every
enabled transport, validated at startup instead of falling
back to a shared default port that two same-host nodes would
race for, letting every node run from byte-identical per-node
configs.
---
core/common/src/types/identifier/mod.rs | 16 +-
core/configs/src/server_ng_config/cluster.rs | 15 +-
core/configs/src/server_ng_config/validators.rs | 80 +++---
.../tests/sdk/consumer_group_membership.rs | 102 ++++++++
core/metadata/src/stm/consumer_group.rs | 281 ++++++++++++++++++---
core/metadata/src/stm/result.rs | 71 +++++-
core/metadata/src/stm/snapshot.rs | 8 +-
core/metadata/src/stm/stream.rs | 34 ++-
core/sdk/src/clients/consumer.rs | 7 +-
core/server-ng/config.toml | 11 +-
core/server-ng/src/bootstrap.rs | 203 +++++++++++----
core/server-ng/src/partition_reconciler.rs | 24 +-
core/server-ng/src/server_error.rs | 10 +-
.../src/workload/ops/delete_consumer_group.rs | 34 ++-
14 files changed, 734 insertions(+), 162 deletions(-)
diff --git a/core/common/src/types/identifier/mod.rs
b/core/common/src/types/identifier/mod.rs
index 1649cf2d0..28439a779 100644
--- a/core/common/src/types/identifier/mod.rs
+++ b/core/common/src/types/identifier/mod.rs
@@ -153,13 +153,10 @@ impl Identifier {
}
/// Creates a new identifier from the given numeric value.
+ ///
+ /// Zero is a valid id: consumer groups are 0-based, and server-side slab
+ /// keys start at 0 (`Identifier::numeric(slab_key)` must never fail).
pub fn numeric(value: u32) -> Result<Self, IggyError> {
- /*
- if value == 0 {
- return Err(IggyError::InvalidIdentifier);
- }
- */
-
Ok(Self {
kind: IdKind::Numeric,
length: 4,
@@ -320,6 +317,13 @@ mod tests {
assert!(Identifier::numeric(1).is_ok());
}
+ #[test]
+ fn identifier_with_a_value_of_zero_should_be_valid() {
+ // 0-based consumer group ids and slab keys go over the wire as
+ // numeric identifiers; rejecting zero would break both.
+ assert!(Identifier::numeric(0).is_ok());
+ }
+
#[test]
fn identifier_with_a_value_of_non_empty_string_should_be_valid() {
assert!(Identifier::named("test").is_ok());
diff --git a/core/configs/src/server_ng_config/cluster.rs
b/core/configs/src/server_ng_config/cluster.rs
index 44b565f58..81e4639d1 100644
--- a/core/configs/src/server_ng_config/cluster.rs
+++ b/core/configs/src/server_ng_config/cluster.rs
@@ -155,6 +155,13 @@ pub struct ClusterNodeConfig {
pub ports: TransportPorts,
}
+/// Per-node listener ports advertised in the cluster roster. In cluster mode
+/// the roster is the single source of ports: every enabled transport needs
+/// an explicit per-node port (validated at startup, no fallback to the
+/// transport's top-level `address` port). The roster entry's `ip` is the
+/// advertised address only: ws/quic/http bind the interface from their own
+/// `address` config, and followers forward HTTP requests to the primary at
+/// `ip:http`.
#[derive(Debug, Deserialize, Serialize, Clone, Default, ConfigEnv)]
pub struct TransportPorts {
pub tcp: Option<u16>,
@@ -171,11 +178,9 @@ pub struct TransportPorts {
/// follower-to-primary HTTP forwarding depends on. Callers gate `http.enabled`
/// themselves; this covers only the key material.
///
-/// Single source for both the boot-time config validator and the server-ng
-/// runtime forwarding gate. If the two ever disagree the validator's roster
-/// http-port guarantee is silently bypassed: forwarding would activate against
-/// a node the validator never required to expose an http port, and every
-/// forward through it fails closed with a 503.
+/// Forwarding targets resolve from the roster (`ip:ports.http`); the config
+/// validator unconditionally requires a roster port for every enabled
+/// transport, so a forward never dials a node without a declared http port.
pub fn http_forwarding_key_material(jwt: &HttpJwtConfig, cluster:
&ClusterConfig) -> bool {
cluster.enabled
&& ((cluster.auth.enabled && !cluster.auth.shared_secret.is_empty())
diff --git a/core/configs/src/server_ng_config/validators.rs
b/core/configs/src/server_ng_config/validators.rs
index 4736d9055..8c00ea10e 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -26,7 +26,6 @@
//! net.
use super::COMPONENT_NG;
-use super::cluster::http_forwarding_key_material;
use super::server_ng::{ExtraConfig, NamespaceConfig, ServerNgConfig};
use crate::ConfigurationError;
use err_trail::ErrContext;
@@ -138,21 +137,29 @@ impl Validatable<ConfigurationError> for ServerNgConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
- // Without key material forwarding is disabled (followers answer a
- // transient 503) and the server still boots, so it is not required
- // here. When it IS present the operator opted into forwarding, and the
- // roster must support it: a node listed without an http port would
- // silently degrade every forward through it to a fail-closed 503.
- let http_forwarding_active =
- self.http.enabled && http_forwarding_key_material(&self.http.jwt,
&self.cluster);
- if http_forwarding_active {
+ // Cluster mode has no port fallbacks: the roster is the single source
+ // of listener ports, so every enabled transport needs an explicit
+ // per-node port. Falling back to the port of a transport's top-level
+ // `address` would hand two same-host nodes the same socket and fail
+ // only at bind time, and a portless node would silently degrade every
+ // follower-to-primary HTTP forward through it to a fail-closed 503.
+ if self.cluster.enabled {
for node in &self.cluster.nodes {
- if node.ports.http.is_none() {
- eprintln!(
- "cluster node '{}' has no ports.http; every node needs
one when http.enabled so followers can forward to the primary",
- node.name
- );
- return Err(ConfigurationError::InvalidConfigurationValue);
+ let required_ports = [
+ ("tcp", true, node.ports.tcp),
+ ("quic", self.quic.enabled, node.ports.quic),
+ ("http", self.http.enabled, node.ports.http),
+ ("websocket", self.websocket.enabled,
node.ports.websocket),
+ ("tcp_replica", true, node.ports.tcp_replica),
+ ];
+ for (transport, enabled, port) in required_ports {
+ if enabled && port.is_none() {
+ eprintln!(
+ "cluster node '{}' has no ports.{transport};
cluster mode requires an explicit roster port for every enabled transport",
+ node.name
+ );
+ return
Err(ConfigurationError::InvalidConfigurationValue);
+ }
}
}
}
@@ -438,9 +445,9 @@ mod tests {
replica_id,
ports: TransportPorts {
tcp: Some(8090 + u16::from(replica_id)),
- quic: None,
+ quic: Some(8080 + u16::from(replica_id)),
http,
- websocket: None,
+ websocket: Some(8070 + u16::from(replica_id)),
tcp_replica: Some(9090 + u16::from(replica_id)),
},
}
@@ -466,11 +473,35 @@ mod tests {
assert!(cfg.validate().is_ok());
}
- // Without key material forwarding is off, so the roster http-port
- // requirement does not apply either.
+ // Cluster mode has no port fallbacks, so a portless roster node is
+ // invalid even when forwarding is off (keyless).
#[test]
- fn validate_accepts_keyless_cluster_http_with_portless_roster_node() {
+ fn validate_rejects_keyless_cluster_http_with_portless_roster_node() {
let cfg = clustered_http_config(vec![cluster_node(0, Some(3000)),
cluster_node(1, None)]);
+ assert!(cfg.validate().is_err());
+ }
+
+ // The explicit-port rule covers every enabled transport, not just http.
+ #[test]
+ fn validate_rejects_cluster_node_without_port_for_enabled_quic() {
+ let mut cfg = clustered_http_config(vec![
+ cluster_node(0, Some(3000)),
+ cluster_node(1, Some(3001)),
+ ]);
+ cfg.quic.enabled = true;
+ cfg.cluster.nodes[1].ports.quic = None;
+ assert!(cfg.validate().is_err());
+ }
+
+ // A disabled transport never binds, so its roster port may stay unset.
+ #[test]
+ fn validate_accepts_cluster_node_without_port_for_disabled_quic() {
+ let mut cfg = clustered_http_config(vec![
+ cluster_node(0, Some(3000)),
+ cluster_node(1, Some(3001)),
+ ]);
+ cfg.quic.enabled = false;
+ cfg.cluster.nodes[1].ports.quic = None;
assert!(cfg.validate().is_ok());
}
@@ -495,13 +526,4 @@ mod tests {
cfg.cluster.auth.shared_secret =
"0123456789abcdef0123456789abcdef".to_string();
assert!(cfg.validate().is_ok());
}
-
- #[test]
- fn validate_rejects_cluster_http_when_a_roster_node_has_no_http_port() {
- let mut cfg =
- clustered_http_config(vec![cluster_node(0, Some(3000)),
cluster_node(1, None)]);
- cfg.cluster.auth.enabled = true;
- cfg.cluster.auth.shared_secret =
"0123456789abcdef0123456789abcdef".to_string();
- assert!(cfg.validate().is_err());
- }
}
diff --git a/core/integration/tests/sdk/consumer_group_membership.rs
b/core/integration/tests/sdk/consumer_group_membership.rs
index 76d413a48..91800cf76 100644
--- a/core/integration/tests/sdk/consumer_group_membership.rs
+++ b/core/integration/tests/sdk/consumer_group_membership.rs
@@ -32,6 +32,14 @@ const PARK_TIMEOUT: Duration = Duration::from_secs(2);
// Generous bound: on regression the poll hangs until this elapses.
const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
+// Legacy wire error codes for the join/leave failure ladder. Pinned as
literals
+// (not derived from `IggyError`) so renumbering the wire contract fails here
+// loudly -- the same guarantee the non-Rust SDKs depend on.
+const STREAM_ID_NOT_FOUND: u32 = 1009;
+const TOPIC_ID_NOT_FOUND: u32 = 2010;
+const CONSUMER_GROUP_ID_NOT_FOUND: u32 = 5000;
+const CONSUMER_GROUP_MEMBER_NOT_FOUND: u32 = 5006;
+
// A consumer-group member holding zero partitions has the same empty
client-side
// assignment as a non-member; only membership tells them apart. When the group
// is deleted under such a member, the poll must surface an error (driving a
@@ -137,3 +145,97 @@ async fn
given_group_member_holds_no_partitions_when_group_deleted_should_surfac
Ok(_) => panic!("expected an error after group deletion, got a
message"),
}
}
+
+// End-to-end wire pin for the consumer-group join/leave error ladder. The
+// metadata STM unit tests pin the committed result codes; this pins that
+// server-ng actually emits them over the wire, so a client observes the same
+// codes the legacy server returns. Binary transports only: the HTTP client
+// has no join/leave (stateless sessions carry no member identity, the SDK
+// returns FeatureUnavailable client-side), so the ladder cannot run there.
+#[iggy_harness(test_client_transport = [Tcp, WebSocket, Quic])]
+async fn
given_join_and_leave_failures_when_sent_over_the_wire_should_return_legacy_error_codes(
+ harness: &TestHarness,
+) {
+ let root_client = harness.root_client().await.expect("root client");
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+ let topic_id = Identifier::named(TOPIC_NAME).unwrap();
+ let group_id = Identifier::named(CONSUMER_GROUP_NAME).unwrap();
+ // Stands in for whichever level is absent; its position in the call marks
+ // the level under test.
+ let missing = Identifier::named("cg-membership-nonexistent").unwrap();
+
+ // Each level is created only after the case proving its absence is
+ // rejected, so no assertion resolves against pre-existing state.
+
+ // Nothing exists yet: the stream miss is the first rejection.
+ assert_rejected(
+ root_client
+ .join_consumer_group(&missing, &topic_id, &group_id)
+ .await,
+ STREAM_ID_NOT_FOUND,
+ "join with a missing stream",
+ );
+
+ root_client.create_stream(STREAM_NAME).await.unwrap();
+ assert_rejected(
+ root_client
+ .join_consumer_group(&stream_id, &missing, &group_id)
+ .await,
+ TOPIC_ID_NOT_FOUND,
+ "join with a missing topic",
+ );
+
+ root_client
+ .create_topic(
+ &stream_id,
+ TOPIC_NAME,
+ 1,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+ assert_rejected(
+ root_client
+ .join_consumer_group(&stream_id, &topic_id, &missing)
+ .await,
+ CONSUMER_GROUP_ID_NOT_FOUND,
+ "join with a missing group",
+ );
+
+ let created = root_client
+ .create_consumer_group(&stream_id, &topic_id, CONSUMER_GROUP_NAME)
+ .await
+ .unwrap();
+ // First group on the topic: ids are 0-based to match the legacy server,
+ // pinned on the wire-visible response id.
+ assert_eq!(created.id, 0, "first consumer group id must be 0-based");
+ // The group resolves now, but this client never joined it: the member
check
+ // is the loud rejection, distinct from the missing-group case above.
+ assert_rejected(
+ root_client
+ .leave_consumer_group(&stream_id, &topic_id, &group_id)
+ .await,
+ CONSUMER_GROUP_MEMBER_NOT_FOUND,
+ "leave a group the client never joined",
+ );
+
+ // Non-vacuous control: the same client joins the group whose leave just
+ // returned member-not-found, proving the setup is live.
+ root_client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .expect("join on an existing group must succeed");
+}
+
+fn assert_rejected(result: Result<(), IggyError>, expected_code: u32, context:
&str) {
+ let error = result.expect_err(context);
+ assert_eq!(
+ error.as_code(),
+ expected_code,
+ "{context}: expected wire error code {expected_code}, got {}
({error})",
+ error.as_code(),
+ );
+}
diff --git a/core/metadata/src/stm/consumer_group.rs
b/core/metadata/src/stm/consumer_group.rs
index c85390ad9..1daf6a72f 100644
--- a/core/metadata/src/stm/consumer_group.rs
+++ b/core/metadata/src/stm/consumer_group.rs
@@ -27,7 +27,10 @@
//! offset -- making a metadata->offset purge unnecessary for correctness.
use crate::stm::StateHandler;
-use crate::stm::result::{ApplyReply, CreateConsumerGroupResult,
DeleteConsumerGroupResult};
+use crate::stm::result::{
+ ApplyReply, CreateConsumerGroupResult, DeleteConsumerGroupResult,
JoinConsumerGroupResult,
+ LeaveConsumerGroupResult,
+};
use crate::stm::stream::StreamsInner;
use bytes::Bytes;
@@ -560,22 +563,29 @@ impl StateHandler for DeleteConsumerGroupRequest {
state: &mut StreamsInner,
_timestamp: iggy_common::IggyTimestamp,
) -> ApplyReply {
- let removed = {
- let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id)
else {
- return ApplyReply::err(DeleteConsumerGroupResult::NotFound);
- };
- if let Some(group_id) = topic.resolve_group_id(&self.group_id)
- && let Some(group) = topic.consumer_groups.remove(&group_id)
- {
- topic.consumer_group_index.remove(&group.name);
- true
- } else {
- false
- }
+ // Same level-by-level resolution as Join/Leave, mirroring the legacy
+ // `resolve_consumer_group` ladder instead of collapsing a missing
+ // stream or topic into the group-not-found code.
+ let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
+ return ApplyReply::err(DeleteConsumerGroupResult::StreamNotFound);
};
- if !removed {
- return ApplyReply::err(DeleteConsumerGroupResult::NotFound);
- }
+ let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id)
else {
+ return ApplyReply::err(DeleteConsumerGroupResult::TopicNotFound);
+ };
+ let Some(topic) = state
+ .items
+ .get_mut(stream_id)
+ .and_then(|stream| stream.topics.get_mut(topic_id))
+ else {
+ return ApplyReply::err(DeleteConsumerGroupResult::TopicNotFound);
+ };
+ let Some(group_id) = topic.resolve_group_id(&self.group_id) else {
+ return
ApplyReply::err(DeleteConsumerGroupResult::ConsumerGroupNotFound);
+ };
+ let Some(group) = topic.consumer_groups.remove(&group_id) else {
+ return
ApplyReply::err(DeleteConsumerGroupResult::ConsumerGroupNotFound);
+ };
+ topic.consumer_group_index.remove(&group.name);
// Bump the partition-shaping revision so the reconciler's fast-skip
// doesn't pass over the delete: it reclaims the group's leftover
// offsets on the topic's surviving partitions.
@@ -589,17 +599,30 @@ impl StateHandler for DeleteConsumerGroupRequest {
impl StateHandler for JoinConsumerGroupRequest {
type State = StreamsInner;
fn apply(&self, state: &mut StreamsInner, timestamp:
iggy_common::IggyTimestamp) -> ApplyReply {
- let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id)
else {
- return ApplyReply::ok(Bytes::new());
+ // Resolve level by level so the committed rejection names the level
that
+ // missed, mirroring the legacy `resolve_consumer_group` ladder
instead of
+ // the old silent-OK no-op.
+ let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
+ return ApplyReply::err(JoinConsumerGroupResult::StreamNotFound);
+ };
+ let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id)
else {
+ return ApplyReply::err(JoinConsumerGroupResult::TopicNotFound);
+ };
+ let Some(topic) = state
+ .items
+ .get_mut(stream_id)
+ .and_then(|stream| stream.topics.get_mut(topic_id))
+ else {
+ return ApplyReply::err(JoinConsumerGroupResult::TopicNotFound);
};
let Some(group_id) = topic.resolve_group_id(&self.group_id) else {
- return ApplyReply::ok(Bytes::new());
+ return
ApplyReply::err(JoinConsumerGroupResult::ConsumerGroupNotFound);
};
// Snapshot the live partition ids before taking a mutable borrow of
the
// group (both borrow the topic).
let partition_ids: Vec<usize> = topic.partitions.iter().map(|p|
p.id).collect();
let Some(group) = topic.consumer_groups.get_mut(&group_id) else {
- return ApplyReply::ok(Bytes::new());
+ return
ApplyReply::err(JoinConsumerGroupResult::ConsumerGroupNotFound);
};
// Idempotent: a re-join from the same client keeps its membership.
let already = group
@@ -633,26 +656,43 @@ impl StateHandler for LeaveConsumerGroupRequest {
state: &mut StreamsInner,
_timestamp: iggy_common::IggyTimestamp,
) -> ApplyReply {
- let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id)
else {
- return ApplyReply::ok(Bytes::new());
+ // Same level-by-level resolution as Join, surfacing a loud rejection
+ // instead of the old silent-OK no-op when a level is gone.
+ let Some(stream_id) = state.resolve_stream_id(&self.stream_id) else {
+ return ApplyReply::err(LeaveConsumerGroupResult::StreamNotFound);
+ };
+ let Some(topic_id) = state.resolve_topic_id(stream_id, &self.topic_id)
else {
+ return ApplyReply::err(LeaveConsumerGroupResult::TopicNotFound);
+ };
+ let Some(topic) = state
+ .items
+ .get_mut(stream_id)
+ .and_then(|stream| stream.topics.get_mut(topic_id))
+ else {
+ return ApplyReply::err(LeaveConsumerGroupResult::TopicNotFound);
};
let Some(group_id) = topic.resolve_group_id(&self.group_id) else {
- return ApplyReply::ok(Bytes::new());
+ return
ApplyReply::err(LeaveConsumerGroupResult::ConsumerGroupNotFound);
};
let partition_ids: Vec<usize> = topic.partitions.iter().map(|p|
p.id).collect();
let Some(group) = topic.consumer_groups.get_mut(&group_id) else {
- return ApplyReply::ok(Bytes::new());
+ return
ApplyReply::err(LeaveConsumerGroupResult::ConsumerGroupNotFound);
};
let member_key = group
.members
.iter()
.find(|(_, m)| m.client_id == self.client_id)
.map(|(key, _)| key);
- if let Some(key) = member_key {
- group.members.remove(key);
- group.rebalance_members(&partition_ids);
- state.recompute_pending_revocations_count();
- }
+ let Some(key) = member_key else {
+ // Group exists but this client never joined it: mirror legacy's
+ // ConsumerGroupMemberNotFound instead of a silent no-op. The
+ // server-internal RemoveConsumerGroupMember (disconnect cleanup)
+ // stays idempotent -- only this client-facing Leave is loud.
+ return
ApplyReply::err(LeaveConsumerGroupResult::ConsumerGroupMemberNotFound);
+ };
+ group.members.remove(key);
+ group.rebalance_members(&partition_ids);
+ state.recompute_pending_revocations_count();
ApplyReply::ok(Bytes::new())
}
}
@@ -958,6 +998,45 @@ mod tests {
)
}
+ fn join(
+ state: &mut StreamsInner,
+ stream: u32,
+ topic: u32,
+ group: u32,
+ client_id: u128,
+ ) -> ApplyReply {
+ StateHandler::apply(
+ &JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream),
+ topic_id: WireIdentifier::numeric(topic),
+ group_id: WireIdentifier::numeric(group),
+ client_id,
+ in_flight: Vec::new(),
+ },
+ state,
+ IggyTimestamp::now(),
+ )
+ }
+
+ fn leave(
+ state: &mut StreamsInner,
+ stream: u32,
+ topic: u32,
+ group: u32,
+ client_id: u128,
+ ) -> ApplyReply {
+ StateHandler::apply(
+ &LeaveConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream),
+ topic_id: WireIdentifier::numeric(topic),
+ group_id: WireIdentifier::numeric(group),
+ client_id,
+ },
+ state,
+ IggyTimestamp::now(),
+ )
+ }
+
#[test]
fn
given_duplicate_name_when_apply_create_consumer_group_should_return_name_already_exists()
{
let mut state = streams_with_topic();
@@ -1018,7 +1097,149 @@ mod tests {
&mut state,
IggyTimestamp::now(),
);
- assert_eq!(apply.code, u32::from(DeleteConsumerGroupResult::NotFound));
+ assert_eq!(
+ apply.code,
+ u32::from(DeleteConsumerGroupResult::ConsumerGroupNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_levels_when_apply_delete_consumer_group_should_mirror_resolution_ladder()
{
+ let mut state = streams_with_topic();
+ let missing_stream = StateHandler::apply(
+ &DeleteConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(999),
+ topic_id: WireIdentifier::numeric(0),
+ group_id: WireIdentifier::numeric(0),
+ },
+ &mut state,
+ IggyTimestamp::now(),
+ );
+ assert_eq!(
+ missing_stream.code,
+ u32::from(DeleteConsumerGroupResult::StreamNotFound)
+ );
+
+ let missing_topic = StateHandler::apply(
+ &DeleteConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(999),
+ group_id: WireIdentifier::numeric(0),
+ },
+ &mut state,
+ IggyTimestamp::now(),
+ );
+ assert_eq!(
+ missing_topic.code,
+ u32::from(DeleteConsumerGroupResult::TopicNotFound)
+ );
+ }
+
+ #[test]
+ fn given_existing_group_when_apply_join_consumer_group_should_succeed() {
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "group").code, 0);
+
+ // Group ids are 0-based, so the group just created resolves as id 0.
+ let apply = join(&mut state, 0, 0, 0, 1);
+ assert_eq!(apply.code, 0);
+ assert!(apply.body.is_empty());
+
+ // A re-join from the same client is an idempotent success.
+ assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0);
+ }
+
+ #[test]
+ fn
given_missing_stream_when_apply_join_consumer_group_should_return_stream_not_found()
{
+ let mut state = streams_with_topic();
+ let apply = join(&mut state, 999, 0, 0, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(JoinConsumerGroupResult::StreamNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_topic_when_apply_join_consumer_group_should_return_topic_not_found()
{
+ let mut state = streams_with_topic();
+ let apply = join(&mut state, 0, 999, 0, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(JoinConsumerGroupResult::TopicNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_group_when_apply_join_consumer_group_should_return_consumer_group_not_found()
{
+ let mut state = streams_with_topic();
+ let apply = join(&mut state, 0, 0, 999, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(JoinConsumerGroupResult::ConsumerGroupNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn given_joined_member_when_apply_leave_consumer_group_should_succeed() {
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "group").code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0);
+
+ let apply = leave(&mut state, 0, 0, 0, 1);
+ assert_eq!(apply.code, 0);
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_absent_member_when_apply_leave_consumer_group_should_return_member_not_found()
{
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "group").code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, 1).code, 0);
+
+ // Client 2 is not a member of the group, which does exist.
+ let apply = leave(&mut state, 0, 0, 0, 2);
+ assert_eq!(
+ apply.code,
+ u32::from(LeaveConsumerGroupResult::ConsumerGroupMemberNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_stream_when_apply_leave_consumer_group_should_return_stream_not_found()
{
+ let mut state = streams_with_topic();
+ let apply = leave(&mut state, 999, 0, 0, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(LeaveConsumerGroupResult::StreamNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_topic_when_apply_leave_consumer_group_should_return_topic_not_found()
{
+ let mut state = streams_with_topic();
+ let apply = leave(&mut state, 0, 999, 0, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(LeaveConsumerGroupResult::TopicNotFound)
+ );
+ assert!(apply.body.is_empty());
+ }
+
+ #[test]
+ fn
given_missing_group_when_apply_leave_consumer_group_should_return_consumer_group_not_found()
+ {
+ let mut state = streams_with_topic();
+ let apply = leave(&mut state, 0, 0, 999, 1);
+ assert_eq!(
+ apply.code,
+ u32::from(LeaveConsumerGroupResult::ConsumerGroupNotFound)
+ );
assert!(apply.body.is_empty());
}
}
diff --git a/core/metadata/src/stm/result.rs b/core/metadata/src/stm/result.rs
index f3be7f775..67a0a1383 100644
--- a/core/metadata/src/stm/result.rs
+++ b/core/metadata/src/stm/result.rs
@@ -230,7 +230,30 @@ result_enum!(CreateConsumerGroupResult {
TopicNotFound = 2010,
NameAlreadyExists = 5004,
});
-result_enum!(DeleteConsumerGroupResult { NotFound = 5000 });
+// Delete/Join/Leave resolve stream -> topic -> group inside the apply (the
+// authz gate passes a resolution miss through), so their codes mirror the
+// legacy `resolve_consumer_group` ladder -- StreamIdNotFound / TopicIdNotFound
+// / ConsumerGroupIdNotFound. One divergence: when the caller is unauthorized
+// AND the group is missing, legacy resolves first and leaks the miss (5000)
+// while ng commits Unauthorized (41) without confirming existence. Leave also
+// mirrors legacy's post-resolution member check: leaving a group the client
+// never joined returns ConsumerGroupMemberNotFound.
+result_enum!(DeleteConsumerGroupResult {
+ StreamNotFound = 1009,
+ TopicNotFound = 2010,
+ ConsumerGroupNotFound = 5000,
+});
+result_enum!(JoinConsumerGroupResult {
+ StreamNotFound = 1009,
+ TopicNotFound = 2010,
+ ConsumerGroupNotFound = 5000,
+});
+result_enum!(LeaveConsumerGroupResult {
+ StreamNotFound = 1009,
+ TopicNotFound = 2010,
+ ConsumerGroupNotFound = 5000,
+ ConsumerGroupMemberNotFound = 5006,
+});
/// `IggyError::Unauthorized`. Any control-plane op can commit as an in-apply
/// authorization no-op, so this code is valid for every op regardless of its
@@ -282,6 +305,8 @@ pub const fn result_code_recognized(operation: Operation,
code: u32) -> bool {
}
Operation::CreateConsumerGroup =>
CreateConsumerGroupResult::from_u32(code).is_some(),
Operation::DeleteConsumerGroup =>
DeleteConsumerGroupResult::from_u32(code).is_some(),
+ Operation::JoinConsumerGroup =>
JoinConsumerGroupResult::from_u32(code).is_some(),
+ Operation::LeaveConsumerGroup =>
LeaveConsumerGroupResult::from_u32(code).is_some(),
_ => true,
}
}
@@ -561,9 +586,49 @@ mod tests {
u32::from(CreateConsumerGroupResult::NameAlreadyExists),
IggyError::ConsumerGroupNameAlreadyExists(String::new(),
id()).as_code(),
);
+ let consumer_group_not_found =
IggyError::ConsumerGroupIdNotFound(id(), id()).as_code();
+
+ // Delete/Join/Leave mirror the legacy `resolve_consumer_group` error
+ // ladder.
+ assert_eq!(
+ u32::from(DeleteConsumerGroupResult::StreamNotFound),
+ stream_not_found
+ );
+ assert_eq!(
+ u32::from(DeleteConsumerGroupResult::TopicNotFound),
+ topic_not_found
+ );
+ assert_eq!(
+ u32::from(DeleteConsumerGroupResult::ConsumerGroupNotFound),
+ consumer_group_not_found,
+ );
+ assert_eq!(
+ u32::from(JoinConsumerGroupResult::StreamNotFound),
+ stream_not_found
+ );
+ assert_eq!(
+ u32::from(JoinConsumerGroupResult::TopicNotFound),
+ topic_not_found
+ );
+ assert_eq!(
+ u32::from(JoinConsumerGroupResult::ConsumerGroupNotFound),
+ consumer_group_not_found,
+ );
+ assert_eq!(
+ u32::from(LeaveConsumerGroupResult::StreamNotFound),
+ stream_not_found
+ );
+ assert_eq!(
+ u32::from(LeaveConsumerGroupResult::TopicNotFound),
+ topic_not_found
+ );
+ assert_eq!(
+ u32::from(LeaveConsumerGroupResult::ConsumerGroupNotFound),
+ consumer_group_not_found,
+ );
assert_eq!(
- u32::from(DeleteConsumerGroupResult::NotFound),
- IggyError::ConsumerGroupIdNotFound(id(), id()).as_code(),
+ u32::from(LeaveConsumerGroupResult::ConsumerGroupMemberNotFound),
+ IggyError::ConsumerGroupMemberNotFound(0, id(), id()).as_code(),
);
// Unauthorized (41) - the global in-apply RBAC denial code.
diff --git a/core/metadata/src/stm/snapshot.rs
b/core/metadata/src/stm/snapshot.rs
index 8d61cb23e..afe9a7081 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -361,7 +361,10 @@ mod tests {
purge_generation: 0,
}],
consumer_groups: Vec::new(),
- next_consumer_group_id: 1,
+ // Nonzero and distinct from every id above so the
+ // roundtrip assert below proves the field survives
+ // instead of matching a default.
+ next_consumer_group_id: 5,
},
)],
},
@@ -388,6 +391,9 @@ mod tests {
let (_, topic) = &stream.topics[0];
assert_eq!(topic.partitions.len(), 1);
assert_eq!(topic.partitions[0].consensus_group_id, 33);
+ // Never-reuse counter: `#[serde(default)]` would silently restore 0 if
+ // the field were dropped from the wire format, so pin its survival.
+ assert_eq!(topic.next_consumer_group_id, 5);
}
#[test]
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 1b5f6b9e6..d6a9524bd 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -179,10 +179,11 @@ pub struct Topic {
/// key (keyed by group id) can't be inherited by a recreated group.
///
/// Ceiling: the partition-plane offset key is `u32`, so a group id must
stay
- /// within `u32::MAX` (the wire rewrite in `server-ng` truncates to u32 and
- /// `expect`s this). ~4 billion group creates on a single topic is
- /// unreachable in practice, but the cap is real -- past it the wire id
would
- /// wrap and could collide with a live group's offset key.
+ /// within `u32::MAX` (the wire rewrite in `server-ng` clamps past-ceiling
+ /// ids to `u32::MAX` rather than panic). ~4 billion group creates on a
+ /// single topic is unreachable in practice, but the cap is real -- past it
+ /// clamped wire ids all collide on `u32::MAX`, including with a live
+ /// group's offset key.
pub next_consumer_group_id: u64,
}
@@ -201,7 +202,7 @@ impl Default for Topic {
round_robin_counter: Arc::new(AtomicUsize::new(0)),
consumer_groups: AHashMap::default(),
consumer_group_index: AHashMap::default(),
- next_consumer_group_id: 1,
+ next_consumer_group_id: 0,
}
}
}
@@ -229,7 +230,7 @@ impl Topic {
round_robin_counter: Arc::new(AtomicUsize::new(0)),
consumer_groups: AHashMap::default(),
consumer_group_index: AHashMap::default(),
- next_consumer_group_id: 1,
+ next_consumer_group_id: 0,
}
}
@@ -1416,7 +1417,7 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
round_robin_counter: Arc::new(AtomicUsize::new(0)),
consumer_groups: AHashMap::default(),
consumer_group_index: AHashMap::default(),
- next_consumer_group_id: 1,
+ next_consumer_group_id: 0,
};
let inserted = stream.topics.insert(topic);
@@ -1862,17 +1863,14 @@ impl Snapshotable for Streams {
.iter()
.map(|(_, group_snap)|
(Arc::from(group_snap.name.as_str()), group_snap.id))
.collect(),
- next_consumer_group_id: topic_snap
- .next_consumer_group_id
- .max(
- topic_snap
- .consumer_groups
- .iter()
- .map(|(id, _)| id + 1)
- .max()
- .unwrap_or(1),
- )
- .max(1),
+ next_consumer_group_id:
topic_snap.next_consumer_group_id.max(
+ topic_snap
+ .consumer_groups
+ .iter()
+ .map(|(id, _)| id + 1)
+ .max()
+ .unwrap_or(0),
+ ),
consumer_groups: topic_snap
.consumer_groups
.into_iter()
diff --git a/core/sdk/src/clients/consumer.rs b/core/sdk/src/clients/consumer.rs
index 980971408..5400d7c85 100644
--- a/core/sdk/src/clients/consumer.rs
+++ b/core/sdk/src/clients/consumer.rs
@@ -39,7 +39,7 @@ use tokio::sync::Notify;
use tokio::task::JoinHandle;
use tokio::time;
use tokio::time::sleep;
-use tracing::{error, info, trace, warn};
+use tracing::{debug, error, info, trace, warn};
const ORDERING: std::sync::atomic::Ordering =
std::sync::atomic::Ordering::SeqCst;
type PollMessagesFuture = Pin<Box<dyn Future<Output = Result<PolledMessages,
IggyError>> + Send>>;
@@ -1209,7 +1209,10 @@ impl IggyConsumer {
.leave_consumer_group(&self.stream_id, &self.topic_id,
&group_id)
.await
{
- warn!(
+ // Expected on clean teardown after an explicit leave (member
+ // not found) or when the group was deleted underneath the
+ // consumer, so this is debug, not a warning.
+ debug!(
"Failed to leave consumer group: {group_id} for stream:
{}, topic: {}. {error}",
self.stream_id, self.topic_id
);
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 7a185a80d..2bed5b4b3 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -44,6 +44,8 @@ enabled = true
# Specifies the network address and port for the HTTP server.
# The format is "HOST:PORT". For example, "127.0.0.1:3000" listens on
localhost only on port 3000.
+# In cluster mode the HOST still picks the bind interface, while the port
+# comes from this node's cluster.nodes ports.http entry.
address = "127.0.0.1:3000"
# Maximum size of the request body in bytes. For security reasons, the default
limit is 2 MB.
@@ -639,6 +641,13 @@ ca_file = ""
# node's identity is resolved at launch from the '--replica-id <N>' CLI
# flag, which selects the entry in this list that describes the current
# node. All other entries are remote peers.
+#
+# 'ip' is the address the cluster advertises for the node (cluster metadata,
+# follower-to-primary HTTP forwarding); it is not the bind interface for
+# quic/http/websocket, which comes from each transport's own 'address'
+# setting above. 'ports' is the single source of listener ports in cluster
+# mode: every enabled transport needs an explicit per-node port, otherwise
+# the server refuses to start.
[[cluster.nodes]]
name = "iggy-node-1"
ip = "127.0.0.1"
@@ -656,7 +665,7 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket =
8093, tcp_replica =
# name = "iggy-node-3"
# ip = "192.168.1.100"
# replica_id = 2
-# ports = { tcp = 8092, http = 3002, tcp_replica = 9092 }
+# ports = { tcp = 8092, quic = 8082, http = 3002, websocket = 8094,
tcp_replica = 9092 }
# Sharding configuration
[system.sharding]
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index b7d578a3f..1fcd78495 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -481,6 +481,7 @@ struct TcpTopology {
replica_listen_addr: Option<SocketAddr>,
ws_listen_addr: Option<SocketAddr>,
quic_listen_addr: Option<SocketAddr>,
+ http_listen_addr: Option<SocketAddr>,
tcp_tls_listen_addr: Option<SocketAddr>,
peers: Vec<(u8, SocketAddr)>,
}
@@ -1472,20 +1473,14 @@ fn spawn_shutdown_watchdog(
/// Copy the configured cluster roster plus this node's own client ports into
/// the shared [`ClusterRoster`] so the binary `GetClusterMetadata` read serves
-/// the real topology. `self_*` back only the cluster-disabled self-synthesis;
-/// the HTTP port is read from config since HTTP binds outside this topology.
+/// the real topology. `self_*` back only the cluster-disabled self-synthesis
+/// and carry the requested listener ports from the resolved topology, not the
+/// bound ones (a `:0` wildcard is reported as 0).
fn build_cluster_roster(
config: &ServerNgConfig,
topology: &TcpTopology,
metadata_view: Arc<AtomicU64>,
) -> ClusterRoster {
- let http_port = if config.http.enabled {
- parse_socket_addr("http.address", &config.http.address)
- .ok()
- .map(|addr| addr.port())
- } else {
- None
- };
ClusterRoster {
enabled: config.cluster.enabled,
name: config.cluster.name.clone(),
@@ -1494,7 +1489,7 @@ fn build_cluster_roster(
self_ports: configs::ng_cluster::TransportPorts {
tcp: Some(topology.client_listen_addr.port()),
quic: topology.quic_listen_addr.map(|addr| addr.port()),
- http: http_port,
+ http: topology.http_listen_addr.map(|addr| addr.port()),
websocket: topology.ws_listen_addr.map(|addr| addr.port()),
tcp_replica: None,
},
@@ -2026,6 +2021,8 @@ fn resolve_tcp_topology(
)?;
let default_quic_addr =
resolve_optional_listener_addr(config.quic.enabled, "quic.address",
&config.quic.address)?;
+ let default_http_addr =
+ resolve_optional_listener_addr(config.http.enabled, "http.address",
&config.http.address)?;
if !config.cluster.enabled {
if let Some(replica_id) = current_replica_id
&& replica_id != SHARD_REPLICA_ID
@@ -2048,6 +2045,7 @@ fn resolve_tcp_topology(
replica_listen_addr:
Some(SocketAddr::new(default_client_addr.ip(), 0)),
ws_listen_addr: default_ws_addr,
quic_listen_addr: default_quic_addr,
+ http_listen_addr: default_http_addr,
tcp_tls_listen_addr:
config.tcp.tls.enabled.then_some(default_client_addr),
peers: Vec::new(),
});
@@ -2068,19 +2066,24 @@ fn resolve_tcp_topology(
count: config.cluster.nodes.len(),
}
})?;
- let (client_listen_addr, ws_listen_addr, quic_listen_addr) =
resolve_cluster_client_addrs(
+ let ClusterClientAddrs {
+ client: client_listen_addr,
+ ws: ws_listen_addr,
+ quic: quic_listen_addr,
+ http: http_listen_addr,
+ } = resolve_cluster_client_addrs(
self_node,
- default_client_addr,
default_ws_addr,
default_quic_addr,
+ default_http_addr,
)?;
- let replica_port =
- self_node
- .ports
- .tcp_replica
- .ok_or(ServerNgError::ClusterReplicaPortMissing {
- replica_id: self_node.replica_id,
- })?;
+ let replica_port = self_node
+ .ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterPortMissing {
+ transport: "tcp_replica",
+ replica_id: self_node.replica_id,
+ })?;
let replica_listen_addr = Some(socket_addr_from_parts(
"cluster.nodes[*].ports.tcp_replica",
&self_node.ip,
@@ -2096,6 +2099,7 @@ fn resolve_tcp_topology(
replica_listen_addr,
ws_listen_addr,
quic_listen_addr,
+ http_listen_addr,
tcp_tls_listen_addr:
config.tcp.tls.enabled.then_some(client_listen_addr),
peers,
})
@@ -2112,44 +2116,82 @@ fn resolve_optional_listener_addr(
Ok(None)
}
+/// Client-facing listener addresses resolved for this cluster node. Each port
+/// comes from the node's roster entry; there is no fallback to the top-level
+/// listener port, an enabled transport without a roster port refuses to boot.
+/// ws/quic/http keep the bind interface from their own `address` config (the
+/// roster ip is advertised, not bound); tcp binds the roster ip directly.
+struct ClusterClientAddrs {
+ client: SocketAddr,
+ ws: Option<SocketAddr>,
+ quic: Option<SocketAddr>,
+ http: Option<SocketAddr>,
+}
+
fn resolve_cluster_client_addrs(
self_node: &configs::ng_cluster::ClusterNodeConfig,
- default_client_addr: SocketAddr,
default_ws_addr: Option<SocketAddr>,
default_quic_addr: Option<SocketAddr>,
-) -> Result<(SocketAddr, Option<SocketAddr>, Option<SocketAddr>),
ServerNgError> {
+ default_http_addr: Option<SocketAddr>,
+) -> Result<ClusterClientAddrs, ServerNgError> {
let client_port = self_node
.ports
.tcp
- .unwrap_or_else(|| default_client_addr.port());
- let client_listen_addr =
- socket_addr_from_parts("cluster.nodes[*].ports.tcp", &self_node.ip,
client_port)?;
- let ws_listen_addr = resolve_cluster_optional_addr(
- self_node,
- "cluster.nodes[*].ports.websocket",
- default_ws_addr,
- |ports| ports.websocket,
- )?;
- let quic_listen_addr = resolve_cluster_optional_addr(
- self_node,
- "cluster.nodes[*].ports.quic",
- default_quic_addr,
- |ports| ports.quic,
- )?;
- Ok((client_listen_addr, ws_listen_addr, quic_listen_addr))
+ .ok_or(ServerNgError::ClusterPortMissing {
+ transport: "tcp",
+ replica_id: self_node.replica_id,
+ })?;
+ let client = socket_addr_from_parts("cluster.nodes[*].ports.tcp",
&self_node.ip, client_port)?;
+ let ws = resolve_cluster_optional_addr(self_node, "websocket",
default_ws_addr, |ports| {
+ ports.websocket
+ })?;
+ let quic =
+ resolve_cluster_optional_addr(self_node, "quic", default_quic_addr,
|ports| ports.quic)?;
+ let http =
+ resolve_cluster_optional_addr(self_node, "http", default_http_addr,
|ports| ports.http)?;
+ Ok(ClusterClientAddrs {
+ client,
+ ws,
+ quic,
+ http,
+ })
}
fn resolve_cluster_optional_addr(
self_node: &configs::ng_cluster::ClusterNodeConfig,
- context: &'static str,
+ transport: &'static str,
default_addr: Option<SocketAddr>,
port_selector: impl Fn(&configs::ng_cluster::TransportPorts) ->
Option<u16>,
) -> Result<Option<SocketAddr>, ServerNgError> {
let Some(default_addr) = default_addr else {
return Ok(None);
};
- let port = port_selector(&self_node.ports).unwrap_or_else(||
default_addr.port());
- socket_addr_from_parts(context, &self_node.ip, port).map(Some)
+ // No fallback to the top-level port: two same-host nodes leaving the same
+ // transport port unset would race for one socket. Either the roster is
+ // explicit or the server refuses to boot.
+ let port =
port_selector(&self_node.ports).ok_or(ServerNgError::ClusterPortMissing {
+ transport,
+ replica_id: self_node.replica_id,
+ })?;
+ // The roster ip is what the cluster advertises (metadata, follower-to-
+ // primary HTTP forwarding targets); the transport's own `address` decides
+ // the bind interface. Merging keeps a loopback-only `127.0.0.1` private
+ // and a `0.0.0.0` wide in cluster mode instead of silently rebinding to
+ // the roster interface.
+ let listen_addr = SocketAddr::new(default_addr.ip(), port);
+ if !listen_addr.ip().is_unspecified()
+ && self_node
+ .ip
+ .parse::<IpAddr>()
+ .is_ok_and(|roster_ip| roster_ip != listen_addr.ip())
+ {
+ warn!(
+ "{transport} listener binds {listen_addr} but the roster
advertises {}:{port}; \
+ peers and clients dialing the advertised endpoint will not reach
this node",
+ self_node.ip
+ );
+ }
+ Ok(Some(listen_addr))
}
fn resolve_cluster_replica_peers(
@@ -2161,12 +2203,13 @@ fn resolve_cluster_replica_peers(
if node.replica_id == self_replica_id {
continue;
}
- let replica_port =
- node.ports
- .tcp_replica
- .ok_or(ServerNgError::ClusterReplicaPortMissing {
- replica_id: node.replica_id,
- })?;
+ let replica_port = node
+ .ports
+ .tcp_replica
+ .ok_or(ServerNgError::ClusterPortMissing {
+ transport: "tcp_replica",
+ replica_id: node.replica_id,
+ })?;
peers.push((
node.replica_id,
socket_addr_from_parts("cluster.nodes[*].ports.tcp_replica",
&node.ip, replica_port)?,
@@ -2208,8 +2251,7 @@ async fn start_tcp_runtime(
// HTTP is served over TCP but sits outside the replica_io / manual client
// reactor, so it binds independently. Shard-0 gating comes from the sole
// caller of this function.
- if config.http.enabled {
- let http_addr = parse_socket_addr("http.address",
&config.http.address)?;
+ if let Some(http_addr) = topology.http_listen_addr {
let self_ports = configs::ng_cluster::TransportPorts {
tcp: config
.tcp
@@ -3303,4 +3345,69 @@ mod tests {
"expected MetadataHandoffAborted, got {err:?}"
);
}
+
+ fn cluster_node(ip: &str, http: Option<u16>) ->
configs::ng_cluster::ClusterNodeConfig {
+ configs::ng_cluster::ClusterNodeConfig {
+ name: "node".to_owned(),
+ ip: ip.to_owned(),
+ replica_id: 0,
+ ports: configs::ng_cluster::TransportPorts {
+ tcp: Some(18070),
+ http,
+ ..Default::default()
+ },
+ }
+ }
+
+ fn addr(value: &str) -> SocketAddr {
+ value.parse().expect("valid socket address literal")
+ }
+
+ #[test]
+ fn cluster_http_addr_takes_port_from_roster() {
+ // A byte-identical top-level [http].address is shared across nodes on
+ // one host; the per-node roster port is the only port source so each
+ // node binds a distinct HTTP socket.
+ let node = cluster_node("127.0.0.1", Some(18090));
+ let addrs = resolve_cluster_client_addrs(&node, None, None,
Some(addr("127.0.0.1:3000")))
+ .expect("cluster address resolution must succeed");
+ assert_eq!(addrs.http, Some(addr("127.0.0.1:18090")));
+ }
+
+ #[test]
+ fn cluster_http_addr_merges_config_ip_with_roster_port() {
+ // Docker/Helm bind `0.0.0.0` and probe loopback; the roster ip is
+ // only the advertised address. Cluster mode must keep the configured
+ // interface and take just the port from the roster.
+ let node = cluster_node("10.0.0.5", Some(18090));
+ let addrs = resolve_cluster_client_addrs(&node, None, None,
Some(addr("0.0.0.0:3000")))
+ .expect("cluster address resolution must succeed");
+ assert_eq!(addrs.http, Some(addr("0.0.0.0:18090")));
+ }
+
+ #[test]
+ fn cluster_http_addr_requires_roster_port_for_enabled_transport() {
+ // No fallback to the top-level port: a silent default could collide
+ // with another same-host node, so a missing roster port for an
+ // enabled transport must refuse to boot.
+ let node = cluster_node("10.0.0.5", None);
+ let result = resolve_cluster_client_addrs(&node, None, None,
Some(addr("127.0.0.1:3000")));
+ assert!(matches!(
+ result,
+ Err(ServerNgError::ClusterPortMissing {
+ transport: "http",
+ replica_id: 0,
+ })
+ ));
+ }
+
+ #[test]
+ fn cluster_http_addr_is_none_when_http_disabled() {
+ // http.enabled = false collapses default_http_addr to None; no roster
+ // port can revive a listener the operator turned off.
+ let node = cluster_node("127.0.0.1", Some(18090));
+ let addrs = resolve_cluster_client_addrs(&node, None, None, None)
+ .expect("cluster address resolution must succeed");
+ assert_eq!(addrs.http, None);
+ }
}
diff --git a/core/server-ng/src/partition_reconciler.rs
b/core/server-ng/src/partition_reconciler.rs
index 37a94b32f..71968aa34 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -1759,15 +1759,15 @@ mod tests {
let ns = IggyNamespace::new(0, 0, 0);
assert!(shard.plane.partitions().contains(&ns));
- // Two groups: "dead" gets id 1, "live" gets id 2 (per-topic
monotonic).
+ // Two groups: "dead" gets id 0, "live" gets id 1 (per-topic
monotonic).
let stm = &shard.plane.metadata().mux_stm;
seed_create_consumer_group(stm, 3, 0, 0, "dead");
seed_create_consumer_group(stm, 4, 0, 0, "live");
// Offsets are keyed by the monotonic group id (the id the store path
is
// rewritten to and the read path / live-set resolve), not the name
hash.
- let dead_key: u32 = 1;
- let live_key: u32 = 2;
+ let dead_key: u32 = 0;
+ let live_key: u32 = 1;
{
let partitions = shard.plane.partitions();
let partition = partitions.get_by_ns(&ns).expect("partition
materialised");
@@ -1781,8 +1781,8 @@ mod tests {
);
}
- // Delete the "dead" group (id 1); "live" (id 2) stays.
- seed_delete_consumer_group(stm, 5, 0, 0, 1);
+ // Delete the "dead" group (id 0); "live" (id 1) stays.
+ seed_delete_consumer_group(stm, 5, 0, 0, 0);
reconcile_pass(&ctx).await;
let partitions = shard.plane.partitions();
@@ -1815,10 +1815,10 @@ mod tests {
vec![assignment(0, 1), assignment(1, 2)],
);
seed_create_consumer_group(&mux, 3, 0, 0, "cg");
- // Single member owns every partition (group id 1, the first in topic).
- seed_join_consumer_group(&mux, 4, 0, 0, 1, 100);
+ // Single member owns every partition (group id 0, the first in topic).
+ seed_join_consumer_group(&mux, 4, 0, 0, 0, 100);
- let group = WireIdentifier::numeric(1);
+ let group = WireIdentifier::numeric(0);
let stream = WireIdentifier::numeric(0);
let topic = WireIdentifier::numeric(0);
let assigned = |mux: &TestMux| -> Vec<u32> {
@@ -1888,13 +1888,13 @@ mod tests {
"topic-dc",
vec![assignment(0, 1), assignment(1, 2)],
);
- seed_create_consumer_group(&mux, 3, 0, 0, "cg"); // group id 1
- seed_join_consumer_group(&mux, 4, 0, 0, 1, 100);
- seed_join_consumer_group(&mux, 5, 0, 0, 1, 200);
+ seed_create_consumer_group(&mux, 3, 0, 0, "cg"); // group id 0
+ seed_join_consumer_group(&mux, 4, 0, 0, 0, 100);
+ seed_join_consumer_group(&mux, 5, 0, 0, 0, 200);
let stream = WireIdentifier::numeric(0);
let topic = WireIdentifier::numeric(0);
- let group = WireIdentifier::numeric(1);
+ let group = WireIdentifier::numeric(0);
let assigned = |client: u128| -> Option<Vec<u32>> {
mux.streams()
.consumer_group_member_assignment(&stream, &topic, &group,
client)
diff --git a/core/server-ng/src/server_error.rs
b/core/server-ng/src/server_error.rs
index f2987c6e0..b19f26778 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -127,8 +127,14 @@ pub enum ServerNgError {
cluster.enabled=true with a matching nodes[] entry, or drop
--replica-id"
)]
ReplicaIdRequiresCluster { supplied: u8, default: u8 },
- #[error("cluster node for replica {replica_id} is missing tcp_replica
port")]
- ClusterReplicaPortMissing { replica_id: u8 },
+ #[error(
+ "cluster node for replica {replica_id} is missing ports.{transport};
cluster mode \
+ requires an explicit roster port for every enabled transport"
+ )]
+ ClusterPortMissing {
+ transport: &'static str,
+ replica_id: u8,
+ },
#[error(
"cluster bootstrap with empty metadata requires both {username_env}
and {password_env} to be set before server-ng can create the root user
deterministically"
)]
diff --git a/core/simulator/src/workload/ops/delete_consumer_group.rs
b/core/simulator/src/workload/ops/delete_consumer_group.rs
index 17e2281cd..cfbe89961 100644
--- a/core/simulator/src/workload/ops/delete_consumer_group.rs
+++ b/core/simulator/src/workload/ops/delete_consumer_group.rs
@@ -15,8 +15,12 @@
// specific language governing permissions and limitations
// under the License.
-//! `DeleteConsumerGroup` op. Targets `Ok` (a live group) or `NotFound` (a live
-//! stream/topic with a fabricated group name).
+//! `DeleteConsumerGroup` op.
+//!
+//! Targets `Ok` (a live group), `StreamNotFound` (a fabricated parent
+//! stream), `TopicNotFound` (a live stream with a fabricated topic) or
+//! `ConsumerGroupNotFound` (a live stream/topic with a fabricated group
+//! name), mirroring the legacy resolution ladder.
use iggy_binary_protocol::RequestHeader;
use rand_xoshiro::Xoshiro256Plus;
@@ -36,7 +40,12 @@ pub struct Input {
pub group: String,
}
-pub const OUTCOMES: &[Outcome] = &[Outcome::Ok, Outcome::NotFound];
+pub const OUTCOMES: &[Outcome] = &[
+ Outcome::Ok,
+ Outcome::StreamNotFound,
+ Outcome::TopicNotFound,
+ Outcome::ConsumerGroupNotFound,
+];
pub fn sample(
shadow: &mut Shadow,
@@ -52,7 +61,20 @@ pub fn sample(
topic,
group,
}),
- Outcome::NotFound => {
+ Outcome::StreamNotFound => Some(Input {
+ stream: shadow.fabricate_absent_name("stream"),
+ topic: shadow.fabricate_absent_name("topic"),
+ group: shadow.fabricate_absent_name("cg"),
+ }),
+ Outcome::TopicNotFound => {
+ let stream = shadow.pick_stream_name(prng)?;
+ Some(Input {
+ stream,
+ topic: shadow.fabricate_absent_name("topic"),
+ group: shadow.fabricate_absent_name("cg"),
+ })
+ }
+ Outcome::ConsumerGroupNotFound => {
let (stream, topic) = shadow.pick_topic_pair(prng)?;
Some(Input {
stream,
@@ -85,6 +107,8 @@ pub fn predicted_effect(input: &Input, outcome: Outcome) ->
Effect {
topic: input.topic.clone(),
name: input.group.clone(),
},
- Outcome::NotFound => Effect::None,
+ Outcome::StreamNotFound | Outcome::TopicNotFound |
Outcome::ConsumerGroupNotFound => {
+ Effect::None
+ }
}
}
