mz_testdrive/action/kafka/
delete_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 crate::action::{ControlFlow, State};
11use crate::parser::BuiltinCommand;
12
13pub async fn run_delete_topic(
14    mut cmd: BuiltinCommand,
15    state: &mut State,
16) -> Result<ControlFlow, anyhow::Error> {
17    let topic_prefix = format!("testdrive-{}", cmd.args.string("topic")?);
18    cmd.args.done()?;
19
20    let topic_name = format!("{}-{}", topic_prefix, state.seed);
21
22    println!("Deleting Kafka topic {topic_name}");
23
24    mz_kafka_util::admin::delete_topic(&state.kafka_admin, &state.kafka_admin_opts, &topic_name)
25        .await?;
26    state.kafka_topics.remove(&topic_name);
27    Ok(ControlFlow::Continue)
28}