mz_testdrive/action/kafka/
wait_topic.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10use std::time::Duration;
11
12use mz_ore::retry::Retry;
13
14use crate::action::{ControlFlow, State};
15use crate::parser::BuiltinCommand;
16
17pub async fn run_wait_topic(
18    mut cmd: BuiltinCommand,
19    state: &State,
20) -> Result<ControlFlow, anyhow::Error> {
21    let topic = cmd.args.string("topic")?;
22    cmd.args.done()?;
23
24    println!("Waiting for Kafka topic {} to exist", topic);
25    Retry::default()
26        .initial_backoff(Duration::from_millis(50))
27        .factor(1.5)
28        .max_duration(state.timeout)
29        .retry_async_canceling(|_| async { check_topic_exists(&topic, &*state).await })
30        .await?;
31
32    Ok(ControlFlow::Continue)
33}
34
35pub(crate) async fn check_topic_exists(topic: &str, state: &State) -> Result<(), anyhow::Error> {
36    let metadata = state
37        .kafka_admin
38        .inner()
39        // N.B. It is extremely important not to ask specifically
40        // about the topic here, even though the API supports it!
41        // Asking about the topic will create it automatically...
42        // with the wrong number of partitions. Yes, this is
43        // unbelievably horrible.
44        .fetch_metadata(None, Some(Duration::from_secs(10)))?;
45
46    let topic_exists = metadata.topics().iter().any(|t| t.name() == topic);
47    if !topic_exists {
48        Err(anyhow::anyhow!("topic {} doesn't exist", topic))
49    } else {
50        Ok(())
51    }
52}
mz_testdrive/action/kafka/wait_topic.rs

mz_testdrive/action/kafka/
wait_topic.rs
