/// Create Kafka topic based on the provided configs.
///
/// This is create a topic with `n_sequencers` partitions.
///
/// This will NOT fail if the topic already exists!
Create Kafka topic based on the provided configs.
This is create a topic with `n_sequencers` partitions.

This will NOT fail if the topic already exists!
Create Kafka topic based on the provided configs . This is create a topic with ` n_sequencers ` partitions . This will NOT fail if the topic already exists!
async fn create_kafka_topic(
    kafka_connection: &str,
    database_name: &str,
    n_sequencers: NonZeroU32,
    cfg: &BTreeMap<String, String>,
) -> Result<(), WriteBufferError> {
    let admin = admin_client(kafka_connection)?;
    let mut topic = NewTopic::new(
        database_name,
        n_sequencers.get() as i32,
        TopicReplication::Fixed(1),
    );
    for (k, v) in cfg {
        topic = topic.set(k, v);
    }
    let opts = AdminOptions::default();
    let mut results = admin.create_topics([&topic], &opts).await?;
    assert_eq!(results.len(), 1, "created exactly one topic");
    let result = results.pop().expect("just checked the vector length");
    match result {
        Ok(topic) | Err((topic, RDKafkaErrorCode::TopicAlreadyExists)) => {
            assert_eq!(topic, database_name);
            Ok(())
        }
        Err((topic, code)) => {
            assert_eq!(topic, database_name);
            Err(format!("Cannot create topic '{}': {}", topic, code).into())
        }
    }
}
async fn create_kafka_topic ( kafka_connection : & str , database_name : & str , n_sequencers : NonZeroU32 , cfg : & BTreeMap < String , String > , ) -> Result < ( ) , WriteBufferError > { let admin = admin_client ( kafka_connection ) ? ; let mut topic = NewTopic :: new ( database_name , n_sequencers . get ( ) as i32 , TopicReplication :: Fixed ( 1 ) , ) ; for ( k , v ) in cfg { topic = topic . set ( k , v ) ; } let opts = AdminOptions :: default ( ) ; let mut results = admin . create_topics ( [ & topic ] , & opts ) . await ? ; assert_eq ! ( results len ( ) 1 " " ) ; let result = results . pop ( ) . expect ( " " ) ; match result { Ok ( topic ) | Err ( ( topic , RDKafkaErrorCode :: TopicAlreadyExists ) ) => { assert_eq ! ( topic database_name ) ; Ok ( ( ) ) } Err ( ( topic , code ) ) => { assert_eq ! ( topic database_name ) ; Err ( format ! ( " " topic code ) . into ( ) ) } } }
Create Kafka topic based on the provided configs.
Create Kafka topic based on the provided configs .
kafka_connection
&str
database_name
&str
n_sequencers
NonZeroU32
cfg
&BTreeMap<String, String>