diff --git a/nisshi-smoke-test/src/broker.rs b/nisshi-smoke-test/src/broker.rs index a7d7fef5..8fbdedd9 100644 --- a/nisshi-smoke-test/src/broker.rs +++ b/nisshi-smoke-test/src/broker.rs @@ -30,12 +30,13 @@ use std::{ fmt, fs::{self, File}, + io, net::TcpListener, path::{Path, PathBuf}, process::{Child, Command, Stdio}, sync::atomic::{AtomicU16, Ordering}, thread, - time::Duration, + time::{Duration, SystemTime}, }; use crate::{KafkaCli, ScramLogin, StorageUrl, label, settings, timed_command, unique_name}; @@ -53,6 +54,13 @@ const CONTAINER_START_TIMEOUT: Duration = Duration::from_secs(60); /// the broker running; `run.sh` stops every broker whose file is still there when the run ends. const PID_FILE: &str = "broker.pid"; +/// The storage URL option that makes a broker run maintenance every 2 seconds instead of every 10 +/// minutes, for a test that waits for a maintenance run. +pub const FREQUENT_MAINTENANCE: &str = "maintenance_interval=2s"; + +/// What `docker cp` prints when the file to copy isn't in the container. +const DOCKER_CP_MISSING_FILE: &str = "Could not find the file"; + /// A broker the tests talk to: the shared one `just smoke` started, or one /// of their own from [`Broker::isolated`]. #[derive(Debug)] @@ -132,6 +140,14 @@ impl Broker { Self::launch(LaunchOptions::new(settings::storage_url_under_test())) } + /// Launches a broker of this test's own, as [`Broker::isolated`] does, that runs maintenance + /// every 2 seconds instead of every 10 minutes, for a test that waits for a maintenance run. + pub fn isolated_with_frequent_maintenance() -> Self { + Self::launch(LaunchOptions::new( + settings::storage_url_under_test().with_query_option(FREQUENT_MAINTENANCE), + )) + } + /// Launches `NISSHI_SMOKE_IMAGE` when set, otherwise `NISSHI_SMOKE_BIN`, and waits until it /// answers. Panics if it doesn't start, or if something else already listens on the port. pub fn launch(options: LaunchOptions) -> Self { @@ -172,30 +188,54 @@ impl Broker { /// Stops the broker as [`Broker::stop`] does, and starts it again on the same storage, on a /// new port. Panics if it fails its checks or doesn't start again. pub fn restart(self) -> Self { - self.restart_with(|_| {}) + self.restart_with(|_| {}, &[] as &[&str]) } /// Like [`Broker::restart`], but the broker starts again with `--authentication`, so every /// client must log in. The harness's readiness check logs in as `login`, so the broker starts /// again only if it still has that user's credentials. pub fn restart_requiring_login(self, login: &ScramLogin) -> Self { - self.restart_with(|options| options.login = Some(login.clone())) + self.restart_with( + |options| options.login = Some(login.clone()), + &[] as &[&str], + ) + } + + /// Like [`Broker::restart`], but the broker starts again on `storage`, with the same cluster + /// id and working directory. A relative SQLite path in `storage` therefore names a file next + /// to the broker's own database. + /// + /// Each of `files_to_delete`, a path relative to the working directory, is deleted while the + /// broker is stopped, so a test can make sure that the broker can't read them after the + /// restart. A file that doesn't exist is skipped. + pub fn restart_on_storage( + self, + storage: StorageUrl, + files_to_delete: &[impl AsRef], + ) -> Self { + self.restart_with(|options| options.storage = storage, files_to_delete) } - fn restart_with(mut self, change_options: impl FnOnce(&mut LaunchOptions)) -> Self { + fn restart_with( + mut self, + change_options: impl FnOnce(&mut LaunchOptions), + files_to_delete: &[impl AsRef], + ) -> Self { let deployment = self .deployment .take() .expect("only a broker this test launched can restart"); - let (stopped, files, mut options) = deployment.stop(); + let (stopped, mut files, mut options) = deployment.stop(); if let Err(reason) = stopped { panic!("{reason}"); } + files.delete(files_to_delete); options.port = free_port(); change_options(&mut options); + files.create_volume_for(&options.storage); let cluster_id = options.cluster_id.clone(); Self::start_with_files(options, files).unwrap_or_else(|failed_start| { @@ -285,21 +325,33 @@ impl Broker { &self.bootstrap } - /// Returns whether `path` exists where the broker runs: on this machine for a broker process, - /// and in its container for a broker container. Panics for the shared broker. + /// Returns whether the file at `path` exists where the broker runs: on this machine for a + /// broker process, and in its container for a broker container. A relative `path` starts at + /// the broker's working directory. + /// + /// Panics for the shared broker. `smoke-broker` launches that broker, so this `Broker` doesn't + /// know where the broker runs. pub fn file_exists(&self, path: &str) -> bool { - let deployment = self - .deployment - .as_ref() - .expect("only a broker this test launched has files the test can check"); + self.launched_deployment().file_metadata(path).is_some() + } - match &deployment.host { - Host::Process(_) => Path::new(path).exists(), + /// Returns when the file at `path`, where the broker runs, was last written, or `None` if the + /// file doesn't exist. A relative `path` starts at the broker's working directory. + /// + /// Panics for the shared broker. `smoke-broker` launches that broker, so this `Broker` doesn't + /// know where the broker runs. + pub fn file_modification_time(&self, path: &str) -> Option { + self.launched_deployment() + .file_metadata(path)? + .modified() + .ok() + } - Host::Container { name } => { - docker(&["exec", name, "test", "-e", path]).is_ok_and(|test| test.code == Some(0)) - } - } + /// Returns the broker that this `Broker` launched. Panics for the shared broker. + fn launched_deployment(&self) -> &Deployment { + self.deployment + .as_ref() + .expect("only a broker this test launched has files the test can check") } /// Stops a broker this test launched, failing if it exited before now, didn't exit with 0 on @@ -392,6 +444,49 @@ impl fmt::Display for FailedStart { } impl Deployment { + /// Returns the metadata of the file at `path` where the broker runs, or `None` if the file + /// doesn't exist. A relative `path` starts at the broker's working directory. + /// + /// Panics if the harness can't read the file for another reason, so that a test never takes + /// such a failure for a missing file. + fn file_metadata(&self, path: &str) -> Option { + match &self.host { + Host::Process(_) => match fs::metadata(self.files.dir.join(path)) { + Ok(metadata) => Some(metadata), + Err(err) if err.kind() == io::ErrorKind::NotFound => None, + Err(err) => panic!("{path}: {err}"), + }, + + Host::Container { name } => { + // The broker's image holds only the broker, without a shell or `stat`, so the + // harness copies the file out with `docker cp`, which keeps its modification time. + // `docker cp` reads a relative path from the container's root, not from its + // working directory. + let working_dir = inspect(name, "{{.Config.WorkingDir}}") + .unwrap_or_else(|reason| panic!("{reason}")); + let path_in_container = Path::new("/").join(working_dir).join(path); + + let copy = self.files.dir.join(unique_name("copied")); + let source = format!("{name}:{}", path_in_container.display()); + + let copied = docker(&["cp", &source, ©.display().to_string()]) + .unwrap_or_else(|reason| panic!("{reason}")); + + let metadata = match copied.code { + Some(0) => Some( + fs::metadata(©) + .unwrap_or_else(|err| panic!("{}: {err}", copy.display())), + ), + _ if copied.stderr.contains(DOCKER_CP_MISSING_FILE) => None, + _ => panic!("docker cp {source}: {}", copied.stderr.trim()), + }; + + _ = fs::remove_file(©); + metadata + } + } + } + /// Stops the broker and checks it. Returns the files and options too, for a restart. fn stop(self) -> (Result<(), String>, BrokerFiles, LaunchOptions) { let mut failures = Vec::new(); @@ -511,19 +606,90 @@ impl BrokerFiles { fs::create_dir_all(&dir) .unwrap_or_else(|err| panic!("broker directory {}: {err}", dir.display())); - let volume = (settings::broker_image().is_some() && storage.is_sqlite()).then(|| { + let mut files = Self { + dir, + volume: None, + starts: 0, + keep_dir: false, + }; + files.create_volume_for(storage); + + files + } + + /// Creates a volume for a SQLite broker in a container, if the files don't have one yet. + /// + /// The volume is the container's working directory, where the broker resolves a relative + /// `sqlite://` path. A database outside the volume is removed with the container, at the next + /// stop. + fn create_volume_for(&mut self, storage: &StorageUrl) { + if self.volume.is_none() && settings::broker_image().is_some() && storage.is_sqlite() { let volume = unique_name("nisshi-smoke-sqlite"); _ = docker(&["volume", "create", &label(), &volume]); - volume - }); + self.volume = Some(volume); + } + } - Self { - dir, - volume, - starts: 0, - keep_dir: false, + /// Deletes each of `paths`, relative to the broker's working directory, while the broker is + /// stopped. Skips a path that doesn't exist, and panics if a deletion fails. + /// + /// A broker container without a volume keeps its files in the container, which the harness + /// removed when the broker stopped, so there is nothing to delete. + fn delete(&self, paths: &[impl AsRef]) { + if paths.is_empty() { + return; + } + + match (&self.volume, settings::broker_image()) { + (_, None) => { + for path in paths { + let path = path.as_ref(); + match fs::remove_file(self.dir.join(path)) { + Ok(()) => {} + Err(err) if err.kind() == io::ErrorKind::NotFound => {} + Err(err) => panic!("{path}: {err}"), + } + } + } + + (Some(volume), Some(_)) => { + // The broker's image holds only the broker, without `rm`, so a container from the + // Kafka tools' image deletes the files from the volume. + let tools_image = inspect(&settings::kafka_container(), "{{.Config.Image}}") + .unwrap_or_else(|reason| panic!("{reason}")); + let mount = format!("{volume}:/data"); + let label = label(); + let mut args = vec![ + "run", + "--rm", + &label, + // The broker writes the volume as root. The tools' image runs as a user that + // can't delete root's files. + "--user=0", + "--volume", + &mount, + "--entrypoint", + "rm", + &tools_image, + "-f", + ]; + let paths_in_volume = paths + .iter() + .map(|path| format!("/data/{}", path.as_ref())) + .collect::>(); + args.extend(paths_in_volume.iter().map(String::as_str)); + + match docker(&args) { + Ok(deleted) if deleted.code == Some(0) => {} + deleted => { + panic!("couldn't delete {paths_in_volume:?} from {volume}: {deleted:?}") + } + } + } + + (None, Some(_)) => {} } } @@ -757,7 +923,8 @@ fn save_logs(name: &str, log: &Path) -> Result<(), String> { .map_err(|err| format!("couldn't write {}: {err}", log.display())) } -/// Removes a broker container, but not its volume, which [`BrokerFiles`] removes. +/// Removes a container that the harness started, but not a broker's volume, which [`BrokerFiles`] +/// removes. fn remove_container(name: &str) { _ = docker(&["rm", "--force", "--volumes", name]); } diff --git a/nisshi-smoke-test/src/kafka_cli.rs b/nisshi-smoke-test/src/kafka_cli.rs index 94f16276..66a55e61 100644 --- a/nisshi-smoke-test/src/kafka_cli.rs +++ b/nisshi-smoke-test/src/kafka_cli.rs @@ -52,6 +52,7 @@ mod configs; mod console_consumer; mod console_producer; mod consumer_groups; +mod delete_records; mod offsets; mod topics; mod verifiable_producer; @@ -68,11 +69,12 @@ pub const SASL_AUTHENTICATION_EXCEPTION: &str = "SaslAuthenticationException"; const CONSOLE_PRODUCER_SEND_ERROR: &str = "Error when sending message"; pub use cluster::{ApiVersions, ClusterId}; -pub use configs::DescribeConfigs; +pub use configs::{DescribeConfigs, DescribeUsers}; pub use console_consumer::{ - Consume, ConsumeMatching, ConsumedRecord, LOG_APPEND_TIME, PrintsPartitionLines, + Consume, ConsumeMatching, ConsumedRecord, LOG_APPEND_TIME, PRINTED_NULL, PrintsPartitionLines, }; pub use consumer_groups::{DescribeGroup, DescribeGroupState, GroupOffsets}; +pub use delete_records::DeleteRecords; pub use offsets::GetOffsets; pub use topics::{ CreateTopic, DeleteTopic, DescribeTopic, ListTopics, PartitionReplicas, RequiresExistingTopic, @@ -166,11 +168,41 @@ impl KafkaCli { )) } + /// Returns tools for the same broker whose clients give up on a call after `timeout`, instead + /// of after the Java client's default of 60 seconds. + pub fn with_api_timeout(&self, timeout: Duration) -> Self { + let timeout_millis = timeout.as_millis(); + + // The admin client raises `default.api.timeout.ms` to `request.timeout.ms`, 30 seconds by + // default, when it is lower, so the tools set both. + self.with_client_properties(&format!( + "default.api.timeout.ms={timeout_millis}\nrequest.timeout.ms={timeout_millis}\n" + )) + } + /// Returns tools for the same broker whose client config file holds these tools' client /// properties, followed by `properties`. fn with_client_properties(&self, properties: &str) -> Self { let client_properties = format!("{}{properties}", self.client_properties); - let path = format!("/tmp/{}.properties", unique_name("client")); + let path = self.write_file_in_container("client", "properties", &client_properties); + + Self { + container: self.container.clone(), + bootstrap: self.bootstrap.clone(), + client_properties, + client_config: Some(path), + } + } + + /// Writes `contents` to a new file in the tools' container, and returns the file's path there. + /// The file is named `-.`. + fn write_file_in_container( + &self, + name_prefix: &str, + extension: &str, + contents: &str, + ) -> String { + let path = format!("/tmp/{}.{extension}", unique_name(name_prefix)); let written = timed_command::run( Command::new("docker").args([ @@ -181,7 +213,7 @@ impl KafkaCli { "-c", &format!("cat > {path}"), ]), - Some(&client_properties), + Some(contents), TOOL_TIMEOUT, ); @@ -191,12 +223,7 @@ impl KafkaCli { self.container ); - Self { - container: self.container.clone(), - bootstrap: self.bootstrap.clone(), - client_properties, - client_config: Some(path), - } + path } /// Runs `/opt/kafka/bin/.sh --bootstrap-server `. The address comes last, @@ -359,7 +386,7 @@ impl KafkaCli { .arg("timeout") .arg(format!("--kill-after={}s", KILL_AFTER.as_secs())) .arg(format!("{}s", timeout.as_secs())) - .arg(format!("/opt/kafka/bin/{}.sh", tool.script())) + .arg(format!("/opt/kafka/bin/{}.sh", tool.name())) .args(args) .args(["--bootstrap-server", &self.bootstrap]); @@ -380,14 +407,16 @@ enum Tool { ConsoleConsumer, ConsoleProducer, ConsumerGroups, + DeleteRecords, GetOffsets, Topics, VerifiableProducer, } impl Tool { - /// Returns the tool's script name, without `.sh`. - fn script(self) -> &'static str { + /// Returns the tool's name, which is also its script's file name in `/opt/kafka/bin` without + /// `.sh`. + fn name(self) -> &'static str { match self { Self::BrokerApiVersions => "kafka-broker-api-versions", Self::Cluster => "kafka-cluster", @@ -395,6 +424,7 @@ impl Tool { Self::ConsoleConsumer => "kafka-console-consumer", Self::ConsoleProducer => "kafka-console-producer", Self::ConsumerGroups => "kafka-consumer-groups", + Self::DeleteRecords => "kafka-delete-records", Self::GetOffsets => "kafka-get-offsets", Self::Topics => "kafka-topics", Self::VerifiableProducer => "kafka-verifiable-producer", @@ -411,6 +441,7 @@ impl Tool { Self::BrokerApiVersions | Self::Configs | Self::ConsumerGroups + | Self::DeleteRecords | Self::GetOffsets | Self::Topics => "--command-config", } @@ -446,7 +477,7 @@ impl Output { timeout: Option, ) -> Self { Self { - command: format!("{} {}", tool.script(), args.join(" ")), + command: format!("{} {}", tool.name(), args.join(" ")), code: finished.code, stdout: finished.stdout, stderr: finished.stderr, diff --git a/nisshi-smoke-test/src/kafka_cli/configs.rs b/nisshi-smoke-test/src/kafka_cli/configs.rs index fcbff4db..fc4041ff 100644 --- a/nisshi-smoke-test/src/kafka_cli/configs.rs +++ b/nisshi-smoke-test/src/kafka_cli/configs.rs @@ -16,11 +16,24 @@ use super::{KafkaCli, Output, ScramMechanism, Tool}; +/// How `kafka-configs --describe --entity-type users` starts the line that lists a user's SCRAM +/// mechanisms, before the user's name. +const SCRAM_CREDENTIALS_LINE_PREFIX: &str = "SCRAM credential configs for user-principal '"; +/// What `kafka-configs --describe --entity-type users` prints between the user's name and the +/// mechanisms, which it separates with [`SCRAM_MECHANISM_SEPARATOR`]. +const SCRAM_CREDENTIALS_LINE_INFIX: &str = "' are "; +/// What `kafka-configs --describe --entity-type users` prints between a user's mechanisms. +const SCRAM_MECHANISM_SEPARATOR: &str = ", "; + /// The command of the [`Output`] of `kafka-configs --describe`, from /// [`KafkaCli::describe_topic_configs`] or [`KafkaCli::describe_broker_defaults`]. #[derive(Clone, Copy, Debug)] pub enum DescribeConfigs {} +/// The command of [`KafkaCli::describe_users`]'s [`Output`]. +#[derive(Clone, Copy, Debug)] +pub enum DescribeUsers {} + impl KafkaCli { /// Describes `topic`'s config overrides, one `= sensitive= synonyms={...}` /// line each. With `all_configs`, it describes every config the topic has, defaults included. @@ -118,6 +131,35 @@ impl KafkaCli { ], ) } + + /// Describes every user's SCRAM credentials, with one + /// `SCRAM credential configs for user-principal '' are =iterations=, ...` + /// line per user. + /// + /// Sends `DescribeUserScramCredentials`. + pub fn describe_users(&self) -> Output { + self.run(Tool::Configs, &["--entity-type", "users", "--describe"]) + } +} + +impl Output { + /// Returns the names of the SCRAM mechanisms that `user` has credentials for, such as + /// `SCRAM-SHA-256`, or `None` if the output doesn't list `user`. + pub fn scram_mechanisms(&self, user: &str) -> Option> { + let user_line_start = + format!("{SCRAM_CREDENTIALS_LINE_PREFIX}{user}{SCRAM_CREDENTIALS_LINE_INFIX}"); + + self.lines().iter().find_map(|line| { + let mechanisms = line.strip_prefix(&user_line_start)?; + + Some( + mechanisms + .split(SCRAM_MECHANISM_SEPARATOR) + .filter_map(|mechanism| Some(mechanism.split_once('=')?.0)) + .collect(), + ) + }) + } } impl Output { diff --git a/nisshi-smoke-test/src/kafka_cli/console_consumer.rs b/nisshi-smoke-test/src/kafka_cli/console_consumer.rs index 963a9bd7..24f841e6 100644 --- a/nisshi-smoke-test/src/kafka_cli/console_consumer.rs +++ b/nisshi-smoke-test/src/kafka_cli/console_consumer.rs @@ -24,6 +24,9 @@ const PARTITION_FIELD_PREFIX: &str = "Partition:"; const OFFSET_FIELD_PREFIX: &str = "Offset:"; /// The timestamp type the console consumer prints for a record whose timestamp the broker set. pub const LOG_APPEND_TIME: &str = "LogAppendTime"; +/// What the console consumer prints for a record's key or value when it is null, such as a +/// tombstone's value. +pub const PRINTED_NULL: &str = "null"; /// The command of [`KafkaCli::consume`]'s [`Output`]. #[derive(Clone, Copy, Debug)] @@ -97,9 +100,33 @@ impl KafkaCli { partition: u32, offset: u64, count: usize, + ) -> Vec { + self.read_records_from(topic, partition, &offset.to_string(), count) + } + + /// Like [`KafkaCli::read_records`], but starts at the partition's earliest offset, for a test + /// that doesn't know which records retention or compaction left. + #[track_caller] + pub fn read_records_from_earliest( + &self, + topic: &str, + partition: u32, + count: usize, + ) -> Vec { + self.read_records_from(topic, partition, "earliest", count) + } + + /// Reads `count` records as [`KafkaCli::read_records`] does, from `offset`, which is an + /// offset or `earliest`. + #[track_caller] + fn read_records_from( + &self, + topic: &str, + partition: u32, + offset: &str, + count: usize, ) -> Vec { let partition = partition.to_string(); - let offset = offset.to_string(); let max_messages = count.to_string(); let read: Output = self.run( @@ -114,7 +141,7 @@ impl KafkaCli { "--partition", &partition, "--offset", - &offset, + offset, "--property", "print.timestamp=true", "--property", @@ -247,9 +274,9 @@ pub struct ConsumedRecord { pub timestamp_type: String, pub timestamp: i64, pub offset: i64, - /// `null` for a record that does not have a key. + /// [`PRINTED_NULL`] for a record that does not have a key. pub key: String, - /// `null` for a record that does not have a value, such as a tombstone. + /// [`PRINTED_NULL`] for a record that does not have a value, such as a tombstone. pub value: String, } diff --git a/nisshi-smoke-test/src/kafka_cli/delete_records.rs b/nisshi-smoke-test/src/kafka_cli/delete_records.rs new file mode 100644 index 00000000..c2323b64 --- /dev/null +++ b/nisshi-smoke-test/src/kafka_cli/delete_records.rs @@ -0,0 +1,65 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! `kafka-delete-records`, which deletes a partition's records before an offset. + +use super::{KafkaCli, Output, Tool}; + +/// How `kafka-delete-records` starts the line for each partition, before `-`. +const PARTITION_LINE_PREFIX: &str = "partition: "; +/// How `kafka-delete-records` starts the field that holds a partition's new low watermark. +const LOW_WATERMARK_FIELD_PREFIX: &str = "low_watermark: "; + +/// The command of [`KafkaCli::delete_records_before`]'s [`Output`]. +#[derive(Clone, Copy, Debug)] +pub enum DeleteRecords {} + +impl KafkaCli { + /// Deletes the records of `topic`'s `partition` before `offset`. The tool prints a + /// `partition: -\tlow_watermark: ` line for the partition. + /// + /// Sends `DeleteRecords`. + pub fn delete_records_before( + &self, + topic: &str, + partition: u32, + offset: i64, + ) -> Output { + let offsets = format!( + r#"{{"partitions":[{{"topic":"{topic}","partition":{partition},"offset":{offset}}}],"# + ) + r#""version":1}"#; + let offset_json_file = self.write_file_in_container("delete-records", "json", &offsets); + + self.run( + Tool::DeleteRecords, + &["--offset-json-file", &offset_json_file], + ) + } +} + +impl Output { + /// Returns the low watermark that the tool printed for `topic`'s `partition`, or `None` if it + /// printed none, such as when the broker returned an error for the partition. + pub fn low_watermark(&self, topic: &str, partition: u32) -> Option { + let partition_line_start = format!("{PARTITION_LINE_PREFIX}{topic}-{partition}\t"); + + self.lines().iter().find_map(|line| { + line.strip_prefix(&partition_line_start)? + .strip_prefix(LOW_WATERMARK_FIELD_PREFIX)? + .trim() + .parse() + .ok() + }) + } +} diff --git a/nisshi-smoke-test/src/kafka_cli/verifiable_producer.rs b/nisshi-smoke-test/src/kafka_cli/verifiable_producer.rs index dc280e6f..0f2fde0f 100644 --- a/nisshi-smoke-test/src/kafka_cli/verifiable_producer.rs +++ b/nisshi-smoke-test/src/kafka_cli/verifiable_producer.rs @@ -58,23 +58,59 @@ impl KafkaCli { max_messages: usize, acks: Acks, value_prefix: u32, + ) -> Output { + self.run_verifiable_producer(topic, max_messages, acks, value_prefix, &[]) + } + + /// Produces `max_messages` records timestamped about `create_time_millis`, so a test can + /// produce records that are already older than a topic's `retention.ms`. + /// + /// Record `n`'s value is `0.`. With `repeating_keys` of 2, the keys are `0`, `1`, `0`, `1`, + /// and so on; without it, the records have no keys. + pub fn verifiable_produce_created_at( + &self, + topic: &str, + max_messages: usize, + create_time_millis: i64, + repeating_keys: Option, + ) -> Output { + const VALUE_PREFIX: u32 = 0; + + let create_time_millis = create_time_millis.to_string(); + let mut args = vec!["--message-create-time", &create_time_millis]; + + let repeating_keys = repeating_keys.map(|keys| keys.to_string()); + if let Some(repeating_keys) = &repeating_keys { + args.extend(["--repeating-keys", repeating_keys]); + } + + self.run_verifiable_producer(topic, max_messages, Acks::FullIsr, VALUE_PREFIX, &args) + } + + fn run_verifiable_producer( + &self, + topic: &str, + max_messages: usize, + acks: Acks, + value_prefix: u32, + more_args: &[&str], ) -> Output { let max_messages = max_messages.to_string(); let value_prefix = value_prefix.to_string(); - self.run( - Tool::VerifiableProducer, - &[ - "--topic", - topic, - "--max-messages", - &max_messages, - "--acks", - acks.setting(), - "--value-prefix", - &value_prefix, - ], - ) + let mut args = vec![ + "--topic", + topic, + "--max-messages", + &max_messages, + "--acks", + acks.setting(), + "--value-prefix", + &value_prefix, + ]; + args.extend(more_args); + + self.run(Tool::VerifiableProducer, &args) } /// Like [`KafkaCli::verifiable_produce`] with [`Acks::FullIsr`], but the producer is diff --git a/nisshi-smoke-test/src/lib.rs b/nisshi-smoke-test/src/lib.rs index 6e6018f8..1c325da8 100644 --- a/nisshi-smoke-test/src/lib.rs +++ b/nisshi-smoke-test/src/lib.rs @@ -18,7 +18,10 @@ //! `just smoke ` passes the suite its configuration through the environment, which //! [`settings`] reads. -use std::time::{SystemTime, UNIX_EPOCH}; +use std::{ + thread, + time::{Duration, Instant, SystemTime, UNIX_EPOCH}, +}; use nanoid::nanoid; @@ -29,14 +32,15 @@ pub mod settings; mod storage_url; mod timed_command; -pub use broker::{Broker, LaunchOptions, free_port}; +pub use broker::{Broker, FREQUENT_MAINTENANCE, LaunchOptions, free_port}; pub use kafka_cli::{ Acks, AnyCommand, ApiVersions, ClusterId, Consume, ConsumeMatching, ConsumedRecord, - CreateTopic, DeleteTopic, DescribeConfigs, DescribeGroup, DescribeGroupState, DescribeTopic, - GetOffsets, GroupOffsets, KafkaCli, LOG_APPEND_TIME, ListTopics, Output, PartitionReplicas, - PrintsPartitionLines, RECORD_TOO_LARGE_EXCEPTION, RequiresExistingTopic, - SASL_AUTHENTICATION_EXCEPTION, ScramLogin, ScramMechanism, TOPIC_EXISTS_EXCEPTION, - UNSUPPORTED_VERSION_EXCEPTION, VerifiableProduce, verifiable_producer_values, + CreateTopic, DeleteRecords, DeleteTopic, DescribeConfigs, DescribeGroup, DescribeGroupState, + DescribeTopic, DescribeUsers, GetOffsets, GroupOffsets, KafkaCli, LOG_APPEND_TIME, ListTopics, + Output, PRINTED_NULL, PartitionReplicas, PrintsPartitionLines, RECORD_TOO_LARGE_EXCEPTION, + RequiresExistingTopic, SASL_AUTHENTICATION_EXCEPTION, ScramLogin, ScramMechanism, + TOPIC_EXISTS_EXCEPTION, UNSUPPORTED_VERSION_EXCEPTION, VerifiableProduce, + verifiable_producer_values, }; pub use record::Record; pub use storage_url::StorageUrl; @@ -75,6 +79,27 @@ pub fn now_in_millis() -> i64 { .unwrap_or_default() } +/// Calls `attempt` until it returns `Ok`, and returns that value. Panics with the last `Err` if +/// `attempt` hasn't returned `Ok` within `timeout`. +/// +/// A test waits with this for something the broker does in the background, such as a maintenance +/// run. The test stops as soon as it sees the result, so `timeout` only limits a test that has +/// already failed. +#[track_caller] +pub fn wait_until(timeout: Duration, mut attempt: impl FnMut() -> Result) -> T { + let deadline = Instant::now() + timeout; + + loop { + match attempt() { + Ok(value) => return value, + Err(reason) if Instant::now() >= deadline => { + panic!("not within {timeout:?}: {reason}"); + } + Err(_) => thread::sleep(Duration::from_millis(500)), + } + } +} + /// The `docker --label` on every container and volume this run creates. fn label() -> String { format!("--label=nisshi-smoke={}", settings::run_id()) diff --git a/nisshi-smoke-test/src/storage_url.rs b/nisshi-smoke-test/src/storage_url.rs index 18b3da1a..9233538e 100644 --- a/nisshi-smoke-test/src/storage_url.rs +++ b/nisshi-smoke-test/src/storage_url.rs @@ -35,6 +35,13 @@ impl StorageUrl { pub fn is_sqlite(&self) -> bool { self.0.starts_with("sqlite:") } + + /// Returns the path of a SQLite URL's database, as the broker reads it: the text after + /// `sqlite://`, without the query. Returns `None` for a URL of another engine. + pub fn sqlite_database_path(&self) -> Option<&str> { + let path = self.0.strip_prefix("sqlite://")?; + Some(path.split_once('?').map_or(path, |(path, _)| path)) + } } impl fmt::Display for StorageUrl { @@ -42,3 +49,22 @@ impl fmt::Display for StorageUrl { f.write_str(&self.0) } } + +#[cfg(test)] +mod tests { + use super::StorageUrl; + + #[test] + fn sqlite_database_path_leaves_out_the_query() { + let storage = StorageUrl::as_given("sqlite://nisshi.db").with_query_option("vacuum_into=x"); + + assert_eq!(storage.sqlite_database_path(), Some("nisshi.db")); + } + + #[test] + fn sqlite_database_path_is_none_for_another_engine() { + let storage = StorageUrl::as_given("postgres://postgres:postgres@localhost"); + + assert_eq!(storage.sqlite_database_path(), None); + } +} diff --git a/nisshi-smoke-test/tests/smoke/compaction.rs b/nisshi-smoke-test/tests/smoke/compaction.rs new file mode 100644 index 00000000..fb030fc7 --- /dev/null +++ b/nisshi-smoke-test/tests/smoke/compaction.rs @@ -0,0 +1,169 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Tests that the broker's maintenance run compacts a topic with `cleanup.policy=compact`: it +//! keeps the latest record of each key, and deletes the older ones. A user reads such a topic as a +//! table of each key's current value. A broker that keeps old values grows without bound, and a +//! broker that deletes the latest value loses the table's contents. +//! +//! These tests run on PostgreSQL and SQLite, the engines that compact a topic. Memory storage is +//! for development, and it doesn't compact. Each test launches a broker that runs maintenance +//! every 2 seconds. + +#![cfg(any(feature = "postgres", feature = "sqlite"))] + +use std::time::Duration; + +use nisshi_smoke_test::{Broker, KafkaCli, PRINTED_NULL, now_in_millis, wait_until}; + +/// How long a test waits for a maintenance run to give the result it expects. +const MAINTENANCE_TIMEOUT: Duration = Duration::from_secs(60); + +/// A record that a partition holds: its offset, and its key and value as the console consumer +/// prints them. +#[derive(Debug, PartialEq, Eq)] +struct StoredRecord<'a> { + offset: i64, + key: &'a str, + value: &'a str, +} + +/// Waits until the records from `topic`'s earliest offset are `expected`. Panics if that doesn't +/// happen within [`MAINTENANCE_TIMEOUT`]. +/// +/// Until a maintenance run compacts the topic, the topic holds at least as many records as +/// `expected`, so each read returns at once. +fn wait_until_records_are(cli: &KafkaCli, topic: &str, expected: &[StoredRecord<'_>]) { + wait_until(MAINTENANCE_TIMEOUT, || { + let consumed = cli.read_records_from_earliest(topic, 0, expected.len()); + let records = consumed + .iter() + .map(|record| StoredRecord { + offset: record.offset, + key: &record.key, + value: &record.value, + }) + .collect::>(); + + if records == expected { + Ok(()) + } else { + Err(format!( + "{topic}'s records are {records:?}, not {expected:?}" + )) + } + }); +} + +/// Compaction must keep only the latest value of each key, at the offset it was produced at, in +/// offset order. A consumer that reads the topic from the beginning then ends with each key's +/// current value. +#[test] +fn compaction_keeps_the_latest_value_of_each_key() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &["cleanup.policy=compact"]); + _ = cli.produce_keyed( + &topic, + &[ + ("k0", Some("v0")), + ("k1", Some("v1")), + ("k0", Some("v2")), + ("k1", Some("v3")), + ("k2", Some("v4")), + ], + ); + + wait_until_records_are( + &cli, + &topic, + &[ + StoredRecord { + offset: 2, + key: "k0", + value: "v2", + }, + StoredRecord { + offset: 3, + key: "k1", + value: "v3", + }, + StoredRecord { + offset: 4, + key: "k2", + value: "v4", + }, + ], + ); +} + +/// A tombstone, a record with a null value, deletes its key. Compaction must delete the key's +/// earlier values, and keep the tombstone itself, as Kafka does for `delete.retention.ms`, 24 +/// hours by default. A consumer that starts later still reads the tombstone, and deletes the key +/// from its own copy of the table. +#[test] +fn tombstone_is_the_only_record_left_for_its_key() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &["cleanup.policy=compact"]); + _ = cli.produce_keyed( + &topic, + &[("k0", Some("v0")), ("k1", Some("v1")), ("k0", None)], + ); + + wait_until_records_are( + &cli, + &topic, + &[ + StoredRecord { + offset: 1, + key: "k1", + value: "v1", + }, + StoredRecord { + offset: 2, + key: "k0", + value: PRINTED_NULL, + }, + ], + ); +} + +/// With `cleanup.policy=compact,delete`, maintenance must both compact the topic and delete the +/// records that `retention.ms` has expired. The expired records have keys of their own, so only +/// retention deletes them, and the new records share a key, so only compaction deletes the older +/// one. +#[test] +fn compact_delete_policy_compacts_and_deletes_expired_records() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = + cli.create_unique_topic(1, &["cleanup.policy=compact,delete", "retention.ms=600000"]); + let an_hour_ago = now_in_millis() - 3_600_000; + // Each record has a key of its own, `0` and `1`, because Kafka refuses a record without a key + // on a compacted topic. + let produced = cli.verifiable_produce_created_at(&topic, 2, an_hour_ago, Some(2)); + assert_eq!(produced.acknowledged_record_count(), 2, "{produced}"); + _ = cli.produce_keyed(&topic, &[("new", Some("v2")), ("new", Some("v3"))]); + + wait_until_records_are( + &cli, + &topic, + &[StoredRecord { + offset: 3, + key: "new", + value: "v3", + }], + ); +} diff --git a/nisshi-smoke-test/tests/smoke/delete_records.rs b/nisshi-smoke-test/tests/smoke/delete_records.rs new file mode 100644 index 00000000..f9ffa7ef --- /dev/null +++ b/nisshi-smoke-test/tests/smoke/delete_records.rs @@ -0,0 +1,95 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Tests that `kafka-delete-records` deletes a partition's records before an offset. A user runs it +//! to drop records that consumers must never read, such as bad data, without deleting the topic. +//! The broker must delete exactly the records before the offset, and report the partition's new +//! earliest offset. + +use nisshi_smoke_test::{Broker, KafkaCli}; + +/// How many records each test produces, at offsets 0 to 4. +const PRODUCED_RECORD_COUNT: usize = 5; + +/// The offset that each test deletes the records before. +const DELETE_BEFORE_OFFSET: i64 = 3; + +/// Launches a broker, and creates a topic on it with one partition that holds +/// [`PRODUCED_RECORD_COUNT`] records. +/// +/// Each test launches its own broker instead of using the shared one, because a broker that +/// panics on `DeleteRecords` would fail every other test on the shared broker too. +fn create_topic_with_records() -> (Broker, KafkaCli, String) { + let broker = Broker::isolated(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[]); + let values = (0..PRODUCED_RECORD_COUNT) + .map(|n| format!("value-{n}")) + .collect::>(); + _ = cli.produce_values(&topic, &values); + + (broker, cli, topic) +} + +/// The tool prints the low watermark that the broker returns, the partition's new earliest offset. +/// A user reads it to confirm what the deletion did. +#[test] +#[ignore = "the broker doesn't answer DeleteRecords: PostgreSQL's query fails, SQLite panics, \ + and memory storage never replies"] +fn delete_records_returns_the_new_low_watermark() { + let (_broker, cli, topic) = create_topic_with_records(); + + let deleted = cli.delete_records_before(&topic, 0, DELETE_BEFORE_OFFSET); + + assert_eq!( + deleted.succeeded().low_watermark(&topic, 0), + Some(DELETE_BEFORE_OFFSET), + "{deleted}" + ); +} + +/// After the deletion, the partition's earliest offset must be the offset that the records were +/// deleted before. A consumer that starts from the beginning then reads only the records the user +/// kept. +#[test] +#[ignore = "the broker doesn't answer DeleteRecords: PostgreSQL's query fails, SQLite panics, \ + and memory storage never replies"] +fn earliest_offset_moves_to_the_offset_deleted_before() { + let (_broker, cli, topic) = create_topic_with_records(); + _ = cli + .delete_records_before(&topic, 0, DELETE_BEFORE_OFFSET) + .succeeded(); + + assert_eq!( + cli.partition_offsets_at(&topic, "earliest"), + [DELETE_BEFORE_OFFSET] + ); +} + +/// The deletion must not change the partition's latest offset. A broker that numbers new records +/// from a lower offset gives two records one offset, and consumers skip or repeat records. +#[test] +#[ignore = "the broker doesn't answer DeleteRecords: PostgreSQL's query fails, SQLite panics, \ + and memory storage never replies"] +fn latest_offset_is_kept_after_delete_records() { + let (_broker, cli, topic) = create_topic_with_records(); + _ = cli + .delete_records_before(&topic, 0, DELETE_BEFORE_OFFSET) + .succeeded(); + + assert_eq!( + cli.partition_offsets_at(&topic, "latest"), + [PRODUCED_RECORD_COUNT as i64] + ); +} diff --git a/nisshi-smoke-test/tests/smoke/main.rs b/nisshi-smoke-test/tests/smoke/main.rs index e097adec..22801ae6 100644 --- a/nisshi-smoke-test/tests/smoke/main.rs +++ b/nisshi-smoke-test/tests/smoke/main.rs @@ -13,11 +13,16 @@ // limitations under the License. pub mod broker; +pub mod compaction; pub mod configs; pub mod consume; +pub mod delete_records; pub mod offsets; pub mod produce; pub mod restart; +pub mod retention; +pub mod scram; +pub mod sqlite_snapshot; pub mod storage_url; pub mod topic_lifecycle; pub mod topics; diff --git a/nisshi-smoke-test/tests/smoke/restart.rs b/nisshi-smoke-test/tests/smoke/restart.rs index e0bca056..1832a6c3 100644 --- a/nisshi-smoke-test/tests/smoke/restart.rs +++ b/nisshi-smoke-test/tests/smoke/restart.rs @@ -13,8 +13,8 @@ // limitations under the License. //! Tests that a broker stopped with SIGTERM and started again on the same storage still has -//! everything clients stored before: topics, records, committed group offsets, topic configs and -//! SCRAM users. A user who restarts the broker, for example to upgrade it, must lose none of it. +//! everything clients stored before: topics, records, committed group offsets and topic configs. A +//! user who restarts the broker, for example to upgrade it, must lose none of it. //! //! Memory storage keeps nothing, so on memory the tests check only that the broker starts again, //! empty and without errors. @@ -22,10 +22,7 @@ /// What clients stored survives a restart, on the engines that keep it. #[cfg(any(feature = "postgres", feature = "sqlite"))] mod persistence { - use nisshi_smoke_test::{ - Broker, KafkaCli, SASL_AUTHENTICATION_EXCEPTION, ScramLogin, ScramMechanism, - random_password, unique_name, - }; + use nisshi_smoke_test::{Broker, KafkaCli, unique_name}; /// A topic config that isn't the default, to find after the restart. const NON_DEFAULT_RETENTION: &str = "retention.ms=123456"; @@ -133,70 +130,6 @@ mod persistence { "the output does not contain {NON_DEFAULT_RETENTION}: {configs}" ); } - - /// A SCRAM user whose credentials vanish on a restart can no longer log in. The broker starts - /// again with `--authentication`, so a client must log in as the user to reach it. - fn scram_user_logs_in_after_restart(mechanism: ScramMechanism) { - let broker = Broker::isolated(); - let cli = KafkaCli::new(broker.bootstrap()); - let login = ScramLogin { - user: unique_name("user"), - password: random_password(), - mechanism, - }; - - _ = cli.add_scram_user(&login.user, &login.password).succeeded(); - - let broker = broker.restart_requiring_login(&login); - let cli = KafkaCli::logged_in_as(broker.bootstrap(), &login); - - let listed = cli.list_topics(); - assert_eq!( - listed.code, - Some(0), - "{} can't log in: {listed}", - login.user - ); - } - - #[test] - fn scram_sha_256_user_logs_in_after_restart() { - scram_user_logs_in_after_restart(ScramMechanism::Sha256); - } - - #[test] - fn scram_sha_512_user_logs_in_after_restart() { - scram_user_logs_in_after_restart(ScramMechanism::Sha512); - } - - /// After a restart, the broker must still check a SCRAM user's password, and refuse a client - /// that logs in with the wrong one. A broker that accepts any password lets anyone log in as - /// the user, and a client that logs in with the right password can't tell. - #[test] - fn wrong_password_is_refused_after_restart() { - let broker = Broker::isolated(); - let cli = KafkaCli::new(broker.bootstrap()); - let login = ScramLogin { - user: unique_name("user"), - password: random_password(), - mechanism: ScramMechanism::Sha256, - }; - - _ = cli.add_scram_user(&login.user, &login.password).succeeded(); - - let broker = broker.restart_requiring_login(&login); - let wrong_login = ScramLogin { - password: random_password(), - ..login - }; - let cli = KafkaCli::logged_in_as(broker.bootstrap(), &wrong_login); - - let listed = cli.list_topics(); - assert!( - listed.code != Some(0) && listed.mentions(SASL_AUTHENTICATION_EXCEPTION), - "the broker didn't refuse the wrong password: {listed}" - ); - } } /// Memory storage keeps nothing across a restart. diff --git a/nisshi-smoke-test/tests/smoke/retention.rs b/nisshi-smoke-test/tests/smoke/retention.rs new file mode 100644 index 00000000..d7cb6671 --- /dev/null +++ b/nisshi-smoke-test/tests/smoke/retention.rs @@ -0,0 +1,310 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Tests that the broker's maintenance run deletes the records that a topic's retention configs +//! expire, and keeps the others. A user sets `retention.ms` or `retention.bytes` to bound how much +//! a topic stores. A broker that deletes too little fills its storage, and a broker that deletes +//! too much loses records that consumers haven't read yet. +//! +//! These tests run on PostgreSQL and SQLite, the engines that apply retention. Memory storage is +//! for development, and it doesn't apply retention. Each test launches a broker that runs +//! maintenance every 2 seconds, and produces records with a timestamp an hour old, so that +//! retention has expired them already. + +#![cfg(any(feature = "postgres", feature = "sqlite"))] + +use std::time::Duration; + +use nisshi_smoke_test::{Broker, KafkaCli, now_in_millis, wait_until}; + +/// How long a test waits for a maintenance run to give the result it expects. +const MAINTENANCE_TIMEOUT: Duration = Duration::from_secs(60); + +/// A retention that has expired a record an hour old, and keeps a record produced during the test +/// for ten minutes, longer than any test runs. +const TEN_MINUTE_RETENTION: &str = "retention.ms=600000"; + +const DELETE_POLICY: &str = "cleanup.policy=delete"; + +/// How many expired records a test produces first, at offsets 0, 1 and 2. +const EXPIRED_RECORD_COUNT: usize = 3; + +/// Produces `count` records to `topic` with a timestamp an hour old. +fn produce_hour_old_records(cli: &KafkaCli, topic: &str, count: usize) { + let an_hour_ago = now_in_millis() - 3_600_000; + let produced = cli.verifiable_produce_created_at(topic, count, an_hour_ago, None); + + assert_eq!(produced.acknowledged_record_count(), count, "{produced}"); +} + +fn earliest_offset(cli: &KafkaCli, topic: &str) -> i64 { + cli.partition_offsets_at(topic, "earliest")[0] +} + +fn latest_offset(cli: &KafkaCli, topic: &str) -> i64 { + cli.partition_offsets_at(topic, "latest")[0] +} + +/// Waits until `topic`'s earliest offset is `expected`, which shows that a maintenance run deleted +/// the records before it. Panics if that doesn't happen within [`MAINTENANCE_TIMEOUT`]. +fn wait_until_earliest_offset_is(cli: &KafkaCli, topic: &str, expected: i64) { + wait_until(MAINTENANCE_TIMEOUT, || match earliest_offset(cli, topic) { + earliest if earliest == expected => Ok(()), + earliest => Err(format!( + "{topic}'s earliest offset is {earliest}, not {expected}" + )), + }); +} + +/// Asserts that `topic` still holds the `count` records produced to it, at offsets from 0. +/// +/// The test reads the records instead of checking that the earliest offset is still 0, because +/// the broker also reports 0 for a partition that retention emptied. +#[track_caller] +fn assert_every_record_kept(cli: &KafkaCli, topic: &str, count: usize) { + let offsets = cli + .read_records(topic, 0, 0, count) + .into_iter() + .map(|record| record.offset) + .collect::>(); + + assert_eq!(offsets, (0..count as i64).collect::>()); +} + +/// Waits until a maintenance run has deleted an expired record from a topic of its own. +/// +/// A test that expects a topic to keep its records can't wait for the records to go. It calls +/// this after it produces them, and then checks them. Each maintenance run applies retention to +/// every topic of the cluster at once, so the run that deleted this record also checked the +/// records that the test produced before it. +fn wait_for_maintenance_run(cli: &KafkaCli) { + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, TEN_MINUTE_RETENTION]); + produce_hour_old_records(cli, &topic, 1); + // The topic keeps a newer record, so retention never empties it. The broker reports the + // earliest offset of an empty partition as 0, and the wait would then never end. + _ = cli.produce_values(&topic, &["new-0"]); + wait_until_earliest_offset_is(cli, &topic, 1); +} + +/// Retention must delete the records older than `retention.ms`, and keep the newer ones that +/// consumers still need. The earliest offset moves to the first record it keeps. +#[test] +fn expired_records_are_deleted_and_newer_records_kept() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, TEN_MINUTE_RETENTION]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + _ = cli.produce_values(&topic, &["new-0", "new-1"]); + + wait_until_earliest_offset_is(&cli, &topic, EXPIRED_RECORD_COUNT as i64); + + let kept_records = cli.read_records_from_earliest(&topic, 0, 2); + let kept_offsets = kept_records.iter().map(|record| record.offset); + let kept_values = kept_records.iter().map(|record| record.value.as_str()); + assert_eq!(kept_offsets.collect::>(), [3, 4]); + assert_eq!(kept_values.collect::>(), ["new-0", "new-1"]); +} + +/// After retention deletes every record of a partition, the broker must give the next record the +/// offset after the last one it stored. A broker that numbers from the records that remain starts +/// again at 0, and a consumer that already read offsets 0 to 2 skips the new records. +#[test] +fn record_produced_after_retention_gets_the_next_offset() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, TEN_MINUTE_RETENTION]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + wait_for_maintenance_run(&cli); + + _ = cli.produce_values(&topic, &["after-retention"]); + + // The read starts at the earliest offset, so it also shows that retention deleted the + // expired records, and the partition was empty before the new record. + assert_eq!( + cli.read_records_from_earliest(&topic, 0, 1) + .into_iter() + .map(|record| (record.offset, record.value)) + .collect::>(), + [(EXPIRED_RECORD_COUNT as i64, "after-retention".to_owned())] + ); +} + +/// When retention deletes every record of a partition, the earliest and the latest offset must +/// both be the offset after the last deleted record, as Kafka reports them. A consumer group's +/// committed offset is at that offset, and a latest offset that goes back makes the group's +/// offset look out of range, so the group starts again from the beginning or the end. +#[test] +#[ignore = "the broker reports offsets 0 for a partition that retention emptied (#864)"] +fn offsets_are_kept_when_retention_empties_a_partition() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, TEN_MINUTE_RETENTION]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + + wait_for_maintenance_run(&cli); + + assert_eq!( + (earliest_offset(&cli, &topic), latest_offset(&cli, &topic)), + (EXPIRED_RECORD_COUNT as i64, EXPIRED_RECORD_COUNT as i64) + ); +} + +/// Kafka's default `cleanup.policy` is `delete`, so a topic that sets only `retention.ms` must be +/// cleaned the same way. Users often set only `retention.ms`. +#[test] +#[ignore = "retention deletes records only from a topic that sets cleanup.policy (#918)"] +fn topic_without_cleanup_policy_is_cleaned_by_retention() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[TEN_MINUTE_RETENTION]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + _ = cli.produce_values(&topic, &["new-0"]); + + wait_until_earliest_offset_is(&cli, &topic, EXPIRED_RECORD_COUNT as i64); +} + +/// `retention.ms=-1` means that the topic keeps every record, however old. A user sets it on a +/// topic that is the only copy of its data. +#[test] +#[ignore = "retention.ms=-1 makes retention delete every record (#919)"] +fn retention_ms_minus_1_keeps_every_record() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, "retention.ms=-1"]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + + wait_for_maintenance_run(&cli); + + assert_every_record_kept(&cli, &topic, EXPIRED_RECORD_COUNT); +} + +/// A topic's `retention.ms` of 30 days must not stop retention on the broker's other topics. The +/// value is larger than a 32-bit integer holds, and a broker that fails on it never deletes +/// another expired record, so its storage grows until it is full. +#[test] +#[cfg_attr( + feature = "postgres", + ignore = "a retention.ms larger than a 32-bit integer stops retention on every topic (#920)" +)] +fn long_retention_on_one_topic_does_not_stop_retention_on_another() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let thirty_day_topic = cli.create_unique_topic(1, &[DELETE_POLICY, "retention.ms=2592000000"]); + // The topic holds a record, so that the maintenance run checks the record's age against the + // topic's `retention.ms`. + _ = cli.produce_values(&thirty_day_topic, &["new-0"]); + + wait_for_maintenance_run(&cli); +} + +/// A topic with a `retention.ms` of 30 days must keep a new record. A broker that reads the value +/// into a 32-bit integer gets a negative or a small retention, and deletes the record at once. +#[test] +#[cfg_attr( + feature = "postgres", + ignore = "a retention.ms larger than a 32-bit integer stops retention on every topic (#920)" +)] +fn thirty_day_retention_keeps_a_new_record() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let thirty_day_topic = cli.create_unique_topic(1, &[DELETE_POLICY, "retention.ms=2592000000"]); + _ = cli.produce_values(&thirty_day_topic, &["new-0"]); + + wait_for_maintenance_run(&cli); + + assert_every_record_kept(&cli, &thirty_day_topic, 1); +} + +/// A `retention.ms` added to an existing topic with `kafka-configs` must apply from the next +/// maintenance run. A user lowers it to free storage at once. The topic keeps its records under +/// the default retention of 7 days first, so the test shows that the added config deleted them. +#[test] +fn added_retention_ms_deletes_older_records() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY]); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + _ = cli.produce_values(&topic, &["new-0"]); + wait_for_maintenance_run(&cli); + assert_every_record_kept(&cli, &topic, EXPIRED_RECORD_COUNT + 1); + + _ = cli + .add_topic_configs(&topic, &[TEN_MINUTE_RETENTION]) + .succeeded(); + + wait_until_earliest_offset_is(&cli, &topic, EXPIRED_RECORD_COUNT as i64); +} + +/// Deleting a topic's `retention.ms` with `kafka-configs` must restore the default of 7 days, which +/// keeps a record an hour old. A broker that keeps applying the deleted value deletes records the +/// user meant to keep. +#[test] +#[ignore = "a topic's own configs are reported as defaults, so kafka-configs refuses to \ + delete them (#904)"] +fn deleted_retention_ms_restores_the_default() { + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, TEN_MINUTE_RETENTION]); + _ = cli + .delete_topic_configs(&topic, &["retention.ms"]) + .succeeded(); + produce_hour_old_records(&cli, &topic, EXPIRED_RECORD_COUNT); + + wait_for_maintenance_run(&cli); + + assert_every_record_kept(&cli, &topic, EXPIRED_RECORD_COUNT); +} + +/// Retention must keep a partition within its topic's `retention.bytes`, by deleting the oldest +/// records, and keep the newest ones. A user sets it to bound a topic's storage, whatever the +/// records' age. +#[test] +#[ignore = "retention ignores retention.bytes (#921)"] +fn retention_bytes_keeps_partition_under_the_cap() { + /// The size of each record's value. The broker stores more bytes than this per record, so a + /// partition under the cap holds at most `RETENTION_BYTES / VALUE_BYTES` records. + const VALUE_BYTES: usize = 1000; + const RETENTION_BYTES: usize = 5 * VALUE_BYTES; + const PRODUCED_RECORD_COUNT: usize = 20; + + let broker = Broker::isolated_with_frequent_maintenance(); + let cli = KafkaCli::new(broker.bootstrap()); + let retention_bytes = format!("retention.bytes={RETENTION_BYTES}"); + let topic = cli.create_unique_topic(1, &[DELETE_POLICY, &retention_bytes]); + let values = (0..PRODUCED_RECORD_COUNT) + .map(|n| format!("{n:0>VALUE_BYTES$}")) + .collect::>(); + _ = cli.produce_values(&topic, &values); + + // A broker can delete the records in several runs, as Kafka does a segment at a time, so the + // test waits until the partition is under the cap, not for the first deletion. + wait_until(MAINTENANCE_TIMEOUT, || { + let kept_record_count = PRODUCED_RECORD_COUNT as i64 - earliest_offset(&cli, &topic); + + if kept_record_count <= (RETENTION_BYTES / VALUE_BYTES) as i64 { + Ok(()) + } else { + Err(format!( + "{topic} keeps {kept_record_count} records of {VALUE_BYTES} bytes, with \ + retention.bytes of {RETENTION_BYTES}" + )) + } + }); + + let newest_offset = PRODUCED_RECORD_COUNT as u64 - 1; + assert_eq!( + cli.read_records(&topic, 0, newest_offset, 1)[0].value, + values[PRODUCED_RECORD_COUNT - 1] + ); +} diff --git a/nisshi-smoke-test/tests/smoke/scram.rs b/nisshi-smoke-test/tests/smoke/scram.rs new file mode 100644 index 00000000..651540f8 --- /dev/null +++ b/nisshi-smoke-test/tests/smoke/scram.rs @@ -0,0 +1,162 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Tests that a user can create SCRAM users with `kafka-configs`, and that clients log in as them. +//! On a broker with `--authentication`, a client that can't log in can't do anything, and a broker +//! that lets in a client with a wrong password or none protects nothing. + +/// Creating and describing users, on a broker that doesn't require a login. +mod user_creation { + use nisshi_smoke_test::{KafkaCli, ScramMechanism, random_password, unique_name}; + + /// `kafka-configs --add-config` creates credentials with both mechanisms, and `kafka-configs + /// --describe` must list both for the user. A user checks the list before giving a client + /// the mechanism to log in with. + #[test] + #[ignore = "kafka-configs describes users' quotas too, with DescribeClientQuotas, which the \ + broker doesn't support (#922)"] + fn kafka_configs_user_is_described_with_both_mechanisms() { + let cli = KafkaCli::shared(); + let user = unique_name("user"); + _ = cli.add_scram_user(&user, &random_password()).succeeded(); + + let described = cli.describe_users(); + + assert_eq!( + described + .succeeded() + .scram_mechanisms(&user) + .map(|mut mechanisms| { + mechanisms.sort_unstable(); + mechanisms + }), + Some(ScramMechanism::ALL.map(ScramMechanism::name).to_vec()), + "{described}" + ); + } +} + +/// Logging in, on a broker with `--authentication`. +/// +/// A broker with `--authentication` refuses to create a user for a client that hasn't logged in, +/// so each test creates its user on a broker without `--authentication`, and restarts the broker +/// with it. Each test therefore also checks that the user, and its password, survive a restart. +/// Only PostgreSQL and SQLite keep the user across that restart, so the tests run on them. +#[cfg(any(feature = "postgres", feature = "sqlite"))] +mod login { + use std::time::Duration; + + use nisshi_smoke_test::{ + Broker, KafkaCli, SASL_AUTHENTICATION_EXCEPTION, ScramLogin, ScramMechanism, + random_password, unique_name, + }; + + fn login_for_new_user(mechanism: ScramMechanism) -> ScramLogin { + ScramLogin { + user: unique_name("user"), + password: random_password(), + mechanism, + } + } + + /// Creates `login`'s user with `kafka-configs` on a new broker, and restarts the broker with + /// `--authentication`. + fn broker_requiring_kafka_configs_user(login: &ScramLogin) -> Broker { + let broker = Broker::isolated(); + _ = KafkaCli::new(broker.bootstrap()) + .add_scram_user(&login.user, &login.password) + .succeeded(); + + broker.restart_requiring_login(login) + } + + /// Asserts that a client that logs in as `login` can list the broker's topics. + #[track_caller] + fn assert_logs_in(broker: &Broker, login: &ScramLogin) { + let listed = KafkaCli::logged_in_as(broker.bootstrap(), login).list_topics(); + + assert_eq!( + listed.code, + Some(0), + "{} can't log in with {}: {listed}", + login.user, + login.mechanism.name() + ); + } + + /// A client must be able to log in with SCRAM-SHA-256 as a user that `kafka-configs` created. + /// A broker that can't check those credentials locks every SCRAM-SHA-256 client out. + #[test] + fn kafka_configs_user_logs_in_with_scram_sha_256() { + let login = login_for_new_user(ScramMechanism::Sha256); + let broker = broker_requiring_kafka_configs_user(&login); + + assert_logs_in(&broker, &login); + } + + /// As for SCRAM-SHA-256, with SCRAM-SHA-512, which uses another hash function and so other + /// credentials. + #[test] + fn kafka_configs_user_logs_in_with_scram_sha_512() { + let login = login_for_new_user(ScramMechanism::Sha512); + let broker = broker_requiring_kafka_configs_user(&login); + + assert_logs_in(&broker, &login); + } + + /// The broker must refuse a client that logs in with the wrong password, and the tool must + /// report why. A broker that accepts any password lets anyone log in as the user. + #[test] + fn wrong_password_is_refused() { + let login = login_for_new_user(ScramMechanism::Sha256); + let broker = broker_requiring_kafka_configs_user(&login); + let wrong_login = ScramLogin { + password: random_password(), + ..login + }; + + let listed = KafkaCli::logged_in_as(broker.bootstrap(), &wrong_login).list_topics(); + + assert!( + listed.code != Some(0) && listed.mentions(SASL_AUTHENTICATION_EXCEPTION), + "the broker didn't refuse the wrong password: {listed}" + ); + } + + /// The broker must refuse a client without credentials, and keep serving the clients that log + /// in. A broker that serves the client lets anyone use the broker without a password. + /// + /// The broker closes the connection of a client that sends a request before it logs in, and + /// the client retries until its API timeout. A broker that stops answering every client also + /// makes the tool fail, so the test then checks that a client that logs in is still served. + #[test] + fn client_without_credentials_is_refused() { + /// Shorter than the Java client's default API timeout of 60 seconds, so that the tool + /// gives up on its own, before the harness stops it. + const CLIENT_API_TIMEOUT: Duration = Duration::from_secs(5); + + let login = login_for_new_user(ScramMechanism::Sha256); + let broker = broker_requiring_kafka_configs_user(&login); + + let listed = KafkaCli::new(broker.bootstrap()) + .with_api_timeout(CLIENT_API_TIMEOUT) + .list_topics(); + + assert!( + listed.timeout.is_none() && listed.code != Some(0), + "the broker served a client without credentials: {listed}" + ); + assert_logs_in(&broker, &login); + } +} diff --git a/nisshi-smoke-test/tests/smoke/sqlite_snapshot.rs b/nisshi-smoke-test/tests/smoke/sqlite_snapshot.rs new file mode 100644 index 00000000..c1e396fe --- /dev/null +++ b/nisshi-smoke-test/tests/smoke/sqlite_snapshot.rs @@ -0,0 +1,81 @@ +// Copyright ⓒ 2024-2026 Peter Morgan +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Tests the SQLite storage URL option `vacuum_into`, which makes each maintenance run write a +//! snapshot of the database to a file. A user keeps the snapshot as a backup, and a backup is +//! useful only if a broker can start on it and find the topics and records that it holds. + +#![cfg(feature = "sqlite")] + +use std::time::Duration; + +use nisshi_smoke_test::{ + Broker, FREQUENT_MAINTENANCE, KafkaCli, LaunchOptions, StorageUrl, settings, wait_until, +}; + +/// The snapshot's path, relative to the broker's working directory, where its database is too. +const SNAPSHOT_PATH: &str = "snapshot.db"; + +/// How long the test waits for a maintenance run to write the snapshot again. +const SNAPSHOT_TIMEOUT: Duration = Duration::from_secs(60); + +/// A broker started on a snapshot must have the topics and records that the snapshot's broker +/// stored before it wrote the snapshot. The test deletes the broker's own database before it +/// starts the broker on the snapshot, as a user restores a lost database, so the records can come +/// only from the snapshot. +#[test] +fn broker_started_on_a_snapshot_has_its_topics_and_records() { + let storage = settings::storage_url_under_test(); + let live_database = storage + .sqlite_database_path() + .expect("the SQLite leg runs on a sqlite:// URL") + .to_owned(); + let broker = Broker::launch(LaunchOptions::new( + storage + .with_query_option(FREQUENT_MAINTENANCE) + .with_query_option(&format!("vacuum_into={SNAPSHOT_PATH}")), + )); + let cli = KafkaCli::new(broker.bootstrap()); + let topic = cli.create_unique_topic(1, &[]); + let values = ["value-0", "value-1", "value-2"]; + _ = cli.produce_values(&topic, &values); + + // A run that started before the records were stored writes a snapshot without them, so the + // test waits until the broker has written the snapshot twice. + let mut last_written = broker.file_modification_time(SNAPSHOT_PATH); + for _ in 0..2 { + last_written = Some(wait_until(SNAPSHOT_TIMEOUT, || { + match broker.file_modification_time(SNAPSHOT_PATH) { + Some(written) if Some(&written) != last_written.as_ref() => Ok(written), + _ => Err(format!("the broker didn't write {SNAPSHOT_PATH} again")), + } + })); + } + + // SQLite keeps the newest writes in the `-wal` file, and its index in the `-shm` file. + let live_database_files = ["", "-wal", "-shm"].map(|suffix| format!("{live_database}{suffix}")); + let broker = broker.restart_on_storage( + StorageUrl::as_given(format!("sqlite://{SNAPSHOT_PATH}")), + &live_database_files, + ); + let cli = KafkaCli::new(broker.bootstrap()); + + assert_eq!( + cli.read_records(&topic, 0, 0, values.len()) + .into_iter() + .map(|record| record.value) + .collect::>(), + values + ); +}