refactor: Vendor Java properties writer into framework module

Add a vendored framework module (mirroring stackable_operator::v2) and move the
Java .properties writer into framework/writer.rs, backed by the same
java-properties crate as product_config::writer so the rendered output is
byte-identical. Swap the zoo.cfg / security.properties rendering in
zk_controller from product_config::writer to the local writer.

This removes the first product-config touch point. Properties only; the Hadoop
XML writer from the hdfs-operator source is not needed here.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: maltesander

Cargo.lock

@@ -20,6 +20,7 @@ const_format = "0.2"
 fnv = "1.0"
 futures = { version = "0.3", features = ["compat"] }
 indoc = "2.0"
+java-properties = "2.0"
 pin-project = "1.1"
 semver = "1.0"
 serde = { version = "1.0", features = ["derive"] }

Cargo.toml

@@ -19,6 +19,7 @@ const_format.workspace = true
 fnv.workspace = true
 futures.workspace = true
 indoc.workspace = true
+java-properties.workspace = true
 pin-project.workspace = true
 semver.workspace = true
 serde.workspace = true

rust/operator-binary/Cargo.toml

@@ -0,0 +1,10 @@
+//! Vendored framework code shared between Stackable operators.
+//!
+//! This module mirrors `stackable_operator::v2` (currently only available on the
+//! `smooth-operator` branch of operator-rs). It is vendored here so the
+//! zookeeper-operator can act as a second consumer that validates whether the
+//! abstractions generalize, before they are promoted into operator-rs proper.
+//! Once the upstream `v2` API has stabilized, these modules should be replaced
+//! by direct usage of `stackable_operator::v2`.
+
+pub mod writer;

rust/operator-binary/src/framework.rs

@@ -0,0 +1,81 @@
+//! Writer for Java `.properties` files (e.g. ZooKeeper's `zoo.cfg` and
+//! `security.properties`).
+//!
+//! Vendored from the `product-config` crate's `writer` module so the operator no
+//! longer depends on `product-config` for rendering. Backed by the same
+//! `java-properties` crate, so the rendered output is byte-identical to the
+//! previous `product_config::writer::to_java_properties_string`.
+
+use std::io::Write;
+
+use java_properties::{PropertiesError, PropertiesWriter};
+use snafu::{ResultExt, Snafu};
+
+#[derive(Debug, Snafu)]
+pub enum PropertiesWriterError {
+    #[snafu(display("failed to create properties file"))]
+    Properties { source: PropertiesError },
+
+    #[snafu(display("failed to convert properties file byte array to UTF-8"))]
+    FromUtf8 { source: std::string::FromUtf8Error },
+}
+
+/// Creates a common Java properties file string in the format:
+/// `property_1=value_1\nproperty_2=value_2\n`.
+pub fn to_java_properties_string<'a, T>(properties: T) -> Result<String, PropertiesWriterError>
+where
+    T: Iterator<Item = (&'a String, &'a Option<String>)>,
+{
+    let mut output = Vec::new();
+    write_java_properties(&mut output, properties)?;
+    String::from_utf8(output).context(FromUtf8Snafu)
+}
+
+/// Writes Java properties to the given writer. A `None` value is written as an
+/// empty value (`key=`).
+fn write_java_properties<'a, W, T>(writer: W, properties: T) -> Result<(), PropertiesWriterError>
+where
+    W: Write,
+    T: Iterator<Item = (&'a String, &'a Option<String>)>,
+{
+    let mut writer = PropertiesWriter::new(writer);
+    for (k, v) in properties {
+        let property_value = v.as_deref().unwrap_or_default();
+        writer.write(k, property_value).context(PropertiesSnafu)?;
+    }
+    writer.flush().context(PropertiesSnafu)?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use super::*;
+
+    fn props(pairs: &[(&str, Option<&str>)]) -> String {
+        let map: BTreeMap<String, Option<String>> = pairs
+            .iter()
+            .map(|(k, v)| (k.to_string(), v.map(str::to_string)))
+            .collect();
+        to_java_properties_string(map.iter()).unwrap()
+    }
+
+    #[test]
+    fn java_properties_renders_key_value() {
+        assert_eq!(props(&[("a", Some("1")), ("b", Some("2"))]), "a=1\nb=2\n");
+    }
+
+    #[test]
+    fn java_properties_renders_none_as_empty() {
+        assert_eq!(props(&[("none", None)]), "none=\n");
+    }
+
+    #[test]
+    fn java_properties_escapes_colon_in_value() {
+        assert_eq!(
+            props(&[("url", Some("file://this/location/file.abc"))]),
+            "url=file\\://this/location/file.abc\n"
+        );
+    }
+}

rust/operator-binary/src/framework/writer.rs

@@ -43,6 +43,7 @@ mod command;
 mod config;
 pub mod crd;
 mod discovery;
+mod framework;
 mod listener;
 mod operations;
 mod product_logging;

rust/operator-binary/src/main.rs

@@ -10,11 +10,7 @@ use std::{
 use const_format::concatcp;
 use fnv::FnvHasher;
 use indoc::formatdoc;
-use product_config::{
-    ProductConfigManager,
-    types::PropertyNameKind,
-    writer::{PropertiesWriterError, to_java_properties_string},
-};
+use product_config::{ProductConfigManager, types::PropertyNameKind};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
     builder::{
@@ -87,6 +83,7 @@ use crate::{
         v1alpha1::{self, ZookeeperServerRoleConfig},
     },
     discovery::{self, build_discovery_configmap},
+    framework::writer::{PropertiesWriterError, to_java_properties_string},
     listener::{build_role_listener, role_listener_name},
     operations::{graceful_shutdown::add_graceful_shutdown_config, pdb::add_pdbs},
     product_logging::extend_role_group_config_map,