test: add tooling to purge kafka topic data
@@ -323,8 +323,12 @@ impl KafkaBufferConsumer {
}
pub mod test_utils {
+ use std::time::Duration;
+
use rdkafka::{
- admin::{AdminClient, AdminOptions, NewTopic, TopicReplication},
+ admin::{
+ AdminClient, AdminOptions, AlterConfig, NewTopic, ResourceSpecifier, TopicReplication,
+ },
client::DefaultClientContext,
ClientConfig,
};
@@ -376,24 +380,51 @@ pub mod test_utils {
}};
}
+ fn admin_client(kafka_connection: &str) -> AdminClient<DefaultClientContext> {
+ let mut cfg = ClientConfig::new();
+ cfg.set("bootstrap.servers", kafka_connection);
+ cfg.set("message.timeout.ms", "5000");
+ cfg.create().unwrap()
+ }
+
/// Create Kafka topic for testing.
pub async fn create_kafka_topic(
kafka_connection: &str,
database_name: &str,
n_sequencers: u32,
) {
- let mut cfg = ClientConfig::new();
- cfg.set("bootstrap.servers", kafka_connection);
- cfg.set("message.timeout.ms", "5000");
+ let admin = admin_client(kafka_connection);
- let admin: AdminClient<DefaultClientContext> = cfg.create().unwrap();
let topic = NewTopic::new(
database_name,
n_sequencers as i32,
TopicReplication::Fixed(1),
- );
+ )
+ .set("cleanup.policy", "delete")
+ .set("retention.ms", "-1")
+ .set("segment.ms", "10");
+
+ let opts = AdminOptions::default();
+ admin.create_topics([&topic], &opts).await.unwrap();
+ }
+
+ /// Purge all records from given topic.
+ ///
+ /// **WARNING: Until <https://github.com/fede1024/rust-rdkafka/issues/385> is fixed, this requires a server-wide
+ /// `log.retention.check.interval.ms` of 100ms!**
+ pub async fn purge_kafka_topic(kafka_connection: &str, database_name: &str) {
+ let admin = admin_client(kafka_connection);
let opts = AdminOptions::default();
- admin.create_topics(&[topic], &opts).await.unwrap();
+
+ let cfg =
+ AlterConfig::new(ResourceSpecifier::Topic(database_name)).set("retention.ms", "1");
+ admin.alter_configs([&cfg], &opts).await.unwrap();
+
+ tokio::time::sleep(Duration::from_millis(200)).await;
+
+ let cfg =
+ AlterConfig::new(ResourceSpecifier::Topic(database_name)).set("retention.ms", "-1");
+ admin.alter_configs([&cfg], &opts).await.unwrap();
}
}
