krishvishal commented on code in PR #4392:
URL: https://github.com/apache/iggy/pull/4392#discussion_r4188049504
##########
core/partitions/src/partition_storage.rs:
##########
@@ -323,53 +283,92 @@ pub async fn configure_consumer_offsets_with_storage<S:
DurableStorage, B: Messa
{
let guard = consumer_group_offsets.pin();
for (group_id, offset) in recovered_groups.entries {
- let recovered_offset = offset.offset.load(Ordering::Relaxed);
- if recovered_offset > offset_space_ceiling {
- warn!(
- consumer_group_id = group_id.0,
- recovered_offset,
- current_offset,
- offset_space_ceiling,
- stream_id,
- topic_id,
- partition_id,
- "recovered consumer group offset ahead of partition data;
clamping"
- );
- offset.offset.store(current_offset, Ordering::Relaxed);
- }
- let committed_offset = offset.offset.load(Ordering::Relaxed);
- partition.seed_recovered_consumer_offset(
+ seed_recovered_offset(
+ partition,
+ namespace,
ConsumerKind::ConsumerGroup,
- offset.consumer_id,
- committed_offset,
- recovered_offset,
+ &offset,
+ Some(offset_space_ceiling),
+ current_offset,
);
guard.insert(group_id, offset);
}
}
+ // A partition made before this kind existed has no directory for it. The
checkpoint syncs
+ // every offset directory, so it has to exist before the first one.
+ storage
+ .create_directories(Path::new(&external_group_offsets_path))
+ .await
+ .map_err(|_| PartitionRecoveryError::ConsumerOffsetsLoad {
Review Comment:
Fixed: the new `CreateOffsetsDirectory` error keeps the io error and prints
it.
##########
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:
Already fixed: the unit test from `838231e89` covers it.
##########
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: the stream builder returns `FeatureUnavailable`, like the server poll
and the HTTP SDK.
##########
core/partitions/src/iggy_partition.rs:
##########
@@ -3301,38 +3264,39 @@ where
.collect()
}
- /// Snapshot dead group keys for deletion through the partition's VSR log.
- /// A local unlink could free the primary's quota while backups retained
- /// every older generation, so reclamation uses the same ordered delete as
- /// an explicit consumer-offset request.
+ /// Snapshot the committed offsets of every kind keyed by consumer group
+ /// whose group `is_live` rejects, ascending by id within each kind, for
+ /// deletion through the partition's VSR log. A local unlink could free the
+ /// primary's quota while backups retained every older generation, so
+ /// reclamation uses the same ordered delete as an explicit consumer-offset
+ /// request.
#[must_use]
- pub fn dead_consumer_group_offset_ids(&self, is_live: impl Fn(u64) ->
bool) -> Vec<u32> {
+ pub fn dead_group_offset_keys(
+ &self,
+ is_live: impl Fn(u64) -> bool,
+ ) -> Vec<(ConsumerKind, u32)> {
if !self.consensus.is_primary() || !self.consensus.is_normal() {
return Vec::new();
}
- // A stranded file already failed normal loading or unlink. Reissuing
- // replicated deletes every reconciliation pass cannot make its
- // filesystem writable and would create a permanent commit loop.
- // A single-replica explicit deletion can retry after repair. On a
- // replicated partition the file must be repaired or removed locally,
- // because older peers do not recognize a delete for a map-missing key.
- let capacity =
self.consumer_offset_capacity_for(ConsumerKind::ConsumerGroup);
let mut dead = Vec::new();
- for key in self.consumer_group_offsets.pin().keys() {
- let Ok(id) = u32::try_from(key.0) else {
- continue;
- };
- if !is_live(u64::from(id))
- && !capacity.is_stranded(id)
- && self
- .durable_consumer_offsets
- .contains(ConsumerKind::ConsumerGroup, id)
- {
- dead.push(id);
- }
+ for kind in [ConsumerKind::ConsumerGroup, ConsumerKind::ExternalGroup]
{
Review Comment:
Fixed: dead keys are sorted by group id, with both kinds of one group side
by side.
--
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]
