krishvishal commented on code in PR #4392:
URL: https://github.com/apache/iggy/pull/4392#discussion_r4186239568
##########
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:
Fixed: a retry for a generation whose reset already ran redoes only the
offsets dir sync and the `purge.gen` write, and an install or a converge clears
the marker.
##########
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:
Fixed: the refusal is back for an offer with no segments, `next_offset` 0
and purge generation 0.
##########
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:
Docs only, your second option: `get_consumer_offset` documents the follower
read, and routing to the primary is in a drafted follow-up issue.
##########
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:
Not fixed here: the follow-up issue is drafted but not filed (your second
option), because refusing it changes behavior for current HTTP callers.
##########
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:
Not fixed, on purpose (your "or leave it"): a delete leaves the dirty set,
so skipping the directory can lose an unsynced unlink.
##########
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:
Fixed: a simulator test fails the external group dir sync across a power
loss, and a partition test drives the retry through `purge()`.
##########
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:
Fixed: `IggyConsumerConfig::validate` logs the bad `consumer_kind` and runs
first in all four builder entry points.
--
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]
