This is an automated email from the ASF dual-hosted git repository.
hubcio 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 84d3b6766 feat(sdk): read offsets of consumer group members over the
HTTP API (#4348)
84d3b6766 is described below
commit 84d3b67660d8155bf282d1311f2c46004e5b7c42
Author: haubur <[email protected]>
AuthorDate: Mon Oct 12 19:35:06 2026 +0200
feat(sdk): read offsets of consumer group members over the HTTP API (#4348)
---
Cargo.lock | 1 +
Cargo.toml | 1 +
.../consumer_offsets/delete_consumer_offset.rs | 3 +
core/common/src/types/consumer/consumer_kind.rs | 7 +-
core/integration/tests/sdk/consumer_offset.rs | 297 ++++++++++++++++++++-
core/integration/tests/server/mod.rs | 2 +-
.../tests/server/scenarios/permissions_scenario.rs | 2 +-
core/sdk/Cargo.toml | 1 +
core/sdk/src/http/consumer_groups.rs | 9 +-
core/sdk/src/http/consumer_offsets.rs | 108 ++------
core/sdk/src/http/messages.rs | 14 +-
core/sdk/src/http/mod.rs | 1 +
core/sdk/src/http/partitions.rs | 5 +-
core/sdk/src/http/path.rs | 46 ++++
core/sdk/src/http/segments.rs | 5 +-
core/sdk/src/http/streams.rs | 3 +-
core/sdk/src/http/topics.rs | 5 +-
core/server/src/dispatch/failure.rs | 7 +-
core/server/src/http/handlers.rs | 86 +++---
core/server/src/http/wire.rs | 171 +++++++++---
20 files changed, 594 insertions(+), 180 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 785e2b02c..38bd403ca 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6934,6 +6934,7 @@ dependencies = [
"iggy_binary_protocol",
"iggy_common",
"mockall",
+ "percent-encoding",
"quinn",
"rand 0.10.2",
"rcgen",
diff --git a/Cargo.toml b/Cargo.toml
index e868a9421..6386a3dda 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -275,6 +275,7 @@ parquet = "58.4.0"
partitions = { path = "core/partitions" }
passterm = "2.0.6"
paste = "1.0"
+percent-encoding = "2.3.2"
pgwire = "0.40.7"
postcard = { version = "1.1.3", features = ["alloc"] }
predicates = "3.1.4"
diff --git a/core/common/src/http/consumer_offsets/delete_consumer_offset.rs
b/core/common/src/http/consumer_offsets/delete_consumer_offset.rs
index b97c9a7f4..b10eada36 100644
--- a/core/common/src/http/consumer_offsets/delete_consumer_offset.rs
+++ b/core/common/src/http/consumer_offsets/delete_consumer_offset.rs
@@ -15,10 +15,13 @@
// specific language governing permissions and limitations
// under the License.
+use crate::ConsumerKind;
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize)]
pub struct DeleteConsumerOffset {
#[serde(default)]
+ pub consumer_kind: ConsumerKind,
+ #[serde(default, skip_serializing_if = "Option::is_none")]
pub partition_id: Option<u32>,
}
diff --git a/core/common/src/types/consumer/consumer_kind.rs
b/core/common/src/types/consumer/consumer_kind.rs
index 391211903..bcd70aa36 100644
--- a/core/common/src/types/consumer/consumer_kind.rs
+++ b/core/common/src/types/consumer/consumer_kind.rs
@@ -30,8 +30,8 @@ use std::str::FromStr;
/// - `id`: the unique identifier of the consumer.
#[derive(Debug, Serialize, Deserialize, PartialEq, Default, Clone)]
pub struct Consumer {
- /// The type of consumer.
- #[serde(skip)]
+ /// The type of consumer. It can be either `Consumer`, `ConsumerGroup` or
`ExternalGroup`.
+ #[serde(rename = "consumer_kind", default)]
pub kind: ConsumerKind,
/// The unique identifier of the consumer.
#[serde(rename = "consumer_id")]
@@ -96,8 +96,7 @@ impl Consumer {
///
/// For offset calls only: no membership check, no range check, never
polled, and no hold on
/// retention. Deleting the group deletes its offsets. A poll with it
returns
- /// [`IggyError::FeatureUnavailable`], and so does every call over HTTP,
because the REST API
- /// cannot name a consumer kind.
+ /// [`IggyError::FeatureUnavailable`].
pub fn external_group(id: Identifier) -> Self {
Self {
kind: ConsumerKind::ExternalGroup,
diff --git a/core/integration/tests/sdk/consumer_offset.rs
b/core/integration/tests/sdk/consumer_offset.rs
index b12d2b4a3..6b49f6365 100644
--- a/core/integration/tests/sdk/consumer_offset.rs
+++ b/core/integration/tests/sdk/consumer_offset.rs
@@ -20,14 +20,295 @@ use std::str::FromStr;
use futures::StreamExt;
use iggy::prelude::*;
use iggy::stream_builder::{IggyConsumerConfig, IggyStreamConsumer};
+use integration::harness::wait_for_consumer_group_assignment;
use integration::iggy_harness;
use tokio::time::{Duration, timeout};
-const STREAM_NAME: &str = "delete-offset-stream";
-const TOPIC_NAME: &str = "delete-offset-topic";
-const CONSUMER_NAME: &str = "delete-offset-consumer";
+const STREAM_NAME: &str = "test-offset-stream";
+const TOPIC_NAME: &str = "test-offset-topic";
+const CONSUMER_NAME: &str = "test-offset-consumer";
+const CONSUMER_GROUP_NAME: &str = "test-offset-group";
+const OFFSET_PARTITION_ID: u32 = 0;
const ASSIGNED_PARTITION_ID: u32 = 3;
const POLL_TIMEOUT: Duration = Duration::from_secs(10);
+const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[iggy_harness(test_client_transport = [Http, Tcp, Quic, WebSocket])]
+async fn
given_consumer_and_group_offsets_when_managed_over_any_transport_should_store_read_and_delete(
+ harness: &TestHarness,
+) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+
+ // A consumer-group member needs a persistent connection. Required
+ // for offset test cases using a consumer group.
+ let seed_client = harness
+ .tcp_root_client()
+ .await
+ .expect("Failed to get tcp seed 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();
+
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(&stream_id, TOPIC_NAME, &TopicCreateOptions::default())
+ .await
+ .unwrap();
+
+ let mut messages = vec![
+ IggyMessage::from_str("message_1").unwrap(),
+ IggyMessage::from_str("message_2").unwrap(),
+ IggyMessage::from_str("message_3").unwrap(),
+ IggyMessage::from_str("message_4").unwrap(),
+ IggyMessage::from_str("message_5").unwrap(),
+ ];
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(OFFSET_PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+
+ // Setup a consumer that joins a group in order to store, read and delete
+ // offsets from a group member.
+ seed_client
+ .create_consumer_group(&stream_id, &topic_id, CONSUMER_GROUP_NAME)
+ .await
+ .unwrap();
+
+ seed_client
+ .join_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .unwrap();
+
+ wait_for_consumer_group_assignment(
+ &seed_client,
+ &stream_id,
+ &topic_id,
+ &group_id,
+ 1,
+ RESOLVE_TIMEOUT,
+ )
+ .await;
+
+ let standalone_consumer =
Consumer::new(Identifier::named(CONSUMER_NAME).unwrap());
+ let group_consumer = Consumer::group(group_id.clone());
+
+ // The group consumer must commit and delete through `seed_client`, the
+ // connection that actually joined the group: the server fences a
+ // group-offset write to the calling connection's own membership.
+ for (consumer, offset, owner) in [
+ (&standalone_consumer, 2_u64, &client),
+ (&group_consumer, 4_u64, &seed_client),
+ ] {
+ owner
+ .store_consumer_offset(
+ consumer,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ offset,
+ )
+ .await
+ .unwrap();
+
+ let stored_offset = client
+ .get_consumer_offset(consumer, &stream_id, &topic_id,
Some(OFFSET_PARTITION_ID))
+ .await
+ .unwrap()
+ .map(|info| info.stored_offset);
+
+ assert_eq!(
+ stored_offset,
+ Some(offset),
+ "{consumer:?} stored offset should be readable over every
transport"
+ );
+
+ owner
+ .delete_consumer_offset(consumer, &stream_id, &topic_id,
Some(OFFSET_PARTITION_ID))
+ .await
+ .unwrap();
+
+ let stored_offset_after_delete = client
+ .get_consumer_offset(consumer, &stream_id, &topic_id,
Some(OFFSET_PARTITION_ID))
+ .await
+ .unwrap()
+ .map(|info| info.stored_offset);
+
+ assert_eq!(
+ stored_offset_after_delete, None,
+ "{consumer:?} stored offset should be gone after delete over every
transport"
+ );
+
+ // HTTP has no persistent connection to fence a group-offset write to,
+ // so store/delete/poll should reject consumer_kind=consumer_group.
+ if consumer.kind == ConsumerKind::ConsumerGroup
+ && harness.transport().expect("harness transport") ==
TransportProtocol::Http
+ {
+ let store_error = client
+ .store_consumer_offset(
+ consumer,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ offset,
+ )
+ .await
+ .expect_err("HTTP store must reject a consumer-group kind");
+ assert!(
+ is_feature_unavailable(&store_error),
+ "expected a feature-unavailable rejection for HTTP store with
consumer_kind=consumer_group, got {store_error:?}"
+ );
+
+ let delete_error = client
+ .delete_consumer_offset(consumer, &stream_id, &topic_id,
Some(OFFSET_PARTITION_ID))
+ .await
+ .expect_err("HTTP delete must reject a consumer-group kind");
+ assert!(
+ is_feature_unavailable(&delete_error),
+ "expected a feature-unavailable rejection for HTTP delete with
consumer_kind=consumer_group, got {delete_error:?}"
+ );
+
+ let poll_error = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ consumer,
+ &PollingStrategy::first(),
+ 1,
+ false,
+ )
+ .await
+ .expect_err("HTTP poll must reject a consumer-group kind");
+ assert!(
+ is_feature_unavailable(&poll_error),
+ "expected a feature-unavailable rejection for HTTP poll with
consumer_kind=consumer_group, got {poll_error:?}"
+ );
+ }
+ }
+}
+
+#[iggy_harness(test_client_transport = [Http])]
+async fn
given_external_group_offset_when_managed_over_http_should_store_read_delete_and_reject_poll(
+ harness: &TestHarness,
+) {
+ let client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+
+ let stream_id = Identifier::named(STREAM_NAME).unwrap();
+ let topic_id = Identifier::named(TOPIC_NAME).unwrap();
+
+ client.create_stream(STREAM_NAME).await.unwrap();
+ client
+ .create_topic(&stream_id, TOPIC_NAME, &TopicCreateOptions::default())
+ .await
+ .unwrap();
+
+ let mut messages = vec![
+ IggyMessage::from_str("message_1").unwrap(),
+ IggyMessage::from_str("message_2").unwrap(),
+ IggyMessage::from_str("message_3").unwrap(),
+ IggyMessage::from_str("message_4").unwrap(),
+ IggyMessage::from_str("message_5").unwrap(),
+ ];
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(OFFSET_PARTITION_ID),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+
+ let group = client
+ .create_consumer_group(&stream_id, &topic_id, CONSUMER_GROUP_NAME)
+ .await
+ .unwrap();
+ let external_group =
+ Consumer::external_group(Identifier::numeric(group.id).expect("valid
group id"));
+
+ client
+ .store_consumer_offset(
+ &external_group,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ 3,
+ )
+ .await
+ .unwrap();
+
+ let stored_offset = client
+ .get_consumer_offset(
+ &external_group,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ )
+ .await
+ .unwrap()
+ .map(|info| info.stored_offset);
+ assert_eq!(
+ stored_offset,
+ Some(3),
+ "external group stored offset should be readable over HTTP"
+ );
+
+ client
+ .delete_consumer_offset(
+ &external_group,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ )
+ .await
+ .unwrap();
+
+ let stored_offset_after_delete = client
+ .get_consumer_offset(
+ &external_group,
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ )
+ .await
+ .unwrap()
+ .map(|info| info.stored_offset);
+ assert_eq!(
+ stored_offset_after_delete, None,
+ "external group stored offset should be gone after delete over HTTP"
+ );
+
+ let poll_error = client
+ .poll_messages(
+ &stream_id,
+ &topic_id,
+ Some(OFFSET_PARTITION_ID),
+ &external_group,
+ &PollingStrategy::first(),
+ 1,
+ false,
+ )
+ .await
+ .expect_err("HTTP poll must reject an external-group kind");
+ assert!(
+ matches!(poll_error, IggyError::FeatureUnavailable),
+ "expected FeatureUnavailable for HTTP poll with
consumer_kind=external_group, got {poll_error:?}"
+ );
+}
#[iggy_harness]
async fn standalone_consumer_deletes_partition_zero_on_none_in_delete_offset(
@@ -178,6 +459,16 @@ async fn
given_standalone_stream_consumer_when_creating_topic_should_select_part
}
}
+/// True when an HTTP error body wraps the `FeatureUnavailable` error code
into a 400.
+fn is_feature_unavailable(error: &IggyError) -> bool {
+ let IggyError::HttpResponseError(_, body) = error else {
+ return false;
+ };
+
+ crate::server::scenarios::permissions_scenario::http_error_id(body)
+ == Some(IggyError::FeatureUnavailable.as_code())
+}
+
/// Read the offset the server has stored for the consumer.
async fn fetch_stored_offset(
client: &IggyClient,
diff --git a/core/integration/tests/server/mod.rs
b/core/integration/tests/server/mod.rs
index 29f5bec54..ade46c258 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -92,6 +92,6 @@ mod message_retrieval;
// The full restart matrix (consumer variants included) runs under the server:
// a restarted replica rejoins via the view probe + journal repair.
mod delete_segments;
-mod scenarios;
+pub(crate) mod scenarios;
mod specific;
mod telemetry;
diff --git a/core/integration/tests/server/scenarios/permissions_scenario.rs
b/core/integration/tests/server/scenarios/permissions_scenario.rs
index 000b2aefd..8143f50fc 100644
--- a/core/integration/tests/server/scenarios/permissions_scenario.rs
+++ b/core/integration/tests/server/scenarios/permissions_scenario.rs
@@ -2894,7 +2894,7 @@ fn assert_error_code<T: std::fmt::Debug>(
/// The `id` field of an HTTP `ErrorResponse` body is the wire `IggyError`
code,
/// so a status-collapsed `HttpResponseError` can still be matched exactly.
-fn http_error_id(body: &str) -> Option<u32> {
+pub(crate) fn http_error_id(body: &str) -> Option<u32> {
let value: serde_json::Value = serde_json::from_str(body).ok()?;
value.get("id")?.as_u64().map(|id| id as u32)
}
diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index daf208661..4361c7caf 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml
@@ -42,6 +42,7 @@ futures = { workspace = true }
futures-util = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
+percent-encoding = { workspace = true }
quinn = { workspace = true }
rand = { workspace = true }
reqwest = { workspace = true }
diff --git a/core/sdk/src/http/consumer_groups.rs
b/core/sdk/src/http/consumer_groups.rs
index ab7659505..dc3e8110a 100644
--- a/core/sdk/src/http/consumer_groups.rs
+++ b/core/sdk/src/http/consumer_groups.rs
@@ -17,6 +17,7 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::IggyError;
use async_trait::async_trait;
use iggy_common::ConsumerGroupClient;
@@ -36,7 +37,7 @@ impl ConsumerGroupClient for HttpClient {
.get(&format!(
"{}/{}",
get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
- group_id
+ encode_segment(&group_id.as_cow_str())
))
.await;
if let Err(error) = response {
@@ -101,7 +102,7 @@ impl ConsumerGroupClient for HttpClient {
let path = format!(
"{}/{}",
get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
- group_id.as_cow_str()
+ encode_segment(&group_id.as_cow_str())
);
self.delete(&path).await?;
Ok(())
@@ -127,5 +128,7 @@ impl ConsumerGroupClient for HttpClient {
}
fn get_path(stream_id: &str, topic_id: &str) -> String {
- format!("streams/{stream_id}/topics/{topic_id}/consumer-groups")
+ let encoded_stream = encode_segment(stream_id);
+ let encoded_topic = encode_segment(topic_id);
+ format!("streams/{encoded_stream}/topics/{encoded_topic}/consumer-groups")
}
diff --git a/core/sdk/src/http/consumer_offsets.rs
b/core/sdk/src/http/consumer_offsets.rs
index f7a00625b..fbeb9dd56 100644
--- a/core/sdk/src/http/consumer_offsets.rs
+++ b/core/sdk/src/http/consumer_offsets.rs
@@ -17,13 +17,15 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::Identifier;
use crate::prelude::IggyError;
use async_trait::async_trait;
use iggy_common::ConsumerOffsetClient;
+use iggy_common::delete_consumer_offset::DeleteConsumerOffset;
use iggy_common::get_consumer_offset::GetConsumerOffset;
use iggy_common::store_consumer_offset::StoreConsumerOffset;
-use iggy_common::{Consumer, ConsumerKind, ConsumerOffsetInfo,
ConsumerPosition};
+use iggy_common::{Consumer, ConsumerOffsetInfo, ConsumerPosition};
#[async_trait]
impl ConsumerOffsetClient for HttpClient {
@@ -34,7 +36,6 @@ impl ConsumerOffsetClient for HttpClient {
topic_id: &Identifier,
position: ConsumerPosition,
) -> Result<(), IggyError> {
- refuse_external_group(consumer)?;
self.put(
&get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
&StoreConsumerOffset {
@@ -56,7 +57,6 @@ impl ConsumerOffsetClient for HttpClient {
partition_id: Option<u32>,
offset: u64,
) -> Result<(), IggyError> {
- refuse_external_group(consumer)?;
self.put(
&get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
&StoreConsumerOffset {
@@ -77,7 +77,6 @@ impl ConsumerOffsetClient for HttpClient {
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<Option<ConsumerOffsetInfo>, IggyError> {
- refuse_external_group(consumer)?;
let response = self
.get_with_query(
&get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
@@ -109,102 +108,47 @@ impl ConsumerOffsetClient for HttpClient {
topic_id: &Identifier,
partition_id: Option<u32>,
) -> Result<(), IggyError> {
- refuse_external_group(consumer)?;
- let partition_id = partition_id
- .map(|id| format!("?partition_id={id}"))
- .unwrap_or_default();
let path = format!(
- "{}/{}{partition_id}",
+ "{}/{}",
get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
- consumer.id
+ encode_segment(&consumer.id.as_cow_str())
);
- self.delete(&path).await?;
+ self.delete_with_query(
+ &path,
+ &DeleteConsumerOffset {
+ consumer_kind: consumer.kind,
+ partition_id,
+ },
+ )
+ .await?;
Ok(())
}
}
fn get_path(stream_id: &str, topic_id: &str) -> String {
- format!("streams/{stream_id}/topics/{topic_id}/consumer-offsets")
-}
-
-/// The REST API cannot name a consumer kind, so an external group would land
on a plain
-/// consumer's offset.
-pub(crate) fn refuse_external_group(consumer: &Consumer) -> Result<(),
IggyError> {
- if consumer.kind == ConsumerKind::ExternalGroup {
- return Err(IggyError::FeatureUnavailable);
- }
- Ok(())
+ let encoded_stream = encode_segment(stream_id);
+ let encoded_topic = encode_segment(topic_id);
+ format!("streams/{encoded_stream}/topics/{encoded_topic}/consumer-offsets")
}
#[cfg(test)]
mod tests {
use super::*;
- use iggy_common::{MessageClient, PartitionContext, PollingStrategy};
-
- /// Nothing listens on this port, so a call that reached the network would
fail with a
- /// transport error instead.
- const UNREACHABLE_API: &str = "http://127.0.0.1:1";;
- #[tokio::test]
- async fn
given_external_group_when_calling_over_http_should_refuse_before_sending() {
- let client = HttpClient::new(UNREACHABLE_API).unwrap();
- let group = Consumer::external_group(Identifier::numeric(1).unwrap());
- let stream = Identifier::numeric(1).unwrap();
- let topic = Identifier::numeric(1).unwrap();
+ #[test]
+ fn given_plain_ids_when_building_path_should_join_segments() {
+ let path = get_path("1", "orders");
- assert!(matches!(
- client
- .store_consumer_offset(&group, &stream, &topic, Some(0), 5)
- .await,
- Err(IggyError::FeatureUnavailable)
- ));
- assert!(matches!(
- client
- .store_consumer_position(
- &group,
- &stream,
- &topic,
- ConsumerPosition {
- partition_id: 0,
- offset: 5,
- context: PartitionContext::default(),
- },
- )
- .await,
- Err(IggyError::FeatureUnavailable)
- ));
- assert!(matches!(
- client
- .get_consumer_offset(&group, &stream, &topic, Some(0))
- .await,
- Err(IggyError::FeatureUnavailable)
- ));
- assert!(matches!(
- client
- .delete_consumer_offset(&group, &stream, &topic, Some(0))
- .await,
- Err(IggyError::FeatureUnavailable)
- ));
- assert!(matches!(
- client
- .poll_messages(
- &stream,
- &topic,
- Some(0),
- &group,
- &PollingStrategy::next(),
- 1,
- false
- )
- .await,
- Err(IggyError::FeatureUnavailable)
- ));
+ assert_eq!(path, "streams/1/topics/orders/consumer-offsets");
}
#[test]
- fn given_plain_consumer_when_guarded_should_pass() {
- let consumer = Consumer::new(Identifier::numeric(1).unwrap());
+ fn given_reserved_characters_when_building_path_should_percent_encode() {
+ let path = get_path("my stream", "my/topic");
- assert!(refuse_external_group(&consumer).is_ok());
+ assert_eq!(
+ path,
+ "streams/my%20stream/topics/my%2Ftopic/consumer-offsets"
+ );
}
}
diff --git a/core/sdk/src/http/messages.rs b/core/sdk/src/http/messages.rs
index c33a685eb..3bf74d13e 100644
--- a/core/sdk/src/http/messages.rs
+++ b/core/sdk/src/http/messages.rs
@@ -17,9 +17,10 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::{
- Consumer, Identifier, IggyError, IggyMessage, PartitionContext,
Partitioning, PollMessages,
- PolledMessages, PollingStrategy, SendMessages, SendMessagesResponse,
+ Consumer, ConsumerKind, Identifier, IggyError, IggyMessage,
PartitionContext, Partitioning,
+ PollMessages, PolledMessages, PollingStrategy, SendMessages,
SendMessagesResponse,
};
use async_trait::async_trait;
use iggy_common::IggyMessagesBatch;
@@ -39,7 +40,10 @@ impl MessageClient for HttpClient {
count: u32,
auto_commit: bool,
) -> Result<PolledMessages, IggyError> {
- crate::http::consumer_offsets::refuse_external_group(consumer)?;
+ if matches!(consumer.kind, ConsumerKind::ExternalGroup) {
+ return Err(IggyError::FeatureUnavailable);
+ }
+
let response = self
.get_with_query(
&get_path(&stream_id.as_cow_str(), &topic_id.as_cow_str()),
@@ -215,7 +219,9 @@ async fn decode_send_response(
}
fn get_path(stream_id: &str, topic_id: &str) -> String {
- format!("streams/{stream_id}/topics/{topic_id}/messages")
+ let encoded_stream = encode_segment(stream_id);
+ let encoded_topic = encode_segment(topic_id);
+ format!("streams/{encoded_stream}/topics/{encoded_topic}/messages")
}
#[cfg(test)]
diff --git a/core/sdk/src/http/mod.rs b/core/sdk/src/http/mod.rs
index 7f5dc1504..8ef3399e5 100644
--- a/core/sdk/src/http/mod.rs
+++ b/core/sdk/src/http/mod.rs
@@ -22,6 +22,7 @@ pub mod http_client;
pub(crate) mod http_transport;
pub mod messages;
pub mod partitions;
+pub(crate) mod path;
pub mod personal_access_tokens;
#[cfg(test)]
pub(crate) mod scripted_server;
diff --git a/core/sdk/src/http/partitions.rs b/core/sdk/src/http/partitions.rs
index 7a2bd7dd6..e216003f5 100644
--- a/core/sdk/src/http/partitions.rs
+++ b/core/sdk/src/http/partitions.rs
@@ -17,6 +17,7 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::{Identifier, IggyError};
use async_trait::async_trait;
use iggy_common::PartitionClient;
@@ -65,5 +66,7 @@ impl PartitionClient for HttpClient {
}
fn get_path(stream_id: &str, topic_id: &str) -> String {
- format!("streams/{stream_id}/topics/{topic_id}/partitions")
+ let encoded_stream = encode_segment(stream_id);
+ let encoded_topic = encode_segment(topic_id);
+ format!("streams/{encoded_stream}/topics/{encoded_topic}/partitions")
}
diff --git a/core/sdk/src/http/path.rs b/core/sdk/src/http/path.rs
new file mode 100644
index 000000000..cf3209c70
--- /dev/null
+++ b/core/sdk/src/http/path.rs
@@ -0,0 +1,46 @@
+// 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 percent_encoding::{AsciiSet, NON_ALPHANUMERIC, PercentEncode,
utf8_percent_encode};
+
+const PATH_SEGMENT: &AsciiSet = NON_ALPHANUMERIC;
+
+pub(crate) fn encode_segment(value: &str) -> PercentEncode<'_> {
+ utf8_percent_encode(value, PATH_SEGMENT)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn given_plain_id_when_encoding_should_pass_through() {
+ assert_eq!(encode_segment("1").to_string(), "1");
+ }
+
+ #[test]
+ fn given_reserved_characters_when_encoding_should_percent_encode() {
+ assert_eq!(encode_segment("my stream").to_string(), "my%20stream");
+ assert_eq!(encode_segment("my/topic").to_string(), "my%2Ftopic");
+ }
+
+ #[test]
+ fn given_dot_segments_when_encoding_should_never_pass_through_literally() {
+ assert_eq!(encode_segment(".").to_string(), "%2E");
+ assert_eq!(encode_segment("..").to_string(), "%2E%2E");
+ }
+}
diff --git a/core/sdk/src/http/segments.rs b/core/sdk/src/http/segments.rs
index 9a64c37db..4a2dfb40e 100644
--- a/core/sdk/src/http/segments.rs
+++ b/core/sdk/src/http/segments.rs
@@ -17,6 +17,7 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::{Identifier, IggyError};
use async_trait::async_trait;
use iggy_common::SegmentClient;
@@ -50,5 +51,7 @@ impl SegmentClient for HttpClient {
}
fn get_path(stream_id: &str, topic_id: &str, partition_id: u32) -> String {
- format!("streams/{stream_id}/topics/{topic_id}/partitions/{partition_id}")
+ let encoded_stream = encode_segment(stream_id);
+ let encoded_topic = encode_segment(topic_id);
+
format!("streams/{encoded_stream}/topics/{encoded_topic}/partitions/{partition_id}")
}
diff --git a/core/sdk/src/http/streams.rs b/core/sdk/src/http/streams.rs
index 65c200a27..7c149640d 100644
--- a/core/sdk/src/http/streams.rs
+++ b/core/sdk/src/http/streams.rs
@@ -17,6 +17,7 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::Identifier;
use crate::prelude::IggyError;
use async_trait::async_trait;
@@ -99,5 +100,5 @@ impl StreamClient for HttpClient {
}
fn get_details_path(stream_id: &str) -> String {
- format!("{PATH}/{stream_id}")
+ format!("{PATH}/{}", encode_segment(stream_id))
}
diff --git a/core/sdk/src/http/topics.rs b/core/sdk/src/http/topics.rs
index 62fba3924..f154903a2 100644
--- a/core/sdk/src/http/topics.rs
+++ b/core/sdk/src/http/topics.rs
@@ -17,6 +17,7 @@
use crate::http::http_client::HttpClient;
use crate::http::http_transport::HttpTransport;
+use crate::http::path::encode_segment;
use crate::prelude::{Identifier, IggyError, IggyExpiry, MaxTopicSize};
use async_trait::async_trait;
use iggy_common::TopicClient;
@@ -143,9 +144,9 @@ impl HttpClient {
}
fn get_path(stream_id: &str) -> String {
- format!("streams/{stream_id}/topics")
+ format!("streams/{}/topics", encode_segment(stream_id))
}
fn get_details_path(stream_id: &str, topic_id: &str) -> String {
- format!("{}/{topic_id}", get_path(stream_id))
+ format!("{}/{}", get_path(stream_id), encode_segment(topic_id))
}
diff --git a/core/server/src/dispatch/failure.rs
b/core/server/src/dispatch/failure.rs
index 1bff997e4..d2e70c6fa 100644
--- a/core/server/src/dispatch/failure.rs
+++ b/core/server/src/dispatch/failure.rs
@@ -827,10 +827,9 @@ mod tests {
);
}
- /// The HTTP mirror (`resync_required_polled_messages` in
- /// `crate::http::wire`) is a JSON DTO, not a wire frame, so the shared
- /// contract asserted here is the sentinel constant itself; the DTO's own
- /// test pins its `partition_id` to the same constant.
+ /// TCP is the only transport that still carries this sentinel. HTTP
+ /// rejects `consumer_kind=consumer_group` polls before a poll ever
+ /// reaches this path (`crate::http::wire::poll_wire_request`).
#[compio::test]
async fn
resync_sentinel_must_ride_a_status_zero_poll_body_led_by_the_sentinel() {
let (bus, shard) = shard_at_commit();
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index a4105cdcb..15adb538f 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -142,7 +142,7 @@ use crate::http::submit::{
};
use crate::http::wire::{
consumer_offset_wire_request, delete_offset_wire_request,
encode_send_messages,
- poll_wire_request, resync_required_polled_messages,
store_offset_wire_request,
+ poll_wire_request, store_offset_wire_request,
};
use crate::reply_frame::{build_polled_messages_body, build_raw_pat_reply};
use crate::responses::connected_client_to_response;
@@ -157,11 +157,11 @@ const PONG: &str = "pong";
/// VSR client id stamped on HTTP data-plane reads (poll / consumer-offset).
/// HTTP reads never Register a VSR client, and shard-0 client ids are minted
-/// from 1, so 0 can never name a live consumer-group member: a group-kind
-/// poll fences closed with `ConsumerGroupPartitionNotOwned` and answers the
-/// re-sync sentinel empty poll - the same failure a stale TCP member sees.
-/// Legacy HTTP polls with client id 0 for the same reason (no persistent
-/// sessions, so no group membership).
+/// from 1, so 0 can never collide with a live client. `resolve_poll_request`
+/// takes a client id only for its consumer-group fencing branch, and HTTP
+/// never reaches it: `poll_wire_request` rejects
`consumer_kind=consumer_group`
+/// up front. Legacy HTTP polls with client id 0 for the same reason (no
+/// persistent sessions, so no group membership).
const HTTP_READ_CLIENT_ID: u128 = 0;
/// Response header attesting what durability a produce response proves:
@@ -1198,7 +1198,10 @@ pub(in crate::http) async fn delete_segments(
/// messages as the same `PolledMessages` JSON the legacy server returns. The
/// query is the same flattened `PollMessages` shape the legacy server accepts
/// (`consumer_id`, `partition_id`, strategy `kind`+`value`, `count`,
-/// `auto_commit`); stream and topic come from the path.
+/// `auto_commit`); stream and topic come from the path. `consumer_kind`
+/// defaults to `consumer` and `consumer_group` is rejected (see
+/// [`poll_wire_request`]) since HTTP has no persistent connection and cannot
observe
+/// group's partition assignment (that can be subject to change).
///
/// A non-replicated read served in band: the same resolution the TCP dispatch
/// runs ([`resolve_poll_request`]), then a mesh read on the owning shard and a
@@ -1228,20 +1231,21 @@ pub(in crate::http) async fn poll_messages(
},
))
.await?;
- let wire = poll_wire_request(&stream_id, &topic_id,
&query).map_err(ReadError::Rejected)?;
+
+ let wire = match poll_wire_request(&stream_id, &topic_id, &query) {
+ Ok(request) => request,
+ Err(error) => return Err(ReadError::Rejected(error)),
+ };
+
let (namespace, partition_id, consumer, args) =
match resolve_poll_request(&state.shard, &wire, HTTP_READ_CLIENT_ID) {
Ok(decoded) => decoded,
- // TCP parity: a fenced group poll answers 200 with the re-sync
- // sentinel partition, not an error (see [`HTTP_READ_CLIENT_ID`]).
- Err(IggyError::ConsumerGroupPartitionNotOwned(..)) => {
- return Ok(Json(resync_required_polled_messages()));
- }
- // A partition id the topic does not have is a client addressing
- // error with its own code; collapsing it into the generic 404 body
- // told an SDK "no such stream/topic" for a request whose stream
and
- // topic both resolved. TCP parity: the dispatch denies typed here.
- Err(error @ IggyError::PartitionNotFound(..)) => {
+ // A partition id the topic does not have, or a consumer kind HTTP
+ // can't poll with, is a client addressing error with its own code;
+ // collapsing it into the generic 404 body told an SDK "no such
+ // stream/topic" for a request whose stream and topic both
+ // resolved. TCP parity: the dispatch denies typed here.
+ Err(error @ (IggyError::PartitionNotFound(..) |
IggyError::FeatureUnavailable)) => {
return Err(ReadError::Rejected(error));
}
// The remaining resolver failures are STM lookups that came up
@@ -1299,13 +1303,18 @@ pub(in crate::http) async fn poll_messages(
/// `GET /streams/{stream_id}/topics/{topic_id}/consumer-offsets`: fetch a
/// consumer's stored offset as the same `ConsumerOffsetInfo` JSON the legacy
-/// server returns. The query is the same flattened `GetConsumerOffset` shape
-/// the legacy server accepts (`consumer_id`, optional `partition_id`).
+/// server returns. The query is the flattened `GetConsumerOffset` shape
+/// (`consumer_id`, optional `consumer_kind` defaulting to `consumer`,
+/// optional `partition_id`).
///
/// A non-replicated read served in band, mirroring [`poll_messages`]. A
-/// missing offset is the legacy 404, which the SDK reads as `None`. An owner
-/// that does not hold the partition yet answers 503 `TransientNotAccepted`,
-/// as TCP does, so it never reads as `None`.
+/// missing offset (never stored, or the partition unknown to its owner) is
+/// the legacy 404: the TCP path replies an empty body the SDK decodes as
+/// `None`, and the legacy HTTP server renders that `None` as
+/// `CustomError::ResourceNotFound`. Unlike poll/store/delete,
+/// `consumer_kind=consumer_group` is allowed, because a group offset read
needs
+/// no held connection, so it is not member-fenced
+/// ([`resolve_consumer_offset_request`]).
pub(in crate::http) async fn get_consumer_offset(
State(state): State<HttpState>,
identity: Identity,
@@ -1455,9 +1464,11 @@ pub(in crate::http) async fn send_messages(
/// `PUT /streams/{stream_id}/topics/{topic_id}/consumer-offsets`: store a
/// consumer's offset. The JSON body is the same `StoreConsumerOffset` shape
the
-/// legacy server accepts (flattened `consumer_id`, optional `partition_id`,
-/// `offset`); stream and topic come from the path. Returns 204 on commit,
-/// matching the legacy server.
+/// legacy server accepts (flattened `consumer_id`, optional `consumer_kind`,
+/// `partition_id`, `offset`); stream and topic come from the path. Returns
+/// 204 on commit, matching the legacy server. `consumer_kind=consumer_group`
+/// is rejected ([`store_offset_wire_request`]) since HTTP has no persistent
+/// connection and group memberships can change.
///
/// Data plane like a produce: the offset write is a replicated op on the
/// partition group's own consensus, awaited through the session's in-process
@@ -1498,12 +1509,14 @@ pub(in crate::http) async fn store_consumer_offset(
}
/// `DELETE
/streams/{stream_id}/topics/{topic_id}/consumer-offsets/{consumer_id}`:
-/// delete a consumer's stored offset. The consumer comes from the path and the
-/// optional `partition_id` from the query, the same `DeleteConsumerOffset`
-/// shape the legacy server accepts. Returns 204 on commit, matching the legacy
-/// server; a delete of a never-stored offset is denied by the partition
-/// primary (`ReplyHeader.status`) and renders the legacy typed 404. Same
-/// replicated partition write as [`store_consumer_offset`].
+/// delete a consumer's stored offset. The consumer id comes from the path and
the
+/// optional `consumer_kind` (defaulting to `consumer`) and `partition_id` come
+/// from the query, the same `DeleteConsumerOffset` shape the legacy server
+/// accepts. Returns 204 on commit, matching the legacy server; a delete of a
+/// never-stored offset is denied by the partition primary
+/// (`ReplyHeader.status`) and renders the legacy typed 404. Same replicated
+/// partition write as [`store_consumer_offset`].
`consumer_kind=consumer_group`
+/// is rejected because HTTP does not persist a connection and group
memberships can change.
pub(in crate::http) async fn delete_consumer_offset(
State(state): State<HttpState>,
identity: Authenticated,
@@ -1522,11 +1535,10 @@ pub(in crate::http) async fn delete_consumer_offset(
Permissioner::delete_consumer_offset,
)
.map_err(PartitionWriteError::Rejected)?;
- // `Consumer::new` fixes the kind to `Consumer`, exactly as the legacy
- // handler does; HTTP cannot express a group-kind offset op.
- let consumer = Consumer::new(
-
Identifier::from_str_value(&consumer_id).map_err(PartitionWriteError::Rejected)?,
- );
+ let consumer = Consumer {
+ kind: query.consumer_kind,
+ id:
Identifier::from_str_value(&consumer_id).map_err(PartitionWriteError::Rejected)?,
+ };
let request = delete_offset_wire_request(&stream_id, &topic_id, &consumer,
query.partition_id)
.map_err(PartitionWriteError::Rejected)?;
let body = request.to_bytes();
diff --git a/core/server/src/http/wire.rs b/core/server/src/http/wire.rs
index 34b291759..83281dd6e 100644
--- a/core/server/src/http/wire.rs
+++ b/core/server/src/http/wire.rs
@@ -34,8 +34,7 @@ use iggy_common::poll_messages::DEFAULT_PARTITION_ID;
use iggy_common::store_consumer_offset::StoreConsumerOffset;
use iggy_common::wire_conversions::{consumer_to_wire, identifier_to_wire,
partitioning_to_wire};
use iggy_common::{
- Consumer, Identifier, IggyError, IggyMessageView, PollMessages,
PolledMessages,
- RESYNC_REQUIRED_PARTITION_SENTINEL, SendMessages,
+ Consumer, ConsumerKind, Identifier, IggyError, IggyMessageView,
PollMessages, SendMessages,
};
use server_common::Message;
@@ -96,14 +95,14 @@ pub(in crate::http) fn encode_send_messages(
/// Map a validated HTTP poll query onto the wire `PollMessagesRequest` the
/// shared TCP resolver consumes, so both transports resolve one request shape.
-/// The query's consumer kind is structurally always `Consumer`
-/// (`Consumer::kind` is `#[serde(skip)]` - the flattened `kind` param names
-/// the polling strategy), matching the legacy HTTP server.
+/// Rejects `consumer_kind=consumer_group` since HTTP has no persistent
connection.
pub(in crate::http) fn poll_wire_request(
stream_id: &Identifier,
topic_id: &Identifier,
query: &PollMessages,
) -> Result<PollMessagesRequest, IggyError> {
+ refuse_consumer_group(query.consumer.kind)?;
+
Ok(PollMessagesRequest {
consumer: WireConsumer {
kind: query.consumer.kind.as_code(),
@@ -143,15 +142,17 @@ pub(in crate::http) fn consumer_offset_wire_request(
/// Map a validated HTTP store-offset body onto the wire request
/// (`StoreConsumerOffsetRequest`), `ack` pinned to `Quorum` so the route can
-/// await the committed reply. The body's consumer kind is structurally always
-/// `Consumer` (`Consumer::kind` is `#[serde(skip)]`), matching the legacy HTTP
-/// server; `partition_id` passes through as the wire `Option` (flag byte +
-/// u32) for the server-side resolvers to ground.
+/// await the committed reply. `partition_id` passes through as the wire
+/// `Option` (flag byte + u32) for the server-side resolvers to ground.
+/// Rejects `consumer_kind=consumer_group` since HTTP has no persistent
+/// connection to hold group membership, so the write would always be fenced.
pub(in crate::http) fn store_offset_wire_request(
stream_id: &Identifier,
topic_id: &Identifier,
command: &StoreConsumerOffset,
) -> Result<StoreConsumerOffsetRequest, IggyError> {
+ refuse_consumer_group(command.consumer.kind)?;
+
Ok(StoreConsumerOffsetRequest {
consumer: consumer_to_wire(&command.consumer)?,
stream_id: identifier_to_wire(stream_id)?,
@@ -164,13 +165,16 @@ pub(in crate::http) fn store_offset_wire_request(
/// Map a validated HTTP delete-offset request onto the wire request
/// (`DeleteConsumerOffsetRequest`), `ack` pinned to `Quorum` like
-/// [`store_offset_wire_request`].
+/// [`store_offset_wire_request`]. Rejects `consumer_kind=consumer_group` for
+/// the same reason [`store_offset_wire_request`] does.
pub(in crate::http) fn delete_offset_wire_request(
stream_id: &Identifier,
topic_id: &Identifier,
consumer: &Consumer,
partition_id: Option<u32>,
) -> Result<DeleteConsumerOffsetRequest, IggyError> {
+ refuse_consumer_group(consumer.kind)?;
+
Ok(DeleteConsumerOffsetRequest {
consumer: consumer_to_wire(consumer)?,
stream_id: identifier_to_wire(stream_id)?,
@@ -180,23 +184,6 @@ pub(in crate::http) fn delete_offset_wire_request(
})
}
-/// The empty `PolledMessages` a fenced consumer-group poll answers, carrying
-/// the re-sync sentinel in `partition_id` exactly as the TCP dispatch replies
-/// it, so an SDK re-syncs its assignment instead of reading end-of-partition.
-pub(in crate::http) const fn resync_required_polled_messages() ->
PolledMessages {
- PolledMessages {
- context:
iggy_binary_protocol::primitives::partition_history::PartitionContext {
- incarnation: 0,
- owner_generation: 0,
- metadata_op: 0,
- },
- partition_id: RESYNC_REQUIRED_PARTITION_SENTINEL,
- current_offset: 0,
- count: 0,
- messages: Vec::new(),
- }
-}
-
/// Build a `Message<RoutedRequestHeader>` for a control-plane write by
filling a zeroed
/// `#[repr(C)]` header, mirroring `wire::rewrite_request_body` and the
partition
/// reconciler's prepare builder. `body` is the already-encoded wire request,
@@ -224,6 +211,13 @@ pub(in crate::http) fn build_request_message(
message
}
+fn refuse_consumer_group(kind: ConsumerKind) -> Result<(), IggyError> {
+ if kind == ConsumerKind::ConsumerGroup {
+ return Err(IggyError::FeatureUnavailable);
+ }
+ Ok(())
+}
+
#[cfg(test)]
mod tests {
use super::*;
@@ -234,7 +228,7 @@ mod tests {
use iggy_binary_protocol::WireEncode;
use iggy_common::delete_consumer_offset::DeleteConsumerOffset;
use iggy_common::{
- Consumer, ConsumerKind, IggyMessagesBatch, IggyTimestamp,
Partitioning, PartitioningKind,
+ IggyMessagesBatch, IggyTimestamp, Partitioning, PartitioningKind,
PolledMessages,
PollingKind, PollingStrategy, Validatable,
};
use partitions::{Fragment, PollFragments};
@@ -429,6 +423,18 @@ mod tests {
assert!(wire.auto_commit);
}
+ #[test]
+ fn poll_wire_request_rejects_consumer_group_kind() {
+ let stream_id = Identifier::from_str_value("orders").expect("valid
stream id");
+ let topic_id = Identifier::from_str_value("1").expect("valid topic
id");
+ let query = PollMessages {
+ consumer: Consumer::group(Identifier::numeric(9).expect("valid
id")),
+ ..Default::default()
+ };
+ let error = poll_wire_request(&stream_id, &topic_id,
&query).expect_err("must reject");
+ assert!(matches!(error, IggyError::FeatureUnavailable));
+ }
+
#[test]
fn consumer_offset_wire_request_defaults_omitted_partition_to_zero() {
let stream_id = Identifier::numeric(1).expect("valid stream id");
@@ -442,6 +448,54 @@ mod tests {
assert_eq!(wire.consumer.kind, 1);
}
+ #[test]
+ fn consumer_offset_wire_request_maps_consumer_group_kind() {
+ let stream_id = Identifier::numeric(1).expect("valid stream id");
+ let topic_id = Identifier::numeric(1).expect("valid topic id");
+ let query = GetConsumerOffset {
+ consumer: Consumer::group(Identifier::numeric(7).expect("valid
id")),
+ partition_id: Some(2),
+ };
+ let wire = consumer_offset_wire_request(&stream_id, &topic_id,
&query).expect("maps");
+ assert_eq!(wire.consumer.kind, 2);
+ assert_eq!(wire.partition_id, Some(2));
+ }
+
+ #[test]
+ fn get_consumer_offset_query_parses_consumer_kind_param() {
+ let uri: Uri =
"/streams/1/topics/1/consumer-offsets?consumer_id=my-group&consumer_kind=consumer_group&partition_id=0"
+ .parse()
+ .expect("valid uri");
+ let Query(query) =
Query::<GetConsumerOffset>::try_from_uri(&uri).expect("parses");
+ assert_eq!(query.consumer.kind, ConsumerKind::ConsumerGroup);
+ assert_eq!(
+ query.consumer.id,
+ Identifier::named("my-group").expect("valid id")
+ );
+ }
+
+ #[test]
+ fn get_consumer_offset_query_defaults_consumer_kind_when_omitted() {
+ let uri: Uri = "/streams/1/topics/1/consumer-offsets?consumer_id=42"
+ .parse()
+ .expect("valid uri");
+ let Query(query) =
Query::<GetConsumerOffset>::try_from_uri(&uri).expect("parses");
+ assert_eq!(query.consumer.kind, ConsumerKind::Consumer);
+ }
+
+ #[test]
+ fn store_consumer_offset_json_body_round_trips_consumer_group_kind() {
+ let json = serde_json::json!({
+ "consumer_id": "my-group",
+ "consumer_kind": "consumer_group",
+ "partition_id": 0,
+ "offset": 42,
+ });
+ let command: StoreConsumerOffset =
+ serde_json::from_value(json).expect("valid StoreConsumerOffset
body");
+ assert_eq!(command.consumer.kind, ConsumerKind::ConsumerGroup);
+ }
+
#[test]
fn store_offset_wire_request_round_trips_with_quorum_ack() {
let stream_id = Identifier::numeric(1).expect("valid stream id");
@@ -519,6 +573,59 @@ mod tests {
assert_eq!(query.partition_id, None);
}
+ #[test]
+ fn delete_offset_query_parses_consumer_kind_param() {
+ let uri: Uri = "/x?consumer_kind=consumer_group&partition_id=3"
+ .parse()
+ .expect("valid uri");
+ let Query(query) =
Query::<DeleteConsumerOffset>::try_from_uri(&uri).expect("parses");
+ assert_eq!(query.consumer_kind, ConsumerKind::ConsumerGroup);
+ let bare: Uri = "/x".parse().expect("valid uri");
+ let Query(query) =
Query::<DeleteConsumerOffset>::try_from_uri(&bare).expect("parses");
+ assert_eq!(query.consumer_kind, ConsumerKind::Consumer);
+ }
+
+ #[test]
+ fn store_offset_wire_request_rejects_consumer_group_kind() {
+ let stream_id = Identifier::numeric(1).expect("valid stream id");
+ let topic_id = Identifier::named("orders").expect("valid topic id");
+ let command = StoreConsumerOffset {
+ context: None,
+ consumer: Consumer::group(Identifier::numeric(9).expect("valid
id")),
+ partition_id: Some(1),
+ offset: 42,
+ };
+ let error =
+ store_offset_wire_request(&stream_id, &topic_id,
&command).expect_err("must reject");
+ assert!(matches!(error, IggyError::FeatureUnavailable));
+ }
+
+ #[test]
+ fn refuse_consumer_group_rejects_group_kind() {
+ let error =
refuse_consumer_group(ConsumerKind::ConsumerGroup).expect_err("must reject");
+ assert!(matches!(error, IggyError::FeatureUnavailable));
+ }
+
+ #[test]
+ fn refuse_consumer_group_passes_non_group_kinds() {
+ assert!(refuse_consumer_group(ConsumerKind::Consumer).is_ok());
+ assert!(refuse_consumer_group(ConsumerKind::ExternalGroup).is_ok());
+ }
+
+ #[test]
+ fn delete_offset_wire_request_rejects_consumer_group_kind() {
+ let stream_id = Identifier::named("stream-1").expect("valid stream
id");
+ let topic_id = Identifier::numeric(2).expect("valid topic id");
+ let error = delete_offset_wire_request(
+ &stream_id,
+ &topic_id,
+ &Consumer::group(Identifier::numeric(9).expect("valid id")),
+ Some(1),
+ )
+ .expect_err("must reject");
+ assert!(matches!(error, IggyError::FeatureUnavailable));
+ }
+
/// Wrap one stored `SendMessages` batch (`[256B header][blob]`) as the
/// poll fragment the owning shard replies, the shape
/// `build_polled_messages_body` consumes.
@@ -593,12 +700,4 @@ mod tests {
Some(b"raw-header-bytes".as_ref())
);
}
-
- #[test]
- fn resync_required_polled_messages_carries_sentinel_partition() {
- let polled = resync_required_polled_messages();
- assert_eq!(polled.partition_id, RESYNC_REQUIRED_PARTITION_SENTINEL);
- assert_eq!(polled.count, 0);
- assert_eq!(polled.messages, []);
- }
}
