hubcio commented on code in PR #4392:
URL: https://github.com/apache/iggy/pull/4392#discussion_r4184303283
##########
core/sdk/src/http/consumer_offsets.rs:
##########
@@ -101,3 +104,12 @@ impl ConsumerOffsetClient for HttpClient {
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> {
Review Comment:
warning: HTTP drops the consumer kind, so `ConsumerGroup` offset calls and
polls also silently act as a plain consumer - the same reason this guard
exists. refuse everything but `ConsumerKind::Consumer` here, or open a
follow-up.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -885,25 +890,37 @@ where
let partition_id = wire.partition_id.unwrap_or(0);
let namespace =
resolve_partition_namespace(shard, &wire.stream_id, &wire.topic_id,
Some(partition_id))?;
+ if wire.consumer.kind != KIND_CONSUMER_GROUP && wire.consumer.kind !=
KIND_EXTERNAL_GROUP {
+ let consumer = polling_consumer_from_wire(&wire.consumer,
partition_id)?;
+ return Ok(Some((
+ namespace,
+ partition_id,
+ PartitionRead::ConsumerOffset { consumer },
+ )));
+ }
// A group offset is keyed by the group's monotonic id (any client may read
// it, member or not), the same key the write path is rewritten to. An
// unresolved group (e.g. deleted) has no offset, so the read reports None.
- let consumer = if wire.consumer.kind == KIND_CONSUMER_GROUP {
- let Some(group_id) = shard
- .plane
- .metadata()
- .mux_stm
- .streams()
- .resolve_consumer_group_id(&wire.stream_id, &wire.topic_id,
&wire.consumer.id)
- else {
- return Ok(None);
- };
- #[allow(clippy::cast_possible_truncation)]
- PollingConsumer::ConsumerGroup(group_id as usize, partition_id as
usize)
+ let Some(group_id) = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .resolve_consumer_group_id(&wire.stream_id, &wire.topic_id,
&wire.consumer.id)
+ else {
+ return Ok(None);
+ };
+ #[allow(clippy::cast_possible_truncation)]
+ let read = if wire.consumer.kind == KIND_EXTERNAL_GROUP {
+ PartitionRead::ExternalGroupOffset {
Review Comment:
warning: the SDK sends this read to the metadata leader, which can be a
lagging backup for this partition, so a kafka `OffsetFetch` right after
`OffsetCommit` can return the old offset. route it to the primary or document
the follower read.
##########
core/sdk/src/stream_builder/build/build_iggy_consumer.rs:
##########
@@ -58,6 +58,8 @@ pub(crate) async fn build_iggy_consumer(
let mut builder = match consumer_kind {
ConsumerKind::Consumer => client.consumer(consumer_name, stream,
topic, partition)?,
ConsumerKind::ConsumerGroup => client.consumer_group(consumer_name,
stream, topic)?,
+ // Holds offsets only, it cannot poll.
+ ConsumerKind::ExternalGroup => return
Err(IggyError::InvalidConfiguration),
Review Comment:
nit: this rejects `ExternalGroup` only after the stream, topic or producer
was already built, and returns a bare `InvalidConfiguration` with no log.
reject it before any server call and log which field is wrong.
##########
core/integration/tests/server/cg.rs:
##########
@@ -81,6 +81,14 @@ async fn offset_cleanup(harness: &TestHarness) {
consumer_group_offset_cleanup_scenario::run(harness).await;
}
+#[iggy_harness(
+ test_client_transport = [Tcp, WebSocket, Quic],
Review Comment:
nit: `Http` is left out here, so nothing tests the client-side
`refuse_external_group` guard or the CLI's `--kind external-group`. add a unit
test for the guard and a CLI set-then-get case.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -8099,6 +8062,11 @@ where
.len();
self.evict_committed_prefix(fenced_prefix).await;
+ if !offset_dirs_synced {
Review Comment:
warning: any sync error other than `NotFound` leaves the purge unrecorded,
so the reconciler's retry wipes sends this replica applied since - peers keep
them. when the chain is already reset, retry only the dir sync and the
`purge.gen` write.
also at line 8099.
##########
core/common/src/types/consumer/consumer_kind.rs:
##########
@@ -88,15 +91,55 @@ impl Consumer {
id,
}
}
+
+ /// Creates a new `ExternalGroup` from the `Identifier` of an Iggy
consumer group.
Review Comment:
nit: these docs say external groups are never polled but don't name the
errors - HTTP calls and polls both return `FeatureUnavailable`. note the HTTP
case here and on `ConsumerOffsetClient`, and the poll case on
`MessageClient::poll_messages`.
##########
core/partitions/src/partition_storage.rs:
##########
@@ -798,7 +838,11 @@ async fn load_partition_consumer_group_offsets<S:
DurableStorage>(
Ok(RecoveredOffsets::default())
}
Err(source) => Err(PartitionRecoveryError::ConsumerOffsetsLoad {
- consumer_kind: "consumer group",
+ consumer_kind: if kind == ConsumerKind::ExternalGroup {
Review Comment:
simplification: pick the label with `kind.as_str()` instead of this if/else,
and use `ConsumerKind::ExternalGroup.as_str()` at line 304 - drops the
duplicated literal. the message text becomes `consumer_group`, which no test or
doc depends on.
##########
core/partitions/src/state_transfer.rs:
##########
@@ -2092,17 +2146,14 @@ where
.retain(|start_offset, _| live.contains_key(start_offset));
}
- // An empty chain at frontier 0 tells the receiver to unlink its own,
so
- // serve it only when a recorded purge says the emptiness is the truth.
- // `install_state_transfer`'s `purge_advances` check re-decides that
- // against the metadata plane and refuses the rest.
+ // An empty chain at frontier 0 is still served. The `NothingCommitted`
Review Comment:
warning: without the old `NothingCommitted` check, a zero-byte replica whose
floor moved through an offsets-only window serves an empty chain, and a
receiver at frontier 0 installs it, claiming history it doesn't hold. keep
refusing when nothing backs the floor.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -1106,14 +1095,8 @@ where
}
}
}
- let mut directories = Vec::with_capacity(4);
- for directory in [
- &self.consumer_offsets_path,
- &self.consumer_group_offsets_path,
- ]
- .into_iter()
- .flatten()
- {
+ let mut directories = Vec::with_capacity(ConsumerKind::COUNT + 2);
+ for directory in self.consumer_offset_dirs.iter().flatten() {
Review Comment:
nit: every WAL checkpoint now syncs a fifth directory, even with no external
offsets. only multi-replica `Persisted` partitions pay it, so it's minor - skip
`external_groups/` while it's empty, or leave it.
##########
core/simulator/src/storage/purge.rs:
##########
@@ -25,7 +25,12 @@
//! The two controls make that distinction observable. Without a purge,
recovery
//! must preserve bookmark 2 and return messages 3 and 4. After a completed
purge,
//! recovery must find no bookmarks and return all five messages, 0 through 4.
-//! Both controls cover individual consumers and groups under both offset
policies.
+//! Both controls cover individual consumers, groups, and external groups under
+//! both offset policies. An external group is never polled, so its bookmark is
+//! read directly. Boot never clamps it, so a resurrected one would survive.
+//!
+//! A third test purges after one consumer or group directory is gone, a state
+//! that recovery accepts. The purge must still record its generation.
Review Comment:
nit: the new test only covers the `NotFound` case. nothing drives the other
sync-failure arm, where the generation stays unrecorded and the purge gets
retried - add a test that fails one offsets dir sync and checks the retry.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -752,8 +747,9 @@ where
/// Set the per-kind durable consumer-offset limit for this partition.
pub fn set_consumer_offsets_max(&mut self, offsets_max: usize) {
- self.consumer_offset_capacity.set_limit(offsets_max);
- self.consumer_group_offset_capacity.set_limit(offsets_max);
+ for capacity in &self.consumer_offset_capacities {
+ capacity.set_limit(offsets_max);
Review Comment:
nit: every kind gets its own `consumer_offsets_max` budget here, so external
groups add a third, but the comment at `core/server/config.toml:983` still
talks about two kinds. mention external groups there so operators size for
three.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
