fix: address review comments

server/src/database.rs

@@ -1854,9 +1854,10 @@ mod tests {
 
         // wait for a bit so the database fails because the mock is missing
         let database_captured = Arc::clone(&database);
-        tokio::time::timeout(Duration::from_millis(100), async move {
-            database_captured.wait_for_init().await.unwrap();
-        })
+        tokio::time::timeout(
+            Duration::from_millis(100),
+            database_captured.wait_for_init(),
+        )
         .await
         .unwrap_err();
 
@@ -1867,11 +1868,10 @@ mod tests {
             .write_buffer_factory()
             .register_mock("my_mock".to_string(), state.clone());
 
-        tokio::time::timeout(Duration::from_secs(10), async move {
-            database.wait_for_init().await.unwrap();
-        })
-        .await
-        .unwrap();
+        tokio::time::timeout(Duration::from_secs(10), database.wait_for_init())
+            .await
+            .unwrap()
+            .unwrap();
     }
 
     /// Normally database rules are provided as grpc messages, but in

write_buffer/src/kafka.rs

@@ -549,6 +549,7 @@ where
             let error_code: RDKafkaErrorCode = error_code.into();
             match error_code {
                 RDKafkaErrorCode::UnknownTopic | RDKafkaErrorCode::UnknownTopicOrPartition => {
+                    // The caller is responsible for creating the topic, so this is somewhat OK.
                     Ok(None)
                 }
                 _ => Err(KafkaError::MetadataFetch(error_code).into()),
@@ -612,11 +613,19 @@ where
     C: Consumer<D> + Send + Sync + 'static,
     D: ConsumerContext,
 {
-    loop {
+    const N_TRIES: usize = 10;
+
+    for i in 0..N_TRIES {
         if let Some(partitions) = get_partitions(database_name, consumer).await? {
             return Ok(partitions);
         }
 
+        // debounce after first round
+        if i > 0 {
+            info!(topic=%database_name, "Topic does not have partitions after creating it, wait a bit and try again.");
+            tokio::time::sleep(Duration::from_millis(250)).await;
+        }
+
         if let Some(creation_config) = creation_config {
             create_kafka_topic(
                 kafka_connection,
@@ -631,6 +640,8 @@ where
                 .into());
         }
     }
+
+    Err(format!("Could not auto-create topic after {} tries.", N_TRIES).into())
 }
 
 /// Our own implementation of [`ClientContext`] to overwrite certain logging behavior.