From 242fcbc6e7be57f40aa4f6f5db663ae08898b5ee Mon Sep 17 00:00:00 2001 From: huxint Date: Sat, 15 Aug 2026 00:36:45 +0800 Subject: [PATCH] Fix lost Kafka catch-up signal that can hang Thunder startup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The catch-up completion signal is only sent inside the `message_buffer.len() >= batch_size` block, but catchup_sender is Some only on the single poll iteration where catch-up is first detected (init_data_downloaded flips to true and stays true). If the buffer holds fewer than batch_size messages on that iteration — guaranteed when a partition's remaining backlog is smaller than one batch, e.g. with --skip_to_latest or a low-traffic partition — the signal is dropped and never re-sent. main() then blocks forever on rx.recv() when --is_serving is set: finalize_init() never runs and the server never reports ready. Flush whatever is buffered and send the signal on the detection iteration itself, regardless of batch fill. When no catch-up signal is pending the control flow is unchanged. This also ensures the tail of the backlog is inserted before finalize_init() sorts and trims. --- thunder/kafka/tweet_events_listener_v2.rs | 47 +++++++++++++---------- 1 file changed, 26 insertions(+), 21 deletions(-) diff --git a/thunder/kafka/tweet_events_listener_v2.rs b/thunder/kafka/tweet_events_listener_v2.rs index 5b510bb4..15eebe02 100644 --- a/thunder/kafka/tweet_events_listener_v2.rs +++ b/thunder/kafka/tweet_events_listener_v2.rs @@ -198,28 +198,33 @@ async fn process_tweet_events_v2( message_buffer.extend(messages); - if message_buffer.len() >= batch_size { - batch_count += 1; - let messages = std::mem::take(&mut message_buffer); - let post_store_clone = Arc::clone(&post_store); - - let permit = if init_data_downloaded { - Some(semaphore.clone().acquire_owned().await.unwrap()) - } else { - None - }; - - let _ = tokio::task::spawn_blocking(move || { - let _permit = permit; - match deserialize_batch(messages) { - Err(e) => warn!("Error processing batch {}: {:#}", batch_count, e), - Ok((light_posts, delete_posts)) => { - post_store_clone.insert_posts(light_posts); - post_store_clone.mark_as_deleted(delete_posts); - } + // catchup_sender is Some only on the single iteration where catch-up + // is detected, so flush and signal even if a full batch has not + // accumulated; otherwise the signal is lost and startup blocks forever. + if message_buffer.len() >= batch_size || catchup_sender.is_some() { + if !message_buffer.is_empty() { + batch_count += 1; + let messages = std::mem::take(&mut message_buffer); + let post_store_clone = Arc::clone(&post_store); + + let permit = if init_data_downloaded { + Some(semaphore.clone().acquire_owned().await.unwrap()) + } else { + None }; - }) - .await; + + let _ = tokio::task::spawn_blocking(move || { + let _permit = permit; + match deserialize_batch(messages) { + Err(e) => warn!("Error processing batch {}: {:#}", batch_count, e), + Ok((light_posts, delete_posts)) => { + post_store_clone.insert_posts(light_posts); + post_store_clone.mark_as_deleted(delete_posts); + } + }; + }) + .await; + } if let Some((sender, lag)) = catchup_sender { info!("Completed kafka init for a single thread");