Fix intermittent IllegalGeneration error in spk publish

spk publish's listen_to_index_status_updates() subscribed to the
index-updates topic via Kafka consumer-group management and then
immediately committed starting offsets, before the consumer had
joined the group / finished rebalancing (the join only happens on
the first poll). Committing offsets with no valid group generation
intermittently failed with:

    failed to commit new starting offsets for the kafka index
    updates consumer: Consumer commit error: IllegalGeneration
    (Broker: Specified group generation id is not valid)

The commit served no purpose anyway: this consumer uses a fresh,
ephemeral Ulid group id every run, so the committed offsets are
never read back.

Switch this read-only consumer from subscribe() + commit() to
manual partition assignment (assign()) with the starting offsets
carried in the TopicPartitionList. Manual assignment has no group
generation, so the race cannot occur. The subscribe() call is
removed since a consumer cannot mix subscribe and assign.

The indexer's long-lived consumer is untouched; it legitimately
uses a stable group id and offset commits.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: jrray

crates/spk-storage/src/storage/messaging/kafka.rs

@@ -214,6 +214,15 @@ fn setup_running_interrupt_handling() -> Arc<AtomicBool> {
 /// the latest message because they want to use it to check whether
 /// the usual message producer is running by seeing if it has sent a
 /// message recently.
+///
+/// This uses manual partition assignment (rather than subscribing to
+/// the topic via consumer group management) so the starting offsets
+/// can be set up front without depending on a group join/rebalance.
+/// The consumers that use this have unique, ephemeral group ids and
+/// never reuse committed offsets, so there is no need for consumer
+/// group coordination here. Committing offsets before the group join
+/// completed was the source of intermittent `IllegalGeneration`
+/// ("Specified group generation id is not valid") errors.
 fn set_offsets_to_one_before_the_latest_message(
     consumer: Arc<StreamConsumer>,
     topic_name: &str,
@@ -258,11 +267,17 @@ fn set_offsets_to_one_before_the_latest_message(
         }
     }
 
-    // Update the offsets so the previous message(s) will be the first
-    // read from the stream.
-    consumer.commit(&offsets, CommitMode::Sync).map_err(|err| {
+    // Assign the partitions with their starting offsets so the
+    // previous message(s) will be the first read from the stream.
+    //
+    // Manual assignment is used instead of committing offsets for a
+    // subscribed (group-managed) consumer. The commit approach raced
+    // against the group join/rebalance and intermittently failed with
+    // `IllegalGeneration` because the consumer had no valid group
+    // generation yet.
+    consumer.assign(&offsets).map_err(|err| {
         Error::String(format!(
-            "failed to commit new starting offsets for the kafka {consumer_label} consumer: {err}"
+            "failed to assign starting offsets for the kafka {consumer_label} consumer: {err}"
         ))
     })
 }
@@ -324,14 +339,11 @@ pub(crate) async fn listen_to_index_status_updates(
         })?;
 
     let consumer = Arc::new(consumer);
-    consumer.subscribe(&[topic_name]).map_err(|err| {
-        Error::String(format!(
-            "failed to subscribe to kafka index updates topic: {err}"
-        ))
-    })?;
 
-    // Change the starting message offset for reading to the just
-    // before the last message in the topic.
+    // Manually assign the topic's partitions with the starting message
+    // offset for reading set to just before the last message in the
+    // topic. This intentionally does not use `subscribe` (consumer
+    // group management); see the note on the function below.
     set_offsets_to_one_before_the_latest_message(
         consumer.clone(),
         topic_name,

cspell.json

@@ -630,6 +630,7 @@
     "readdir",
     "readdirplus",
     "readlink",
+    "rebalance",
     "recursivebuilds",
     "reexec",
     "reinit",