This is an automated email from the ASF dual-hosted git repository. spetz pushed a commit to branch offset_hardening in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4cae047092835e89eb73cd92b494ac7631ab84d2 Merge: 89f4341e3 6c47193fe Author: Piotr Gankiewicz <[email protected]> AuthorDate: Sat Sep 5 08:29:10 2026 +0200 Merge branch 'master' into offset_hardening .github/workflows/pr-triage-apply.yml | 106 ++++++++++++++++++++- CONTRIBUTING.md | 16 +++- bdd/go/tests/tcp_test/offset_feature_delete.go | 6 +- .../tests/server/consumer_offset_quota_vsr.rs | 23 +++++ core/server/src/consumer_group.rs | 41 ++++++-- core/server/src/responses.rs | 11 ++- foreign/go/errors/errors_test.go | 19 ++++ 7 files changed, 207 insertions(+), 15 deletions(-) diff --cc bdd/go/tests/tcp_test/offset_feature_delete.go index 0d8d93d92,0d8d93d92..fb4dcedf9 --- a/bdd/go/tests/tcp_test/offset_feature_delete.go +++ b/bdd/go/tests/tcp_test/offset_feature_delete.go @@@ -129,10 -129,10 +129,8 @@@ var _ = ginkgo.Describe("DELETE CONSUME &partitionId, ) -- // A consumer-offset request is routed by its packed namespace, so -- // the shard that answers reports a missing resource rather than -- // naming the group. -- itShouldReturnSpecificError(err, ierror.ErrResourceNotFound) ++ // The stream and topic resolve, so the server names the group. ++ itShouldReturnSpecificError(err, ierror.ErrConsumerGroupIdNotFound) }) ginkgo.Context("and attempts to delete an offset from a non-existing stream", func() { diff --cc core/integration/tests/server/consumer_offset_quota_vsr.rs index 2ccac51c4,000000000..ee29f1b38 mode 100644,000000..100644 --- a/core/integration/tests/server/consumer_offset_quota_vsr.rs +++ b/core/integration/tests/server/consumer_offset_quota_vsr.rs @@@ -1,470 -1,0 +1,493 @@@ +// 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. + +use iggy::prelude::*; +use iggy_binary_protocol::codec::WireEncode; +use iggy_binary_protocol::consensus::Operation; +use iggy_binary_protocol::requests::consumer_offsets::StoreConsumerOffsetRequest; +use iggy_binary_protocol::{AckLevel, WireConsumer, WireIdentifier}; +use iggy_common::store_consumer_offset::StoreConsumerOffset; +use integration::harness::TestBinary; +use integration::iggy_harness; +use reqwest::StatusCode; +use std::collections::BTreeMap; +use std::fs; + +use super::http_client::HttpClient; +use super::raw_tcp; + +const STREAM_NAME: &str = "consumer-offset-quota-stream"; +const TOPIC_NAME: &str = "consumer-offset-quota-topic"; +const PARTITION_ID: u32 = 0; +const LIMIT: u32 = 4; + +#[iggy_harness( + cluster_nodes = 1, + server(partition.consumer_offsets_max = "4") +)] +async fn given_full_consumer_offset_table_when_creating_another_should_reject_without_new_file( + harness: &TestHarness, +) { + let client = harness.tcp_root_client().await.expect("TCP root client"); + let stream = Identifier::named(STREAM_NAME).expect("stream identifier"); + let topic = Identifier::named(TOPIC_NAME).expect("topic identifier"); + let stream_details = client + .create_stream(STREAM_NAME) + .await + .expect("create stream"); + let topic_details = client + .create_topic( + &stream, + TOPIC_NAME, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); + let mut messages = vec![ + IggyMessage::builder() + .payload("offset-quota".into()) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream, + &topic, + &Partitioning::partition_id(PARTITION_ID), + &mut messages, + ) + .await + .expect("seed non-empty partition"); + + client + .create_user( + "offset-poll-only", + "password123", + UserStatus::Active, + Some(Permissions { + global: GlobalPermissions::default(), + streams: Some(BTreeMap::from([( + stream_details.id as usize, + StreamPermissions { + topics: Some(BTreeMap::from([( + topic_details.id as usize, + TopicPermissions { + poll_messages: true, + ..Default::default() + }, + )])), + ..Default::default() + }, + )])), + }), + ) + .await + .expect("create a topic-scoped consumer"); + let client = harness.tcp_new_client().await.expect("consumer TCP client"); + client + .login_user("offset-poll-only", "password123") + .await + .expect("consumer login"); + + let first_consumer = Consumer::new(Identifier::numeric(1).unwrap()); + let polled = client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &first_consumer, + &PollingStrategy::first(), + 1, + true, + ) + .await + .expect("new auto-commit consumer fits"); + assert_eq!(polled.messages.len(), 1); + let first_file = harness.server().data_path().join(format!( + "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers/1", + stream_details.id, topic_details.id + )); + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(10); + while !first_file.is_file() { + assert!( + tokio::time::Instant::now() < deadline, + "auto-commit never reached its file" + ); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!( + client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &first_consumer, + &PollingStrategy::next(), + 1, + true + ) + .await + .expect("next poll") + .messages + .is_empty() + ); + + for consumer_id in 1..=LIMIT { + client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(consumer_id).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await + .expect("store offset within limit"); + } + + let rejected_consumer = + Consumer::new(Identifier::numeric(LIMIT + 1).expect("consumer identifier")); + let rejected = client + .store_consumer_offset(&rejected_consumer, &stream, &topic, Some(PARTITION_ID), 0) + .await; + assert!( + matches!(rejected, Err(IggyError::TooManyConsumerOffsets)), + "the first key above the limit must receive the typed capacity error" + ); + + client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(1).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await + .expect("existing key remains writable at the limit"); + + let poll_rejected = client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &rejected_consumer, + &PollingStrategy::first(), + 1, + true, + ) + .await; + assert!( + matches!(poll_rejected, Err(IggyError::TooManyConsumerOffsets)), + "auto-commit must not return data when its new key cannot be admitted" + ); + client + .poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &rejected_consumer, + &PollingStrategy::first(), + 1, + false, + ) + .await + .expect("the same poll succeeds when auto-commit is disabled"); + + client + .delete_consumer_offset( + &Consumer::new(Identifier::numeric(1).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + ) + .await + .expect("delete one accepted offset"); + client + .store_consumer_offset(&rejected_consumer, &stream, &topic, Some(PARTITION_ID), 0) + .await + .expect("delete releases one durable slot"); + + let mut raw = raw_tcp::connect(harness).await; + let raw_client_id = 0xC0FF_EE03; + let session = raw_tcp::register_root(&mut raw, raw_client_id).await; + let unresolved_group = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer_group(WireIdentifier::Numeric(999)), + stream_id: WireIdentifier::Numeric(stream_details.id), + topic_id: WireIdentifier::Numeric(topic_details.id), + partition_id: Some(PARTITION_ID), + offset: 0, + ack: AckLevel::Quorum, + } + .to_bytes(); + let header = raw_tcp::request_header( + Operation::StoreConsumerOffset, + raw_client_id, + session, + 1, + unresolved_group.len(), + ); + let (reply, _) = raw_tcp::exchange(&mut raw, &header, &unresolved_group).await; + assert_eq!( + raw_tcp::reply_status(&reply), + IggyError::ConsumerGroupIdNotFound(Identifier::numeric(999).unwrap(), topic.clone()) + .as_code() + ); + + let offsets_dir = harness.server().data_path().join(format!( + "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers", + stream_details.id, topic_details.id + )); + let file_count = integration::harness::disk::consumer_offset_file_ids( + &harness.server().data_path(), + stream_details.id, + topic_details.id, + PARTITION_ID, + ConsumerKind::Consumer, + ) + .expect("consumer offsets directory") + .len(); + assert_eq!(file_count, LIMIT as usize); + let groups_dir = offsets_dir + .parent() + .expect("consumer offset directory has offsets parent") + .join("groups"); + let group_file_count = fs::read_dir(groups_dir) + .map(|entries| entries.filter_map(Result::ok).count()) + .unwrap_or_default(); + assert_eq!(group_file_count, 0); + + let named_group = StoreConsumerOffsetRequest { + consumer: WireConsumer::consumer_group(WireIdentifier::named("unknown-group").unwrap()), + stream_id: WireIdentifier::Numeric(stream_details.id), + topic_id: WireIdentifier::Numeric(topic_details.id), + partition_id: Some(PARTITION_ID), + offset: 0, + ack: AckLevel::Quorum, + } + .to_bytes(); + let header = raw_tcp::request_header( + Operation::StoreConsumerOffset, + raw_client_id, + session, + 2, + named_group.len(), + ); + let (reply, _) = raw_tcp::exchange(&mut raw, &header, &named_group).await; + assert_eq!( + raw_tcp::reply_status(&reply), + IggyError::ConsumerGroupNameNotFound("unknown-group".to_owned(), topic.clone()).as_code() + ); + ++ let unknown_stream = StoreConsumerOffsetRequest { ++ consumer: WireConsumer::consumer_group(WireIdentifier::Numeric(999)), ++ stream_id: WireIdentifier::Numeric(999_999), ++ topic_id: WireIdentifier::Numeric(topic_details.id), ++ partition_id: Some(PARTITION_ID), ++ offset: 0, ++ ack: AckLevel::Quorum, ++ } ++ .to_bytes(); ++ let header = raw_tcp::request_header( ++ Operation::StoreConsumerOffset, ++ raw_client_id, ++ session, ++ 3, ++ unknown_stream.len(), ++ ); ++ let (reply, _) = raw_tcp::exchange(&mut raw, &header, &unknown_stream).await; ++ assert_eq!( ++ raw_tcp::reply_status(&reply), ++ IggyError::ResourceNotFound(String::new()).as_code(), ++ "a missing stream is not reported as a missing group" ++ ); ++ + let http = HttpClient::login_root(harness).await; + let response = http + .client + .put(http.url(&format!( + "/streams/{STREAM_NAME}/topics/{TOPIC_NAME}/consumer-offsets" + ))) + .bearer_auth(&http.token) + .json(&StoreConsumerOffset { + consumer: Consumer::new(Identifier::numeric(6).expect("consumer identifier")), + partition_id: Some(PARTITION_ID), + offset: 0, + }) + .send() + .await + .expect("HTTP capacity request"); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let body: serde_json::Value = response.json().await.expect("HTTP error body"); + assert_eq!(body["id"], 3024); + assert_eq!(body["code"], "too_many_consumer_offsets"); + + let metrics = http + .client + .get(http.url("/metrics")) + .bearer_auth(&http.token) + .send() + .await + .expect("metrics response") + .text() + .await + .expect("metrics text"); + let denied: u64 = metrics + .lines() + .filter(|line| line.starts_with("partition_consumer_offsets_denied_total{")) + .map(|line| { + line.split_whitespace() + .last() + .expect("counter value") + .parse::<u64>() + .expect("numeric counter") + }) + .sum(); + assert_eq!(denied, 3, "one explicit TCP, one poll, and one HTTP denial"); +} + +#[iggy_harness( + cluster_nodes = 1, + server(partition.consumer_offsets_max = "2") +)] +async fn given_full_consumer_offset_table_when_server_restarts_should_preserve_admission_state( + harness: &mut TestHarness, +) { + let client = harness.tcp_root_client().await.expect("TCP root client"); + let stream = Identifier::named("consumer-offset-restart-stream").expect("stream identifier"); + let topic = Identifier::named("consumer-offset-restart-topic").expect("topic identifier"); + let stream_details = client + .create_stream("consumer-offset-restart-stream") + .await + .expect("create stream"); + let topic_details = client + .create_topic( + &stream, + "consumer-offset-restart-topic", + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); + let mut messages = vec![ + IggyMessage::builder() + .payload("offset-restart".into()) + .build() + .expect("build message"), + ]; + client + .send_messages( + &stream, + &topic, + &Partitioning::partition_id(PARTITION_ID), + &mut messages, + ) + .await + .expect("seed non-empty partition"); + for consumer_id in 1..=2 { + client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(consumer_id).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await + .expect("store offset before restart"); + } + drop(client); + harness.server_mut().stop().expect("stop server"); + let offsets_dir = harness.server().data_path().join(format!( + "streams/{}/topics/{}/partitions/{PARTITION_ID}/offsets/consumers", + stream_details.id, topic_details.id + )); + fs::copy(offsets_dir.join("1"), offsets_dir.join("3")) + .expect("create first historical over-limit offset"); + fs::copy(offsets_dir.join("1"), offsets_dir.join("4")) + .expect("create second historical over-limit offset"); + harness.server_mut().start().expect("restart server"); + let client = harness + .root_client() + .await + .expect("post-restart root client"); + + let rejected = client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(5).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await; + assert!(matches!(rejected, Err(IggyError::TooManyConsumerOffsets))); + client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(4).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await + .expect("recovered existing key remains writable"); + client + .delete_consumer_offset( + &Consumer::new(Identifier::numeric(4).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + ) + .await + .expect("delete recovered key"); + client + .delete_consumer_offset( + &Consumer::new(Identifier::numeric(3).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + ) + .await + .expect("delete second historical key"); + client + .delete_consumer_offset( + &Consumer::new(Identifier::numeric(2).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + ) + .await + .expect("delete below configured limit"); + client + .store_consumer_offset( + &Consumer::new(Identifier::numeric(5).expect("consumer identifier")), + &stream, + &topic, + Some(PARTITION_ID), + 0, + ) + .await + .expect("deleting below the limit releases a slot"); +} diff --cc core/server/src/consumer_group.rs index 1a6beed9b,d4b77eb95..11ec19999 --- a/core/server/src/consumer_group.rs +++ b/core/server/src/consumer_group.rs @@@ -232,15 -231,11 +232,23 @@@ wher macro_rules! rewrite_group_offset { ($ty:ty) => {{ let mut wire = <$ty>::decode_from(body).map_err(|_| IggyError::InvalidCommand)?; + if wire.consumer.kind != KIND_CONSUMER_GROUP { + return Ok(request); + } - let group_id = resolve_group_offset_id( - shard, - &wire.consumer, - (&wire.stream_id, &wire.topic_id), - ) - .ok_or_else(|| missing_consumer_group_error(&wire.consumer.id, &wire.topic_id))?; + let Some(group_id) = + resolve_group_offset_id(shard, &wire.consumer, (&wire.stream_id, &wire.topic_id)) + else { - return Ok(request); ++ // An unknown stream or topic is not a missing group: let the ++ // namespace resolution below answer it with the same ++ // not-found every other partition op reports. ++ if !topic_exists(shard, &wire.stream_id, &wire.topic_id) { ++ return Ok(request); ++ } ++ return Err(missing_consumer_group_error( ++ &wire.consumer.id, ++ &wire.topic_id, ++ )); + }; // The partition-plane group-offset key is u32 (see the documented // ceiling on `Topic::next_consumer_group_id`). Clamp on the // ~4-billion-creates overflow rather than panic this live @@@ -260,7 -255,9 +268,28 @@@ rewrite_request_body(&request, &rewritten) } -/// Resolve the monotonic group id for a group consumer-offset op, or `None` for -/// an individual consumer (kind != 2) / unresolved group (leave the body as-is; -/// the apply / read path handle the miss). ++fn topic_exists<B, MJ, S, SB>( ++ shard: &Rc<ShellShard<B, MJ, S, SB>>, ++ stream_id: &WireIdentifier, ++ topic_id: &WireIdentifier, ++) -> bool ++where ++ B: ShellBus, ++ MJ: JournalHandle + 'static, ++ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>, ++ S: 'static, ++ SB: SuperblockStore + 'static, ++{ ++ shard ++ .plane ++ .metadata() ++ .mux_stm ++ .streams() ++ .topic_partitions_count(stream_id, topic_id) ++ .is_some() ++} ++ +/// Resolve the monotonic group id for a group consumer-offset op. fn resolve_group_offset_id<B, MJ, S, SB>( shard: &Rc<ShellShard<B, MJ, S, SB>>, consumer: &WireConsumer, diff --cc core/server/src/responses.rs index 2b9b524a5,6cea58642..f19002c00 --- a/core/server/src/responses.rs +++ b/core/server/src/responses.rs @@@ -292,28 -295,10 +292,35 @@@ wher false, ) .map(|_| ()) - .ok_or(IggyError::ConsumerGroupPartitionNotOwned( - client_id as u32, - partition_id, - )) + .ok_or_else(|| { + if streams + .resolve_consumer_group_id(stream_id, topic_id, &consumer.id) - .is_none() ++ .is_some() ++ { ++ IggyError::ConsumerGroupPartitionNotOwned(client_id as u32, partition_id) ++ } else if streams ++ .topic_partitions_count(stream_id, topic_id) ++ .is_some() + { + missing_consumer_group_error(&consumer.id, topic_id) + } else { - IggyError::ConsumerGroupPartitionNotOwned(client_id as u32, partition_id) ++ // An unknown stream or topic is not a missing group: report ++ // the same not-found every other partition op gives. ++ IggyError::ResourceNotFound(String::new()) + } + }) +} + +pub fn missing_consumer_group_error(group: &WireIdentifier, topic: &WireIdentifier) -> IggyError { + let topic = wire_identifier_for_display(topic); + match group { + WireIdentifier::Numeric(_) => { + IggyError::ConsumerGroupIdNotFound(wire_identifier_for_display(group), topic) + } + WireIdentifier::String(name) => { + IggyError::ConsumerGroupNameNotFound(name.as_str().to_owned(), topic) + } + } } /// Fence a consumer-group offset op then resolve its target partition diff --cc foreign/go/errors/errors_test.go index cacb8f90b,cacb8f90b..815fe404c --- a/foreign/go/errors/errors_test.go +++ b/foreign/go/errors/errors_test.go @@@ -95,6 -95,6 +95,25 @@@ func TestIggyError_ConsensusErrors(t *t } } ++func TestIggyError_TooManyConsumerOffsets(t *testing.T) { ++ err := TooManyConsumerOffsets{} ++ if err.Code() != TooManyConsumerOffsetsCode { ++ t.Errorf("Code() = %v, want %v", err.Code(), TooManyConsumerOffsetsCode) ++ } ++ if err.Error() != "too many consumer offsets for partition" { ++ t.Errorf("Error() = %q", err.Error()) ++ } ++ if !errors.Is(err, ErrTooManyConsumerOffsets) { ++ t.Errorf("errors.Is(%v, ErrTooManyConsumerOffsets) = false", err) ++ } ++ if errors.Is(err, ErrConsumerOffsetNotFound) { ++ t.Errorf("errors.Is(%v, ErrConsumerOffsetNotFound) = true", err) ++ } ++ if resolved := FromCode(TooManyConsumerOffsetsCode); !errors.Is(resolved, ErrTooManyConsumerOffsets) { ++ t.Errorf("FromCode(%d) = %v", TooManyConsumerOffsetsCode, resolved) ++ } ++} ++ func TestFromCode_FallsBackToTheGenericError(t *testing.T) { if resolved := FromCode(Code(0xFFFFFF)); !errors.Is(resolved, ErrError) { t.Errorf("FromCode(unknown) = %v, want ErrError", resolved)
