diff --git a/CHANGELOG.md b/CHANGELOG.md
index 65e054ca..3c50a6c5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -32,6 +32,9 @@ All notable changes to this project will be documented in this file.
   `app.kubernetes.io/role-group: none` labels.
   StatefulSet selectors and volume claim templates are unchanged, so upgrading is non-breaking.
 - Make operations infallible where dependent on static inputs ([#824], [#829]).
+- Internal operator refactoring: the validated cluster carries each role's configuration in typed
+  per-role fields instead of maps keyed by role, so role-specific values are resolved once and the
+  shared resource builders can no longer be handed another role's configuration ([#830]).
 
 ### Fixed
 
@@ -40,6 +43,12 @@ All notable changes to this project will be documented in this file.
   See [our internal issue](https://github.com/stackabletech/hdfs-operator/issues/626) and [the fix](https://github.com/kube-rs/kube/pull/2042) for details ([#814]).
 - The operator now watches all resources that it creates and early-exits the reconcile action when the
   cluster is marked for deletion ([#821]).
+- A role group that does not set `replicas` is now counted as one replica instead of zero when the
+  replica counts are checked, so such a role group no longer produces a warning event claiming it
+  has zero replicas configured ([#830]).
+- The warning that `dfsReplication` is greater than the number of datanodes is now raised for
+  datanodes. It was previously raised for journalnodes, even though its text refers to datanode
+  replicas, and never for datanodes ([#830]).
 
 [#801]: https://github.com/stackabletech/hdfs-operator/pull/801
 [#806]: https://github.com/stackabletech/hdfs-operator/pull/806
@@ -50,6 +59,7 @@ All notable changes to this project will be documented in this file.
 [#821]: https://github.com/stackabletech/hdfs-operator/pull/821
 [#824]: https://github.com/stackabletech/hdfs-operator/pull/824
 [#829]: https://github.com/stackabletech/hdfs-operator/pull/829
+[#830]: https://github.com/stackabletech/hdfs-operator/pull/830
 [#831]: https://github.com/stackabletech/hdfs-operator/pull/831
 
 ## [26.7.0] - 2026-07-21
diff --git a/rust/operator-binary/src/controller/build/container.rs b/rust/operator-binary/src/controller/build/container.rs
index c3a5e298..45e93d35 100644
--- a/rust/operator-binary/src/controller/build/container.rs
+++ b/rust/operator-binary/src/controller/build/container.rs
@@ -53,6 +53,7 @@ use stackable_operator::{
             STACKABLE_LOG_DIR, ValidatedContainerLogConfigChoice, VectorContainerLogConfig,
             vector_container,
         },
+        role_utils::{JavaCommonConfig, RoleGroupConfig},
         types::{
             common::Port,
             kubernetes::{ConfigMapName, ContainerName, VolumeName},
@@ -64,9 +65,9 @@ use strum::{Display, EnumDiscriminants, IntoStaticStr};
 
 use crate::{
     controller::{
-        ValidatedCluster, ValidatedRoleGroupConfig,
+        ValidatedCluster,
         build::{
-            self,
+            self, ResolvedRoleGroup, RoleGroupResolver, RoleSpecificValues,
             jvm::{self, construct_global_jvm_args, construct_role_specific_jvm_args},
             kerberos::KERBEROS_CONTAINER_PATH,
             properties::product_logging::{
@@ -79,8 +80,7 @@ use crate::{
         },
     },
     crd::{
-        AnyNodeConfig, DataNodeContainer, HdfsNodeRole, HdfsPodRef, NameNodeContainer,
-        UpgradeState,
+        DataNodeConfig, HdfsNodeRole, HdfsPodRef, JournalNodeConfig, NameNodeConfig, UpgradeState,
         constants::{
             DATANODE_ROOT_DATA_DIR_PREFIX, DEFAULT_DATA_NODE_METRICS_PORT,
             DEFAULT_JOURNAL_NODE_METRICS_PORT, DEFAULT_NAME_NODE_METRICS_PORT, LISTENER_VOLUME_DIR,
@@ -92,6 +92,7 @@ use crate::{
             SERVICE_PORT_NAME_RPC, STACKABLE_ROOT_DATA_DIR,
         },
         storage::DataNodeStorageConfig,
+        v1alpha1,
     },
 };
 
@@ -211,35 +212,41 @@ impl ContainerConfig {
     const ZKFC_LOG_VOLUME_MOUNT_NAME: &'static str = "zkfc-log-config";
 
     /// Add all main, side and init containers as well as required volumes to the pod builder.
-    pub fn add_containers_and_volumes(
+    ///
+    /// Every role-specific value is resolved by the caller into `resolved`; the role itself comes
+    /// from `C::ROLE`, the same `C` that produced it.
+    pub fn add_containers_and_volumes<C: RoleGroupResolver>(
         pb: &mut PodBuilder,
         cluster: &ValidatedCluster,
         cluster_info: &KubernetesClusterInfo,
-        role: &HdfsNodeRole,
         role_group_name: &RoleGroupName,
-        rolegroup_config: &ValidatedRoleGroupConfig,
-        labels: &Labels,
+        rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+        resolved: &ResolvedRoleGroup<C>,
     ) -> Result<(), Error> {
+        let role = &C::ROLE;
         let namenode_podrefs = build::pod_refs(cluster, &HdfsNodeRole::Name);
 
         // HDFS main container
         let main_container_config = Self::from(*role);
         let resource_names = cluster.role_group_resource_names(role, role_group_name);
         let object_name = resource_names.qualified_role_group_name().to_string();
-        let merged_config = &rolegroup_config.config;
 
-        pb.add_volumes(main_container_config.volumes(merged_config, &object_name, labels)?)
-            .context(AddVolumeSnafu)?;
+        pb.add_volumes(main_container_config.volumes(
+            &resolved.logging.hdfs,
+            resolved.role.listener_volume(),
+            &object_name,
+        ))
+        .context(AddVolumeSnafu)?;
         pb.add_container(main_container_config.main_container(
             cluster,
             cluster_info,
-            role,
+            &resolved.logging.hdfs,
             rolegroup_config,
-            labels,
+            resolved,
         )?);
 
         // Vector sidecar container.
-        if merged_config.vector_logging_enabled() {
+        if let Some(vector_logging) = &resolved.logging.vector {
             let vector_aggregator_config_map_name = cluster
                 .cluster_config
                 .logging
@@ -247,7 +254,7 @@ impl ContainerConfig {
                 .clone()
                 .context(VectorAggregatorConfigMapMissingSnafu)?;
 
-            let log_config = match &*merged_config.vector_logging() {
+            let log_config = match vector_logging {
                 ContainerLogConfig {
                     choice:
                         Some(ContainerLogConfigChoice::Custom(CustomContainerLogConfig {
@@ -296,8 +303,9 @@ impl ContainerConfig {
                         .with_format(SecretFormat::TlsPkcs12)
                         .with_tls_pkcs12_password(TLS_STORE_PASSWORD)
                         .with_auto_tls_cert_lifetime(
-                            merged_config
-                                .requested_secret_lifetime()
+                            resolved
+                                .common
+                                .requested_secret_lifetime
                                 .context(MissingSecretLifetimeSnafu)?,
                         )
                         .build()
@@ -330,126 +338,152 @@ impl ContainerConfig {
             .context(AddVolumeSnafu)?;
         }
 
-        // role specific pod settings configured here
-        match role {
-            HdfsNodeRole::Name => {
+        // The role-specific containers and their log configs come from one enum, so a container
+        // is never built without the log config that belongs to it.
+        match &resolved.role {
+            RoleSpecificValues::Journal => {}
+            RoleSpecificValues::Name {
+                zkfc,
+                format_namenodes,
+                format_zookeeper,
+            } => {
                 // Zookeeper fail over container
                 let zkfc_container_config = Self::Zkfc;
-                pb.add_volumes(zkfc_container_config.volumes(
-                    merged_config,
-                    &object_name,
-                    labels,
-                )?)
-                .context(AddVolumeSnafu)?;
+                pb.add_volumes(zkfc_container_config.volumes(zkfc, None, &object_name))
+                    .context(AddVolumeSnafu)?;
                 pb.add_container(zkfc_container_config.main_container(
                     cluster,
                     cluster_info,
-                    role,
+                    zkfc,
                     rolegroup_config,
-                    labels,
+                    resolved,
                 )?);
 
                 // Format namenode init container
                 let format_namenodes_container_config = Self::FormatNameNodes;
                 pb.add_volumes(format_namenodes_container_config.volumes(
-                    merged_config,
+                    format_namenodes,
+                    None,
                     &object_name,
-                    labels,
-                )?)
+                ))
                 .context(AddVolumeSnafu)?;
                 pb.add_init_container(format_namenodes_container_config.init_container(
                     cluster,
                     cluster_info,
-                    role,
+                    format_namenodes,
                     rolegroup_config,
+                    resolved,
                     &namenode_podrefs,
-                    labels,
                 )?);
 
                 // Format ZooKeeper init container
                 let format_zookeeper_container_config = Self::FormatZooKeeper;
                 pb.add_volumes(format_zookeeper_container_config.volumes(
-                    merged_config,
+                    format_zookeeper,
+                    None,
                     &object_name,
-                    labels,
-                )?)
+                ))
                 .context(AddVolumeSnafu)?;
                 pb.add_init_container(format_zookeeper_container_config.init_container(
                     cluster,
                     cluster_info,
-                    role,
+                    format_zookeeper,
                     rolegroup_config,
+                    resolved,
                     &namenode_podrefs,
-                    labels,
                 )?);
             }
-            HdfsNodeRole::Data => {
+            RoleSpecificValues::Data {
+                wait_for_namenodes, ..
+            } => {
                 // Wait for namenode init container
                 let wait_for_namenodes_container_config = Self::WaitForNameNodes;
                 pb.add_volumes(wait_for_namenodes_container_config.volumes(
-                    merged_config,
+                    wait_for_namenodes,
+                    None,
                     &object_name,
-                    labels,
-                )?)
+                ))
                 .context(AddVolumeSnafu)?;
                 pb.add_init_container(wait_for_namenodes_container_config.init_container(
                     cluster,
                     cluster_info,
-                    role,
+                    wait_for_namenodes,
                     rolegroup_config,
+                    resolved,
                     &namenode_podrefs,
-                    labels,
                 )?);
             }
-            HdfsNodeRole::Journal => {}
         }
 
         Ok(())
     }
 
-    pub fn volume_claim_templates(
-        merged_config: &AnyNodeConfig,
+    /// The PVC templates for a namenode role group: one data PVC plus the listener PVC.
+    pub fn namenode_volume_claim_templates(
+        config: &NameNodeConfig,
         labels: &Labels,
     ) -> Result<Vec<PersistentVolumeClaim>> {
-        match merged_config {
-            AnyNodeConfig::Name(node) => {
-                let listener = ListenerOperatorVolumeSourceBuilder::new(
-                    &ListenerReference::ListenerClass(node.listener_class.to_string()),
-                    labels,
-                )
-                .build_ephemeral()
-                .context(BuildListenerVolumeSnafu)?
-                .volume_claim_template
-                .expect("The listener volume source builder always sets a volume claim template.");
-
-                let pvcs = vec![
-                    node.resources.storage.data.build_pvc(
-                        ContainerConfig::DATA_VOLUME_MOUNT_NAME,
-                        Some(vec!["ReadWriteOnce"]),
-                    ),
-                    PersistentVolumeClaim {
-                        metadata: ObjectMeta {
-                            name: Some(LISTENER_VOLUME_NAME.to_string()),
-                            ..listener.metadata.expect(
-                                "The listener volume claim template always carries metadata.",
-                            )
-                        },
-                        spec: Some(listener.spec),
-                        ..Default::default()
-                    },
-                ];
+        let listener = ListenerOperatorVolumeSourceBuilder::new(
+            &ListenerReference::ListenerClass(config.listener_class.to_string()),
+            labels,
+        )
+        .build_ephemeral()
+        .context(BuildListenerVolumeSnafu)?
+        .volume_claim_template
+        .expect("The listener volume source builder always sets a volume claim template.");
 
-                Ok(pvcs)
-            }
-            AnyNodeConfig::Journal(node) => Ok(vec![node.resources.storage.data.build_pvc(
+        Ok(vec![
+            config.resources.storage.data.build_pvc(
                 ContainerConfig::DATA_VOLUME_MOUNT_NAME,
                 Some(vec!["ReadWriteOnce"]),
-            )]),
-            AnyNodeConfig::Data(node) => Ok(DataNodeStorageConfig {
-                pvcs: node.resources.storage.clone(),
-            }
-            .build_pvcs()),
+            ),
+            PersistentVolumeClaim {
+                metadata: ObjectMeta {
+                    name: Some(LISTENER_VOLUME_NAME.to_string()),
+                    ..listener
+                        .metadata
+                        .expect("The listener volume claim template always carries metadata.")
+                },
+                spec: Some(listener.spec),
+                ..Default::default()
+            },
+        ])
+    }
+
+    /// The PVC template for a journalnode role group: one data PVC.
+    pub fn journalnode_volume_claim_templates(
+        config: &JournalNodeConfig,
+    ) -> Vec<PersistentVolumeClaim> {
+        vec![config.resources.storage.data.build_pvc(
+            ContainerConfig::DATA_VOLUME_MOUNT_NAME,
+            Some(vec!["ReadWriteOnce"]),
+        )]
+    }
+
+    /// The PVC templates for a datanode role group, one per configured data volume.
+    pub fn datanode_volume_claim_templates(config: &DataNodeConfig) -> Vec<PersistentVolumeClaim> {
+        DataNodeStorageConfig {
+            pvcs: config.resources.storage.clone(),
         }
+        .build_pvcs()
+    }
+
+    /// The ephemeral listener volume of a datanode role group.
+    ///
+    /// Datanodes use an ephemeral listener volume, while namenodes use a persistent volume claim
+    /// template for stable per-pod identity (see [`Self::namenode_volume_claim_templates`]) and
+    /// journalnodes have no listener at all.
+    pub fn datanode_listener_volume(config: &DataNodeConfig, labels: &Labels) -> Result<Volume> {
+        Ok(VolumeBuilder::new(&*LISTENER_VOLUME_NAME)
+            .ephemeral(
+                ListenerOperatorVolumeSourceBuilder::new(
+                    &ListenerReference::ListenerClass(config.listener_class.to_string()),
+                    labels,
+                )
+                .build_ephemeral()
+                .context(BuildListenerVolumeSnafu)?,
+            )
+            .build())
     }
 
     /// Creates the main/side containers for:
@@ -457,24 +491,24 @@ impl ContainerConfig {
     /// - Namenode ZooKeeper fail over controller (ZKFC)
     /// - Datanode main process
     /// - Journalnode main process
-    fn main_container(
+    fn main_container<C: RoleGroupResolver>(
         &self,
         cluster: &ValidatedCluster,
         cluster_info: &KubernetesClusterInfo,
-        role: &HdfsNodeRole,
-        rolegroup_config: &ValidatedRoleGroupConfig,
-        labels: &Labels,
+        container_log_config: &ContainerLogConfig,
+        rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+        resolved: &ResolvedRoleGroup<C>,
     ) -> Result<Container, Error> {
-        let merged_config = &rolegroup_config.config;
+        let role = &C::ROLE;
         let mut cb = new_container_builder(self.container_name());
 
-        let resources = self.resources(merged_config);
+        let resources = self.resources(&resolved.resources);
 
         cb.image_from_product_image(&cluster.image)
             .command(Self::command())
-            .args(self.args(cluster, cluster_info, role, merged_config, &[])?)
+            .args(self.args(cluster, cluster_info, role, container_log_config, &[])?)
             .add_env_vars(self.env(cluster, role, rolegroup_config, resources.as_ref())?)
-            .add_volume_mounts(self.volume_mounts(cluster, merged_config, labels)?)
+            .add_volume_mounts(self.volume_mounts(cluster, &resolved.volume_claim_templates))
             .context(AddVolumeMountSnafu)?
             .add_container_ports(self.container_ports(cluster));
 
@@ -505,29 +539,35 @@ impl ContainerConfig {
     /// Creates respective init containers for:
     /// - Namenode (format-namenodes, format-zookeeper)
     /// - Datanode (wait-for-namenodes)
-    fn init_container(
+    fn init_container<C: RoleGroupResolver>(
         &self,
         cluster: &ValidatedCluster,
         cluster_info: &KubernetesClusterInfo,
-        role: &HdfsNodeRole,
-        rolegroup_config: &ValidatedRoleGroupConfig,
+        container_log_config: &ContainerLogConfig,
+        rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+        resolved: &ResolvedRoleGroup<C>,
         namenode_podrefs: &[HdfsPodRef],
-        labels: &Labels,
     ) -> Result<Container, Error> {
-        let merged_config = &rolegroup_config.config;
+        let role = &C::ROLE;
         let mut cb = new_container_builder(self.container_name());
 
         cb.image_from_product_image(&cluster.image)
             .command(Self::command())
-            .args(self.args(cluster, cluster_info, role, merged_config, namenode_podrefs)?)
+            .args(self.args(
+                cluster,
+                cluster_info,
+                role,
+                container_log_config,
+                namenode_podrefs,
+            )?)
             .add_env_vars(self.env(cluster, role, rolegroup_config, None)?)
-            .add_volume_mounts(self.volume_mounts(cluster, merged_config, labels)?)
+            .add_volume_mounts(self.volume_mounts(cluster, &resolved.volume_claim_templates))
             .context(AddVolumeMountSnafu)?;
 
         // We use the main app container resources here in contrast to several operators (which use
         // hardcoded resources) due to the different code structure.
         // Going forward this should be replaced by calculating init container resources in the pod builder.
-        if let Some(resources) = self.resources(merged_config) {
+        if let Some(resources) = self.resources(&resolved.resources) {
             cb.resources(resources);
         }
 
@@ -598,7 +638,7 @@ impl ContainerConfig {
         cluster: &ValidatedCluster,
         cluster_info: &KubernetesClusterInfo,
         role: &HdfsNodeRole,
-        merged_config: &AnyNodeConfig,
+        container_log_config: &ContainerLogConfig,
         namenode_podrefs: &[HdfsPodRef],
     ) -> Result<Vec<String>, Error> {
         let mut args = String::new();
@@ -620,10 +660,9 @@ impl ContainerConfig {
 
         match self {
             ContainerConfig::Hdfs { role, .. } => {
-                args.push_str(&self.copy_log4j_properties_cmd(
-                    HDFS_LOG4J_CONFIG_FILE,
-                    &merged_config.hdfs_logging(),
-                ));
+                args.push_str(
+                    &self.copy_log4j_properties_cmd(HDFS_LOG4J_CONFIG_FILE, container_log_config),
+                );
 
                 args.push_str(&formatdoc!(
                     r#"\
@@ -650,14 +689,9 @@ impl ContainerConfig {
                 ));
             }
             ContainerConfig::Zkfc => {
-                if let Some(container_config) = merged_config
-                    .as_namenode()
-                    .map(|node| node.logging.for_container(&NameNodeContainer::Zkfc))
-                {
-                    args.push_str(
-                        &self.copy_log4j_properties_cmd(ZKFC_LOG4J_CONFIG_FILE, &container_config),
-                    );
-                }
+                args.push_str(
+                    &self.copy_log4j_properties_cmd(ZKFC_LOG4J_CONFIG_FILE, container_log_config),
+                );
                 args.push_str(&format!(
                     "{hadoop_home}/bin/hdfs zkfc\n",
                     hadoop_home = Self::HADOOP_HOME
@@ -666,15 +700,10 @@ impl ContainerConfig {
             ContainerConfig::FormatNameNodes => {
                 args.push_str(&bash_capture_shell_helper(self.container_name().as_ref()));
 
-                if let Some(container_config) = merged_config.as_namenode().map(|node| {
-                    node.logging
-                        .for_container(&NameNodeContainer::FormatNameNodes)
-                }) {
-                    args.push_str(&self.copy_log4j_properties_cmd(
-                        FORMAT_NAMENODES_LOG4J_CONFIG_FILE,
-                        &container_config,
-                    ));
-                }
+                args.push_str(&self.copy_log4j_properties_cmd(
+                    FORMAT_NAMENODES_LOG4J_CONFIG_FILE,
+                    container_log_config,
+                ));
                 // First step we check for active namenodes. This step should return an active namenode
                 // for e.g. scaling. It may fail if the active namenode is restarted and the standby
                 // namenode takes over.
@@ -742,15 +771,10 @@ impl ContainerConfig {
             ContainerConfig::FormatZooKeeper => {
                 args.push_str(&bash_capture_shell_helper(self.container_name().as_ref()));
 
-                if let Some(container_config) = merged_config.as_namenode().map(|node| {
-                    node.logging
-                        .for_container(&NameNodeContainer::FormatZooKeeper)
-                }) {
-                    args.push_str(&self.copy_log4j_properties_cmd(
-                        FORMAT_ZOOKEEPER_LOG4J_CONFIG_FILE,
-                        &container_config,
-                    ));
-                }
+                args.push_str(&self.copy_log4j_properties_cmd(
+                    FORMAT_ZOOKEEPER_LOG4J_CONFIG_FILE,
+                    container_log_config,
+                ));
                 args.push_str(&formatdoc!(
                     r###"
                     echo "Attempt to format ZooKeeper ZNode for $POD_NAME ..."
@@ -774,15 +798,10 @@ impl ContainerConfig {
             ContainerConfig::WaitForNameNodes => {
                 args.push_str(&bash_capture_shell_helper(self.container_name().as_ref()));
 
-                if let Some(container_config) = merged_config.as_datanode().map(|node| {
-                    node.logging
-                        .for_container(&DataNodeContainer::WaitForNameNodes)
-                }) {
-                    args.push_str(&self.copy_log4j_properties_cmd(
-                        WAIT_FOR_NAMENODES_LOG4J_CONFIG_FILE,
-                        &container_config,
-                    ));
-                }
+                args.push_str(&self.copy_log4j_properties_cmd(
+                    WAIT_FOR_NAMENODES_LOG4J_CONFIG_FILE,
+                    container_log_config,
+                ));
                 if cluster.has_kerberos_enabled() {
                     args.push_str(&Self::get_kerberos_ticket(cluster, role, cluster_info)?);
                 }
@@ -862,11 +881,11 @@ impl ContainerConfig {
     }
 
     /// Returns the container env variables.
-    fn env(
+    fn env<C>(
         &self,
         cluster: &ValidatedCluster,
         role: &HdfsNodeRole,
-        rolegroup_config: &ValidatedRoleGroupConfig,
+        rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
         resources: Option<&ResourceRequirements>,
     ) -> Result<Vec<EnvVar>, Error> {
         // Maps env var name to env var object. This allows env_overrides to work
@@ -973,8 +992,13 @@ impl ContainerConfig {
         Ok(env.into_values().collect())
     }
 
-    /// Returns the container resources.
-    pub fn resources(&self, merged_config: &AnyNodeConfig) -> Option<ResourceRequirements> {
+    /// Returns the container resources. `role_group_resources` is the role group's own
+    /// `resources`, already converted, and is used by the main and init containers; the ZKFC
+    /// sidecar has fixed requirements of its own.
+    pub fn resources(
+        &self,
+        role_group_resources: &ResourceRequirements,
+    ) -> Option<ResourceRequirements> {
         match self {
             // Namenode sidecar containers
             ContainerConfig::Zkfc => Some(
@@ -989,11 +1013,7 @@ impl ContainerConfig {
             ContainerConfig::Hdfs { .. }
             | ContainerConfig::FormatNameNodes
             | ContainerConfig::FormatZooKeeper
-            | ContainerConfig::WaitForNameNodes => match merged_config {
-                AnyNodeConfig::Name(node) => Some(node.resources.clone().into()),
-                AnyNodeConfig::Data(node) => Some(node.resources.clone().into()),
-                AnyNodeConfig::Journal(node) => Some(node.resources.clone().into()),
-            },
+            | ContainerConfig::WaitForNameNodes => Some(role_group_resources.clone()),
         }
     }
 
@@ -1057,29 +1077,20 @@ impl ContainerConfig {
     }
 
     /// Return the container volumes.
+    ///
+    /// `container_log_config` is this container's own, chosen by the caller from
+    /// [`build::RoleGroupLogging`] or [`RoleSpecificValues`]. `listener_volume` is the role
+    /// group's ephemeral listener volume, which only the datanode main container has.
     fn volumes(
         &self,
-        merged_config: &AnyNodeConfig,
+        container_log_config: &ContainerLogConfig,
+        listener_volume: Option<&Volume>,
         object_name: &str,
-        labels: &Labels,
-    ) -> Result<Vec<Volume>> {
+    ) -> Vec<Volume> {
         let mut volumes = vec![];
 
         if let ContainerConfig::Hdfs { .. } = self {
-            if let AnyNodeConfig::Data(node) = merged_config {
-                volumes.push(
-                    VolumeBuilder::new(&*LISTENER_VOLUME_NAME)
-                        .ephemeral(
-                            ListenerOperatorVolumeSourceBuilder::new(
-                                &ListenerReference::ListenerClass(node.listener_class.to_string()),
-                                labels,
-                            )
-                            .build_ephemeral()
-                            .context(BuildListenerVolumeSnafu)?,
-                        )
-                        .build(),
-                );
-            }
+            volumes.extend(listener_volume.cloned());
 
             volumes.push(
                 VolumeBuilder::new(ContainerConfig::STACKABLE_LOG_VOLUME_MOUNT_NAME)
@@ -1099,42 +1110,26 @@ impl ContainerConfig {
             );
         }
 
-        let container_log_config = match self {
-            ContainerConfig::Hdfs { .. } => Some(merged_config.hdfs_logging()),
-            ContainerConfig::Zkfc => merged_config
-                .as_namenode()
-                .map(|node| node.logging.for_container(&NameNodeContainer::Zkfc)),
-            ContainerConfig::FormatNameNodes => merged_config.as_namenode().map(|node| {
-                node.logging
-                    .for_container(&NameNodeContainer::FormatNameNodes)
-            }),
-            ContainerConfig::FormatZooKeeper => merged_config.as_namenode().map(|node| {
-                node.logging
-                    .for_container(&NameNodeContainer::FormatZooKeeper)
-            }),
-            ContainerConfig::WaitForNameNodes => merged_config.as_datanode().map(|node| {
-                node.logging
-                    .for_container(&DataNodeContainer::WaitForNameNodes)
-            }),
-        };
         let volume_mount_dirs = self.volume_mount_dirs();
         volumes.extend(Self::common_container_volumes(
-            container_log_config.as_deref(),
+            Some(container_log_config),
             object_name,
             volume_mount_dirs.config_mount_name(),
             volume_mount_dirs.log_mount_name(),
         ));
 
-        Ok(volumes)
+        volumes
     }
 
     /// Returns the container volume mounts.
+    ///
+    /// `volume_claim_templates` are the role group's PVC templates; the datanode main container
+    /// mounts one data directory per data PVC, named after it.
     fn volume_mounts(
         &self,
         cluster: &ValidatedCluster,
-        merged_config: &AnyNodeConfig,
-        labels: &Labels,
-    ) -> Result<Vec<VolumeMount>> {
+        volume_claim_templates: &[PersistentVolumeClaim],
+    ) -> Vec<VolumeMount> {
         let volume_mount_dirs = self.volume_mount_dirs();
         let mut volume_mounts = vec![
             VolumeMountBuilder::new(Self::STACKABLE_LOG_VOLUME_MOUNT_NAME, STACKABLE_LOG_DIR)
@@ -1192,7 +1187,7 @@ impl ContainerConfig {
                         );
                     }
                     HdfsNodeRole::Data => {
-                        for pvc in Self::volume_claim_templates(merged_config, labels)? {
+                        for pvc in volume_claim_templates {
                             let pvc_name = pvc.name_any();
                             volume_mounts.push(VolumeMount {
                                 mount_path: format!("{DATANODE_ROOT_DATA_DIR_PREFIX}{pvc_name}"),
@@ -1209,7 +1204,7 @@ impl ContainerConfig {
             | ContainerConfig::FormatZooKeeper => {}
         }
 
-        Ok(volume_mounts)
+        volume_mounts
     }
 
     /// Create a config directory for the respective container.
@@ -1258,11 +1253,11 @@ impl ContainerConfig {
     }
 
     /// Build HADOOP_{*node}_OPTS for each namenode, datanodes and journalnodes.
-    fn build_hadoop_opts(
+    fn build_hadoop_opts<C>(
         &self,
         cluster: &ValidatedCluster,
         resources: Option<&ResourceRequirements>,
-        rolegroup_config: &ValidatedRoleGroupConfig,
+        rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
     ) -> Result<String, Error> {
         match self {
             ContainerConfig::Hdfs {
@@ -1524,6 +1519,7 @@ mod tests {
     use strum::IntoEnumIterator;
 
     use super::*;
+    use crate::crd::{DataNodeContainer, NameNodeContainer};
 
     #[test]
     fn test_constants() {
diff --git a/rust/operator-binary/src/controller/build/jvm.rs b/rust/operator-binary/src/controller/build/jvm.rs
index 54d9d0bd..640ed4de 100644
--- a/rust/operator-binary/src/controller/build/jvm.rs
+++ b/rust/operator-binary/src/controller/build/jvm.rs
@@ -98,7 +98,10 @@ mod tests {
     use crate::{
         controller::build::container::ContainerConfig,
         crd::constants::DEFAULT_NAME_NODE_METRICS_PORT,
-        test_support::{deserialize_and_validate_cluster, role_group_config, role_group_name},
+        test_support::{
+            deserialize_and_validate_cluster, namenode_config, namenode_role_group_config,
+            role_group_name,
+        },
     };
 
     #[test]
@@ -197,10 +200,11 @@ mod tests {
         let role = HdfsNodeRole::Name;
 
         let validated_cluster = deserialize_and_validate_cluster(hdfs_cluster);
-        let role_group_config =
-            role_group_config(&validated_cluster, &role, &role_group_name("default"));
-
-        let resources = ContainerConfig::from(role).resources(&role_group_config.config);
+        let role_group_name = role_group_name("default");
+        let role_group_config = namenode_role_group_config(&validated_cluster, &role_group_name);
+        let namenode_config = namenode_config(&validated_cluster, &role_group_name);
+        let resources =
+            ContainerConfig::from(role).resources(&namenode_config.resources.clone().into());
 
         construct_role_specific_jvm_args(
             &role,
diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs
index 7f4ff4fc..2eeb36d5 100644
--- a/rust/operator-binary/src/controller/build/mod.rs
+++ b/rust/operator-binary/src/controller/build/mod.rs
@@ -1,13 +1,22 @@
-use std::{collections::HashMap, marker::PhantomData};
+use std::{
+    collections::{BTreeMap, HashMap},
+    marker::PhantomData,
+};
 
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::meta::ObjectMetaBuilder,
+    k8s_openapi::api::{
+        apps::v1::StatefulSet,
+        core::v1::{ConfigMap, Service},
+        policy::v1::PodDisruptionBudget,
+    },
     kvp::{LabelError, Labels},
     utils::cluster_info::KubernetesClusterInfo,
     v2::{
         builder::meta::ownerreference_from_resource,
         kvp::label,
+        role_utils::{JavaCommonConfig, RoleGroupConfig},
         types::{
             common::Port,
             operator::{RoleGroupName, RoleName},
@@ -38,6 +47,7 @@ use crate::{
             SERVICE_PORT_NAME_IPC, SERVICE_PORT_NAME_JMX_METRICS, SERVICE_PORT_NAME_METRICS,
             SERVICE_PORT_NAME_RPC,
         },
+        v1alpha1,
     },
 };
 
@@ -47,6 +57,7 @@ pub mod jvm;
 pub mod kerberos;
 pub mod opa;
 pub mod properties;
+pub mod resolve;
 pub mod resource;
 
 #[derive(Snafu, Debug)]
@@ -74,6 +85,100 @@ pub enum Error {
 
     #[snafu(display("failed to build the discovery ConfigMap"))]
     DiscoveryConfigMap { source: resource::discovery::Error },
+
+    #[snafu(display("failed to build selector labels for role {role} role group {role_group}", role = role.as_ref()))]
+    RoleGroupSelectorLabels {
+        source: LabelError,
+        role: HdfsNodeRole,
+        role_group: RoleGroupName,
+    },
+
+    #[snafu(display("failed to build volume claim templates for role {role} role group {role_group}", role = role.as_ref()))]
+    VolumeClaimTemplates {
+        source: container::Error,
+        role: HdfsNodeRole,
+        role_group: RoleGroupName,
+    },
+
+    #[snafu(display("failed to build listener volume for role {role} role group {role_group}", role = role.as_ref()))]
+    ListenerVolume {
+        source: container::Error,
+        role: HdfsNodeRole,
+        role_group: RoleGroupName,
+    },
+}
+
+pub(crate) use resolve::RoleGroupResolver;
+pub use resolve::{ResolvedRoleGroup, RoleGroupLogging, RoleSpecificValues};
+
+/// The resources of every role, accumulated one role at a time by [`build_role`].
+#[derive(Default)]
+struct RoleGroupResources {
+    services: Vec<Service>,
+    config_maps: Vec<ConfigMap>,
+    /// Keyed by role, so flattening the map yields the StatefulSets in the rollout order
+    /// [`HdfsNodeRole`]'s variant order defines, whatever order the roles were built in.
+    stateful_sets: BTreeMap<HdfsNodeRole, Vec<StatefulSet>>,
+    pod_disruption_budgets: Vec<PodDisruptionBudget>,
+}
+
+/// Builds every resource of every role group of one role, plus that role's PDB, appending them to
+/// `rg_resources`.
+fn build_role<C: RoleGroupResolver>(
+    cluster: &ValidatedCluster,
+    cluster_info: &KubernetesClusterInfo,
+    role_group_configs: &BTreeMap<
+        RoleGroupName,
+        RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    >,
+    rg_resources: &mut RoleGroupResources,
+) -> Result<(), Error> {
+    let role = &C::ROLE;
+
+    for (role_group_name, rg_config) in role_group_configs {
+        build_role_group_services(cluster, role, role_group_name, &mut rg_resources.services)?;
+
+        let selector_labels = rolegroup_selector_labels(cluster, role, role_group_name).context(
+            RoleGroupSelectorLabelsSnafu {
+                role: *role,
+                role_group: role_group_name.clone(),
+            },
+        )?;
+        let resolved = rg_config.config.resolve(role_group_name, selector_labels)?;
+
+        rg_resources.config_maps.push(
+            resource::config_map::build_rolegroup_config_map(
+                cluster,
+                cluster_info,
+                role_group_name,
+                rg_config,
+                &resolved,
+            )
+            .context(ConfigMapSnafu {
+                role: *role,
+                role_group: role_group_name.clone(),
+            })?,
+        );
+        rg_resources.stateful_sets.entry(C::ROLE).or_default().push(
+            resource::statefulset::build_rolegroup_statefulset(
+                cluster,
+                cluster_info,
+                role_group_name,
+                rg_config,
+                &resolved,
+            )
+            .context(StatefulSetSnafu {
+                role: *role,
+                role_group: role_group_name.clone(),
+            })?,
+        );
+    }
+
+    if let Some(pdb) = resource::pdb::build_pdb(cluster, role) {
+        rg_resources.pod_disruption_budgets.push(pdb);
+    }
+
+    Ok(())
 }
 
 /// Builds every Kubernetes resource for the given validated cluster.
@@ -83,8 +188,9 @@ pub enum Error {
 /// `cluster_info` carries static cluster information resolved at operator startup (e.g. the
 /// cluster domain used to build Kerberos principals), not a live client.
 ///
-/// The resources are returned as flat, unordered collections. The reconcile step re-groups the
-/// StatefulSets by role to preserve HDFS's ordered, rollout-gated deployment during upgrades.
+/// The resources are returned as flat collections. `stateful_sets` comes out in [`HdfsNodeRole`]
+/// order, which the apply step depends on; that is structural, from a [`BTreeMap`] flattened in
+/// key order, not from the order the roles are built in.
 /// The discovery `ConfigMap` is included when it can be built or re-emitted (see
 /// [`resource::discovery::build_discovery_config_map`]); it is only absent before its first
 /// successful build.
@@ -92,58 +198,35 @@ pub fn build(
     cluster: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
 ) -> Result<KubernetesResources<Prepared>, Error> {
-    let mut services = vec![];
-    let mut config_maps = vec![];
-    let mut stateful_sets = vec![];
-    let mut pod_disruption_budgets = vec![];
-
-    for (role, role_group_configs) in &cluster.role_groups {
-        for (role_group_name, rg_config) in role_group_configs {
-            services.push(
-                resource::service::rolegroup_headless_service(cluster, role, role_group_name)
-                    .context(ServiceSnafu {
-                        role: *role,
-                        role_group: role_group_name.clone(),
-                    })?,
-            );
-            services.push(
-                resource::service::rolegroup_metrics_service(cluster, role, role_group_name)
-                    .context(ServiceSnafu {
-                        role: *role,
-                        role_group: role_group_name.clone(),
-                    })?,
-            );
-            config_maps.push(
-                resource::config_map::build_rolegroup_config_map(
-                    cluster,
-                    cluster_info,
-                    role,
-                    role_group_name,
-                )
-                .context(ConfigMapSnafu {
-                    role: *role,
-                    role_group: role_group_name.clone(),
-                })?,
-            );
-            stateful_sets.push(
-                resource::statefulset::build_rolegroup_statefulset(
-                    cluster,
-                    cluster_info,
-                    role,
-                    role_group_name,
-                    rg_config,
-                )
-                .context(StatefulSetSnafu {
-                    role: *role,
-                    role_group: role_group_name.clone(),
-                })?,
-            );
-        }
+    let mut built = RoleGroupResources::default();
 
-        if let Some(pdb) = resource::pdb::build_pdb(cluster, role) {
-            pod_disruption_budgets.push(pdb);
-        }
-    }
+    // These three calls are free to be reordered: the StatefulSets are keyed by role, and the
+    // apply step does not depend on the order of the other three collections.
+    build_role(
+        cluster,
+        cluster_info,
+        &cluster.journalnode_role_group_configs,
+        &mut built,
+    )?;
+    build_role(
+        cluster,
+        cluster_info,
+        &cluster.namenode_role_group_configs,
+        &mut built,
+    )?;
+    build_role(
+        cluster,
+        cluster_info,
+        &cluster.datanode_role_group_configs,
+        &mut built,
+    )?;
+
+    let RoleGroupResources {
+        services,
+        mut config_maps,
+        stateful_sets,
+        pod_disruption_budgets,
+    } = built;
 
     // The discovery ConfigMap is skipped only before its first successful build (no namenode
     // Listener addresses yet, nothing stored to re-emit); afterwards a stored ConfigMap is
@@ -159,13 +242,77 @@ pub fn build(
         services,
         config_maps,
         pod_disruption_budgets,
-        stateful_sets,
+        // `BTreeMap` iterates in key order, so this is the rollout order the apply step needs.
+        stateful_sets: stateful_sets.into_values().flatten().collect(),
         service_accounts: vec![build_service_account(cluster)],
         role_bindings: vec![build_role_binding(cluster)],
         status: PhantomData,
     })
 }
 
+/// Builds the two Services for one role group. Role-agnostic: it reads nothing from the role
+/// config.
+fn build_role_group_services(
+    cluster: &ValidatedCluster,
+    role: &HdfsNodeRole,
+    role_group_name: &RoleGroupName,
+    services: &mut Vec<Service>,
+) -> Result<(), Error> {
+    services.push(
+        resource::service::rolegroup_headless_service(cluster, role, role_group_name).context(
+            ServiceSnafu {
+                role: *role,
+                role_group: role_group_name.clone(),
+            },
+        )?,
+    );
+    services.push(
+        resource::service::rolegroup_metrics_service(cluster, role, role_group_name).context(
+            ServiceSnafu {
+                role: *role,
+                role_group: role_group_name.clone(),
+            },
+        )?,
+    );
+
+    Ok(())
+}
+
+/// The replica count a role group gets when it does not set one: Kubernetes runs a single pod for
+/// a `StatefulSet` with `replicas: null`.
+pub(crate) const DEFAULT_REPLICAS: u16 = 1;
+
+/// The replica count of every role group in the map, defaulting to [`DEFAULT_REPLICAS`] where it
+/// is unset. The single place that applies that default.
+fn role_group_replicas<'a, C>(
+    role_group_configs: &'a BTreeMap<
+        RoleGroupName,
+        RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    >,
+) -> impl Iterator<Item = (&'a RoleGroupName, u16)> + 'a {
+    role_group_configs
+        .iter()
+        .map(|(role_group_name, role_group)| {
+            (
+                role_group_name,
+                role_group.replicas.unwrap_or(DEFAULT_REPLICAS),
+            )
+        })
+}
+
+/// The total number of replicas across the role groups of one role, counting a role group without
+/// an explicit replica count as [`DEFAULT_REPLICAS`].
+pub(crate) fn total_replicas<C>(
+    role_group_configs: &BTreeMap<
+        RoleGroupName,
+        RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    >,
+) -> u16 {
+    role_group_replicas(role_group_configs)
+        .map(|(_, replicas)| replicas)
+        .sum()
+}
+
 /// Builds the [`HdfsPodRef`]s expected for every pod of the given `role`, across all
 /// of its role groups.
 ///
@@ -180,17 +327,22 @@ pub(crate) fn pod_refs(cluster: &ValidatedCluster, role: &HdfsNodeRole) -> Vec<H
         .into_iter()
         .collect();
 
-    cluster
-        .role_groups
-        .get(role)
+    let replicas_per_role_group: Vec<(&RoleGroupName, u16)> = match role {
+        HdfsNodeRole::Name => role_group_replicas(&cluster.namenode_role_group_configs).collect(),
+        HdfsNodeRole::Data => role_group_replicas(&cluster.datanode_role_group_configs).collect(),
+        HdfsNodeRole::Journal => {
+            role_group_replicas(&cluster.journalnode_role_group_configs).collect()
+        }
+    };
+
+    replicas_per_role_group
         .into_iter()
-        .flatten()
-        .flat_map(|(role_group_name, role_group)| {
+        .flat_map(|(role_group_name, replicas)| {
             let service_name = cluster.governing_service_name(role, role_group_name);
             let object_name = service_name.to_string();
             let namespace = cluster.namespace.clone();
             let ports = ports.clone();
-            (0..role_group.replicas.unwrap_or(1)).map(move |i| HdfsPodRef {
+            (0..replicas).map(move |i| HdfsPodRef {
                 namespace: namespace.clone(),
                 role_group_service_name: service_name.clone(),
                 pod_name: format!("{object_name}-{i}"),
@@ -255,13 +407,7 @@ pub(crate) fn rolegroup_selector_labels(
 
 /// The total number of datanode replicas across all datanode role groups.
 pub(crate) fn num_datanodes(cluster: &ValidatedCluster) -> u16 {
-    cluster
-        .role_groups
-        .get(&HdfsNodeRole::Data)
-        .into_iter()
-        .flatten()
-        .map(|(_, role_group)| role_group.replicas.unwrap_or(1))
-        .sum()
+    total_replicas(&cluster.datanode_role_group_configs)
 }
 
 /// The ports exposed by the rolegroup headless service for the given `role`.
@@ -432,7 +578,7 @@ pub(crate) fn role_group_selector(
 
 #[cfg(test)]
 mod tests {
-    use stackable_operator::kube::Resource;
+    use stackable_operator::{k8s_openapi::api::core::v1::ConfigMap, kube::Resource};
 
     use super::build;
     use crate::{
@@ -500,6 +646,32 @@ mod tests {
         assert_eq!(sorted_names(&resources.role_bindings), ["hdfs-rolebinding"]);
     }
 
+    /// The StatefulSets must come out in role order — journalnodes, then namenodes, then
+    /// datanodes — because the apply step rolls them out in exactly that order during upgrades,
+    /// each role gated on the previous one's rollout completing (see
+    /// [`crate::controller::apply::Applier::apply`]). The other tests here sort the names, which
+    /// would hide a reordering, so this one asserts on the order as built.
+    #[test]
+    fn stateful_sets_are_ordered_by_role() {
+        let cluster = validated_cluster();
+        let resources = build(&cluster, &cluster_info()).expect("build succeeds");
+
+        let names: Vec<String> = resources
+            .stateful_sets
+            .iter()
+            .filter_map(|stateful_set| stateful_set.meta().name.clone())
+            .collect();
+
+        assert_eq!(
+            names,
+            [
+                "hdfs-journalnode-default",
+                "hdfs-namenode-default",
+                "hdfs-datanode-default",
+            ]
+        );
+    }
+
     /// With every namenode Listener carrying an ingress address, the build step emits the
     /// discovery ConfigMap (named after the cluster) alongside the role-group ConfigMaps, so
     /// the apply step tracks it like any other resource. Without ready Listeners it is
@@ -548,4 +720,150 @@ mod tests {
             );
         }
     }
+
+    /// The named role group `ConfigMap`.
+    fn config_map<'a>(config_maps: &'a [ConfigMap], name: &str) -> &'a ConfigMap {
+        config_maps
+            .iter()
+            .find(|config_map| config_map.meta().name.as_deref() == Some(name))
+            .unwrap_or_else(|| panic!("the {name} ConfigMap is built"))
+    }
+
+    /// The sorted data keys of the named role group `ConfigMap`.
+    fn config_map_keys(config_maps: &[ConfigMap], name: &str) -> Vec<String> {
+        let mut keys: Vec<String> = config_map(config_maps, name)
+            .data
+            .as_ref()
+            .unwrap_or_else(|| panic!("the {name} ConfigMap has data"))
+            .keys()
+            .cloned()
+            .collect();
+        keys.sort();
+        keys
+    }
+
+    /// Each role gets the `log4j.properties` of exactly its own containers: the namenode's
+    /// `zkfc`, `format-namenodes` and `format-zookeeper`, the datanode's `wait-for-namenodes`,
+    /// and nothing role-specific for the journalnode. A missing file here is invisible at
+    /// runtime — the container just falls back to Hadoop's built-in logging and Vector collects
+    /// nothing for it — so it is pinned here.
+    #[test]
+    fn each_role_gets_only_its_own_log4j_configs() {
+        let cluster = validated_cluster();
+        let resources = build(&cluster, &cluster_info()).expect("build succeeds");
+        let config_maps = &resources.config_maps;
+
+        assert_eq!(
+            config_map_keys(config_maps, "hdfs-journalnode-default"),
+            [
+                "core-site.xml",
+                "hadoop-policy.xml",
+                "hdfs-site.xml",
+                "hdfs.log4j.properties",
+                "security.properties",
+                "ssl-client.xml",
+                "ssl-server.xml",
+            ]
+        );
+        assert_eq!(
+            config_map_keys(config_maps, "hdfs-namenode-default"),
+            [
+                "core-site.xml",
+                "format-namenodes.log4j.properties",
+                "format-zookeeper.log4j.properties",
+                "hadoop-policy.xml",
+                "hdfs-site.xml",
+                "hdfs.log4j.properties",
+                "security.properties",
+                "ssl-client.xml",
+                "ssl-server.xml",
+                "zkfc.log4j.properties",
+            ]
+        );
+        assert_eq!(
+            config_map_keys(config_maps, "hdfs-datanode-default"),
+            [
+                "core-site.xml",
+                "hadoop-policy.xml",
+                "hdfs-site.xml",
+                "hdfs.log4j.properties",
+                "security.properties",
+                "ssl-client.xml",
+                "ssl-server.xml",
+                "wait-for-namenodes.log4j.properties",
+            ]
+        );
+    }
+
+    /// `dfs.datanode.data.dir` must appear in the datanode's `hdfs-site.xml` and nowhere else.
+    /// Losing it is silent: the datanodes fall back to Hadoop's default directory, which is
+    /// container-local, so their blocks are gone on the next restart.
+    #[test]
+    fn only_the_datanode_config_map_sets_the_datanode_data_dir() {
+        let cluster = validated_cluster();
+        let resources = build(&cluster, &cluster_info()).expect("build succeeds");
+
+        for (name, expected) in [
+            ("hdfs-datanode-default", true),
+            ("hdfs-namenode-default", false),
+            ("hdfs-journalnode-default", false),
+        ] {
+            let hdfs_site = config_map(&resources.config_maps, name)
+                .data
+                .as_ref()
+                .and_then(|data| data.get("hdfs-site.xml"))
+                .unwrap_or_else(|| panic!("the {name} ConfigMap has an hdfs-site.xml"));
+
+            assert_eq!(
+                hdfs_site.contains("dfs.datanode.data.dir"),
+                expected,
+                "{name}'s hdfs-site.xml should {}contain dfs.datanode.data.dir",
+                if expected { "" } else { "not " }
+            );
+        }
+    }
+
+    /// Datanodes get their listener from an ephemeral pod volume; namenodes get theirs from a
+    /// volume claim template, for stable per-pod identity; journalnodes have no listener at all.
+    /// Both at once for one role would mean a pod volume and a claim template of the same name,
+    /// which the API server rejects at apply time.
+    #[test]
+    fn the_listener_is_a_pod_volume_for_datanodes_and_a_claim_template_for_namenodes() {
+        let cluster = validated_cluster();
+        let resources = build(&cluster, &cluster_info()).expect("build succeeds");
+
+        for (name, expect_pod_volume, expect_claim_template) in [
+            ("hdfs-datanode-default", true, false),
+            ("hdfs-namenode-default", false, true),
+            ("hdfs-journalnode-default", false, false),
+        ] {
+            let spec = resources
+                .stateful_sets
+                .iter()
+                .find(|stateful_set| stateful_set.meta().name.as_deref() == Some(name))
+                .and_then(|stateful_set| stateful_set.spec.as_ref())
+                .unwrap_or_else(|| panic!("the {name} StatefulSet is built"));
+
+            let has_pod_volume = spec
+                .template
+                .spec
+                .as_ref()
+                .and_then(|pod_spec| pod_spec.volumes.as_ref())
+                .is_some_and(|volumes| volumes.iter().any(|volume| volume.name == "listener"));
+            let has_claim_template = spec.volume_claim_templates.as_ref().is_some_and(|claims| {
+                claims
+                    .iter()
+                    .any(|claim| claim.meta().name.as_deref() == Some("listener"))
+            });
+
+            assert_eq!(
+                has_pod_volume, expect_pod_volume,
+                "{name} listener pod volume"
+            );
+            assert_eq!(
+                has_claim_template, expect_claim_template,
+                "{name} listener volume claim template"
+            );
+        }
+    }
 }
diff --git a/rust/operator-binary/src/controller/build/properties/hdfs_site.rs b/rust/operator-binary/src/controller/build/properties/hdfs_site.rs
index 3bf3b42c..c121a115 100644
--- a/rust/operator-binary/src/controller/build/properties/hdfs_site.rs
+++ b/rust/operator-binary/src/controller/build/properties/hdfs_site.rs
@@ -13,7 +13,7 @@ use stackable_operator::{
 use crate::{
     controller::{ValidatedCluster, build},
     crd::{
-        AnyNodeConfig, HdfsNodeRole, HdfsPodRef,
+        HdfsNodeRole, HdfsPodRef,
         constants::{
             DEFAULT_JOURNAL_NODE_RPC_PORT, DEFAULT_NAME_NODE_HTTP_PORT,
             DEFAULT_NAME_NODE_HTTPS_PORT, DEFAULT_NAME_NODE_RPC_PORT, DFS_DATANODE_DATA_DIR,
@@ -32,7 +32,7 @@ use crate::{
 pub fn build(
     cluster: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
-    merged_config: &AnyNodeConfig,
+    datanode_storage: Option<DataNodeStorageConfigInnerType>,
     overrides: KeyValueConfigOverrides,
 ) -> String {
     let cluster_config = &cluster.cluster_config;
@@ -52,11 +52,7 @@ pub fn build(
     let mut hdfs_site = HdfsSiteConfigBuilder::new(cluster.name.as_ref().to_owned());
     hdfs_site
         .dfs_namenode_name_dir()
-        .dfs_datanode_data_dir(
-            merged_config
-                .as_datanode()
-                .map(|node| node.resources.storage.clone()),
-        )
+        .dfs_datanode_data_dir(datanode_storage)
         .dfs_journalnode_edits_dir()
         .dfs_replication(cluster_config.dfs_replication)
         .dfs_name_services()
@@ -337,26 +333,16 @@ mod tests {
     use super::*;
     use crate::{
         controller::build::properties::test_support::{cluster_info, validated_cluster},
-        crd::HdfsNodeRole,
-        test_support::{anynode_config, role_group_name},
+        test_support::{datanode_config, role_group_name},
     };
 
-    fn namenode_merged_config(validated_cluster: &ValidatedCluster) -> &AnyNodeConfig {
-        anynode_config(
-            validated_cluster,
-            &HdfsNodeRole::Name,
-            &role_group_name("default"),
-        )
-    }
-
     #[test]
     fn renders_operator_defaults() {
         let validated_cluster = validated_cluster();
-        let merged = namenode_merged_config(&validated_cluster);
         let xml = build(
             &validated_cluster,
             &cluster_info(),
-            merged,
+            None,
             KeyValueConfigOverrides::default(),
         );
         assert!(
@@ -376,11 +362,10 @@ mod tests {
     #[test]
     fn user_overrides_win_over_defaults() {
         let validated_cluster = validated_cluster();
-        let merged = namenode_merged_config(&validated_cluster);
         let xml = build(
             &validated_cluster,
             &cluster_info(),
-            merged,
+            None,
             [("dfs.replication", "5")].into(),
         );
         assert!(
@@ -390,4 +375,30 @@ mod tests {
             "{xml}"
         );
     }
+
+    /// With a datanode's storage config, `dfs.datanode.data.dir` names one directory per PVC,
+    /// tagged with its HDFS storage type. Losing this property is silent: the datanodes fall back
+    /// to Hadoop's default directory, which is container-local, so their blocks are gone on the
+    /// next restart.
+    #[test]
+    fn datanode_storage_renders_the_data_dir() {
+        let validated_cluster = validated_cluster();
+        let storage = datanode_config(&validated_cluster, &role_group_name("default"))
+            .resources
+            .storage
+            .clone();
+
+        let xml = build(
+            &validated_cluster,
+            &cluster_info(),
+            Some(storage),
+            KeyValueConfigOverrides::default(),
+        );
+
+        assert!(
+            xml.contains("<name>dfs.datanode.data.dir</name>")
+                && xml.contains("<value>[DISK]/stackable/data/data/datanode</value>"),
+            "rendered hdfs-site.xml:\n{xml}"
+        );
+    }
 }
diff --git a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs
index 6b956cd1..f59e9b4f 100644
--- a/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs
+++ b/rust/operator-binary/src/controller/build/properties/product_logging/mod.rs
@@ -1,8 +1,6 @@
 //! Builders for the logging-related files in the rolegroup `ConfigMap`: the per-container
 //! `*.log4j.properties` configs and the (static) Vector agent config (`vector.yaml`).
 
-use std::borrow::Cow;
-
 use stackable_operator::{
     memory::{BinaryMultiple, MemoryQuantity},
     product_logging::{
@@ -12,12 +10,12 @@ use stackable_operator::{
     v2::product_logging::framework::STACKABLE_LOG_DIR,
 };
 
-use crate::{
-    controller::build::container::{
+use crate::controller::build::{
+    RoleGroupLogging, RoleSpecificValues,
+    container::{
         FORMAT_NAMENODES_CONTAINER_NAME, FORMAT_ZOOKEEPER_CONTAINER_NAME,
         WAIT_FOR_NAMENODES_CONTAINER_NAME, ZKFC_CONTAINER_NAME,
     },
-    crd::{AnyNodeConfig, DataNodeContainer, NameNodeContainer},
 };
 
 // We have a maximum of 4 continuous logging files for Namenodes. Datanodes and Journalnodes
@@ -79,75 +77,82 @@ pub fn vector_config_file_content() -> String {
 ///
 /// Returns `(filename, rendered content)` pairs; containers using a custom log ConfigMap are
 /// skipped, so the result is empty when none use automatic logging.
-pub fn build_log4j_configs(merged_config: &AnyNodeConfig) -> Vec<(&'static str, String)> {
+pub fn build_log4j_configs(
+    logging: &RoleGroupLogging,
+    role: &RoleSpecificValues,
+) -> Vec<(&'static str, String)> {
     let mut configs = Vec::new();
 
     add_log4j_config_if_automatic(
         &mut configs,
-        Some(merged_config.hdfs_logging()),
+        &logging.hdfs,
         HDFS_LOG4J_CONFIG_FILE,
         "hdfs",
         HDFS_LOG_FILE,
         MAX_HDFS_LOG_FILE_SIZE,
     );
-    add_log4j_config_if_automatic(
-        &mut configs,
-        merged_config
-            .as_namenode()
-            .map(|nn| nn.logging.for_container(&NameNodeContainer::Zkfc)),
-        ZKFC_LOG4J_CONFIG_FILE,
-        ZKFC_CONTAINER_NAME.as_ref(),
-        ZKFC_LOG_FILE,
-        MAX_ZKFC_LOG_FILE_SIZE,
-    );
-    add_log4j_config_if_automatic(
-        &mut configs,
-        merged_config.as_namenode().map(|nn| {
-            nn.logging
-                .for_container(&NameNodeContainer::FormatNameNodes)
-        }),
-        FORMAT_NAMENODES_LOG4J_CONFIG_FILE,
-        FORMAT_NAMENODES_CONTAINER_NAME.as_ref(),
-        FORMAT_NAMENODES_LOG_FILE,
-        MAX_FORMAT_NAMENODE_LOG_FILE_SIZE,
-    );
-    add_log4j_config_if_automatic(
-        &mut configs,
-        merged_config.as_namenode().map(|nn| {
-            nn.logging
-                .for_container(&NameNodeContainer::FormatZooKeeper)
-        }),
-        FORMAT_ZOOKEEPER_LOG4J_CONFIG_FILE,
-        FORMAT_ZOOKEEPER_CONTAINER_NAME.as_ref(),
-        FORMAT_ZOOKEEPER_LOG_FILE,
-        MAX_FORMAT_ZOOKEEPER_LOG_FILE_SIZE,
-    );
-    add_log4j_config_if_automatic(
-        &mut configs,
-        merged_config.as_datanode().map(|dn| {
-            dn.logging
-                .for_container(&DataNodeContainer::WaitForNameNodes)
-        }),
-        WAIT_FOR_NAMENODES_LOG4J_CONFIG_FILE,
-        WAIT_FOR_NAMENODES_CONTAINER_NAME.as_ref(),
-        WAIT_FOR_NAMENODES_LOG_FILE,
-        MAX_WAIT_NAMENODES_LOG_FILE_SIZE,
-    );
+
+    // Exhaustive, so a role's containers and their log4j configs cannot drift apart.
+    match role {
+        RoleSpecificValues::Journal => {}
+        RoleSpecificValues::Name {
+            zkfc,
+            format_namenodes,
+            format_zookeeper,
+        } => {
+            add_log4j_config_if_automatic(
+                &mut configs,
+                zkfc,
+                ZKFC_LOG4J_CONFIG_FILE,
+                ZKFC_CONTAINER_NAME.as_ref(),
+                ZKFC_LOG_FILE,
+                MAX_ZKFC_LOG_FILE_SIZE,
+            );
+            add_log4j_config_if_automatic(
+                &mut configs,
+                format_namenodes,
+                FORMAT_NAMENODES_LOG4J_CONFIG_FILE,
+                FORMAT_NAMENODES_CONTAINER_NAME.as_ref(),
+                FORMAT_NAMENODES_LOG_FILE,
+                MAX_FORMAT_NAMENODE_LOG_FILE_SIZE,
+            );
+            add_log4j_config_if_automatic(
+                &mut configs,
+                format_zookeeper,
+                FORMAT_ZOOKEEPER_LOG4J_CONFIG_FILE,
+                FORMAT_ZOOKEEPER_CONTAINER_NAME.as_ref(),
+                FORMAT_ZOOKEEPER_LOG_FILE,
+                MAX_FORMAT_ZOOKEEPER_LOG_FILE_SIZE,
+            );
+        }
+        RoleSpecificValues::Data {
+            wait_for_namenodes, ..
+        } => {
+            add_log4j_config_if_automatic(
+                &mut configs,
+                wait_for_namenodes,
+                WAIT_FOR_NAMENODES_LOG4J_CONFIG_FILE,
+                WAIT_FOR_NAMENODES_CONTAINER_NAME.as_ref(),
+                WAIT_FOR_NAMENODES_LOG_FILE,
+                MAX_WAIT_NAMENODES_LOG_FILE_SIZE,
+            );
+        }
+    }
 
     configs
 }
 
 fn add_log4j_config_if_automatic(
     configs: &mut Vec<(&'static str, String)>,
-    log_config: Option<Cow<ContainerLogConfig>>,
+    log_config: &ContainerLogConfig,
     log_config_file: &'static str,
     log_dir_name: &str,
     log_file: &str,
     max_log_file_size: MemoryQuantity,
 ) {
-    if let Some(ContainerLogConfig {
+    if let ContainerLogConfig {
         choice: Some(ContainerLogConfigChoice::Automatic(log_config)),
-    }) = log_config.as_deref()
+    } = log_config
     {
         configs.push((
             log_config_file,
diff --git a/rust/operator-binary/src/controller/build/resolve.rs b/rust/operator-binary/src/controller/build/resolve.rs
new file mode 100644
index 00000000..66fcea75
--- /dev/null
+++ b/rust/operator-binary/src/controller/build/resolve.rs
@@ -0,0 +1,268 @@
+//! Resolving one role group into the values the shared builders cannot derive themselves.
+//!
+//! One [`RoleGroupResolver`] impl per role config type, so a role's resolution is written once.
+
+use std::{fmt::Display, marker::PhantomData};
+
+use snafu::ResultExt;
+use stackable_operator::{
+    k8s_openapi::api::core::v1::{PersistentVolumeClaim, ResourceRequirements, Volume},
+    kvp::Labels,
+    product_logging::spec::{ContainerLogConfig, Logging},
+    v2::types::operator::RoleGroupName,
+};
+
+use super::{Error, ListenerVolumeSnafu, VolumeClaimTemplatesSnafu, container::ContainerConfig};
+use crate::crd::{
+    CommonNodeConfig, DataNodeConfig, DataNodeContainer, HdfsNodeRole, JournalNodeConfig,
+    JournalNodeContainer, NameNodeConfig, NameNodeContainer,
+    storage::DataNodeStorageConfigInnerType,
+};
+
+/// The log config of the two containers every role has. Containers only one role runs carry theirs
+/// in [`RoleSpecificValues`], which is the single place the role is decided.
+#[derive(Debug)]
+pub struct RoleGroupLogging {
+    /// The main `hdfs` container, which every role has.
+    pub hdfs: ContainerLogConfig,
+    /// The Vector sidecar; `None` when the Vector agent is disabled for this role group.
+    pub vector: Option<ContainerLogConfig>,
+}
+
+/// The values the shared builders cannot derive themselves, resolved by
+/// [`RoleGroupResolver::resolve`], which knows the role.
+///
+/// Every builder takes `RoleGroupConfig<C, ..>` and `ResolvedRoleGroup<C>` together, so one role's
+/// overrides and replica count cannot be paired with another role's resolved values: both are the
+/// same `C` or they do not compile.
+pub struct ResolvedRoleGroup<C> {
+    /// The selector labels of the role group's pods, also used as the `StatefulSet` selector and
+    /// on its listener volume.
+    ///
+    /// We must use the selector labels and not the recommended labels for the listener volumes.
+    /// This is because the recommended set contains a "managed-by" label. That label triggers the
+    /// cluster resources to "manage" listeners, which is wrong and leads to errors. The listeners
+    /// are managed by the listener-operator.
+    pub selector_labels: Labels,
+    /// The role group's merged config that is common to every role.
+    pub common: CommonNodeConfig,
+    /// The resource requirements of the role group's main and init containers; the ZKFC sidecar
+    /// has fixed requirements of its own and ignores this.
+    pub resources: ResourceRequirements,
+    /// The `StatefulSet`'s persistent volume claim templates.
+    pub volume_claim_templates: Vec<PersistentVolumeClaim>,
+    /// The values that exist for this role only.
+    pub role: RoleSpecificValues,
+    /// The log config of each of the role group's containers.
+    pub logging: RoleGroupLogging,
+    /// Ties the bundle to its config type. Needed because `C` appears in no other field, which on
+    /// its own does not compile (`E0392`). Private, so [`RoleGroupResolver::resolve`] is the only
+    /// constructor outside this module — a struct literal elsewhere is `E0451`.
+    _config: PhantomData<C>,
+}
+
+/// Everything that exists for one role only: the containers that role runs, their log configs, and
+/// its storage and listener arrangements.
+///
+/// An enum rather than `Option` fields, so consumers are exhaustive and three silent failures do
+/// not compile: a datanode without its storage drops `dfs.datanode.data.dir` and sends its blocks
+/// to container-local storage; a namenode with a pod-level listener volume collides with the
+/// identically named claim template and is rejected at apply time; a container without its log
+/// config falls back to Hadoop's built-in logging, uncollected by Vector.
+pub enum RoleSpecificValues {
+    /// Journalnodes run no role-specific container, have no listener and no role-specific
+    /// storage configuration.
+    Journal,
+    /// Namenodes run the `zkfc` side container and the `format-namenodes` and `format-zookeeper`
+    /// init containers. They get their listener from a volume claim template in
+    /// [`ResolvedRoleGroup::volume_claim_templates`], for stable per-pod identity, so they have
+    /// no pod-level listener volume.
+    Name {
+        zkfc: ContainerLogConfig,
+        format_namenodes: ContainerLogConfig,
+        format_zookeeper: ContainerLogConfig,
+    },
+    /// Datanodes run the `wait-for-namenodes` init container. They need no stable per-pod
+    /// identity, so their listener is an ephemeral pod volume, and they are the only role that
+    /// configures `dfs.datanode.data.dir`.
+    Data {
+        listener_volume: Volume,
+        storage: DataNodeStorageConfigInnerType,
+        wait_for_namenodes: ContainerLogConfig,
+    },
+}
+
+impl RoleSpecificValues {
+    /// The role group's ephemeral listener volume; only datanodes have one.
+    pub fn listener_volume(&self) -> Option<&Volume> {
+        match self {
+            Self::Data {
+                listener_volume, ..
+            } => Some(listener_volume),
+            Self::Journal | Self::Name { .. } => None,
+        }
+    }
+
+    /// The datanode data volume configuration, which drives `dfs.datanode.data.dir`; `None` for
+    /// the other roles.
+    pub fn datanode_storage(&self) -> Option<&DataNodeStorageConfigInnerType> {
+        match self {
+            Self::Data { storage, .. } => Some(storage),
+            Self::Journal | Self::Name { .. } => None,
+        }
+    }
+}
+
+/// The log config of the two containers every role has: the main `hdfs` container, and the Vector
+/// sidecar, which is `None` when the Vector agent is disabled for the role group.
+///
+/// Each role names these containers with its own enum, so this is generic over that enum rather
+/// than repeated once per role.
+fn common_container_logging<T>(
+    logging: &Logging<T>,
+    hdfs: T,
+    vector: T,
+) -> (ContainerLogConfig, Option<ContainerLogConfig>)
+where
+    T: Clone + Display + Ord,
+{
+    (
+        logging.for_container(&hdfs).into_owned(),
+        logging
+            .enable_vector_agent
+            .then(|| logging.for_container(&vector).into_owned()),
+    )
+}
+
+/// How to resolve one role group's role-specific values, implemented once per role config type.
+///
+/// The trait supplies the role and the single role-dependent step, which is what lets
+/// [`build_role`](super::build_role) be written once. The shared builders read [`Self::ROLE`]
+/// instead of taking a role parameter a caller could pair with the wrong config.
+pub(crate) trait RoleGroupResolver: Sized {
+    /// The role whose config this is.
+    const ROLE: HdfsNodeRole;
+
+    /// Resolves everything the shared builders cannot derive themselves. Takes the selector
+    /// labels because two of the three roles need them to build their listener.
+    fn resolve(
+        &self,
+        role_group_name: &RoleGroupName,
+        selector_labels: Labels,
+    ) -> Result<ResolvedRoleGroup<Self>, Error>;
+}
+
+impl RoleGroupResolver for JournalNodeConfig {
+    const ROLE: HdfsNodeRole = HdfsNodeRole::Journal;
+
+    fn resolve(
+        &self,
+        _role_group_name: &RoleGroupName,
+        selector_labels: Labels,
+    ) -> Result<ResolvedRoleGroup<Self>, Error> {
+        let (hdfs, vector) = common_container_logging(
+            &self.logging,
+            JournalNodeContainer::Hdfs,
+            JournalNodeContainer::Vector,
+        );
+
+        Ok(ResolvedRoleGroup {
+            selector_labels,
+            common: self.common.clone(),
+            resources: self.resources.clone().into(),
+            volume_claim_templates: ContainerConfig::journalnode_volume_claim_templates(self),
+            role: RoleSpecificValues::Journal,
+            logging: RoleGroupLogging { hdfs, vector },
+            _config: PhantomData,
+        })
+    }
+}
+
+impl RoleGroupResolver for NameNodeConfig {
+    const ROLE: HdfsNodeRole = HdfsNodeRole::Name;
+
+    fn resolve(
+        &self,
+        role_group_name: &RoleGroupName,
+        selector_labels: Labels,
+    ) -> Result<ResolvedRoleGroup<Self>, Error> {
+        // Namenodes get their listener from a persistent volume claim template, for stable
+        // per-pod identity, rather than from an ephemeral volume.
+        let volume_claim_templates =
+            ContainerConfig::namenode_volume_claim_templates(self, &selector_labels).context(
+                VolumeClaimTemplatesSnafu {
+                    role: Self::ROLE,
+                    role_group: role_group_name.clone(),
+                },
+            )?;
+
+        let (hdfs, vector) = common_container_logging(
+            &self.logging,
+            NameNodeContainer::Hdfs,
+            NameNodeContainer::Vector,
+        );
+
+        Ok(ResolvedRoleGroup {
+            selector_labels,
+            common: self.common.clone(),
+            resources: self.resources.clone().into(),
+            volume_claim_templates,
+            role: RoleSpecificValues::Name {
+                zkfc: self
+                    .logging
+                    .for_container(&NameNodeContainer::Zkfc)
+                    .into_owned(),
+                format_namenodes: self
+                    .logging
+                    .for_container(&NameNodeContainer::FormatNameNodes)
+                    .into_owned(),
+                format_zookeeper: self
+                    .logging
+                    .for_container(&NameNodeContainer::FormatZooKeeper)
+                    .into_owned(),
+            },
+            logging: RoleGroupLogging { hdfs, vector },
+            _config: PhantomData,
+        })
+    }
+}
+
+impl RoleGroupResolver for DataNodeConfig {
+    const ROLE: HdfsNodeRole = HdfsNodeRole::Data;
+
+    fn resolve(
+        &self,
+        role_group_name: &RoleGroupName,
+        selector_labels: Labels,
+    ) -> Result<ResolvedRoleGroup<Self>, Error> {
+        // Datanodes use an ephemeral listener volume, since they need no stable per-pod identity.
+        let listener_volume = ContainerConfig::datanode_listener_volume(self, &selector_labels)
+            .context(ListenerVolumeSnafu {
+                role: Self::ROLE,
+                role_group: role_group_name.clone(),
+            })?;
+
+        let (hdfs, vector) = common_container_logging(
+            &self.logging,
+            DataNodeContainer::Hdfs,
+            DataNodeContainer::Vector,
+        );
+
+        Ok(ResolvedRoleGroup {
+            selector_labels,
+            common: self.common.clone(),
+            resources: self.resources.clone().into(),
+            volume_claim_templates: ContainerConfig::datanode_volume_claim_templates(self),
+            role: RoleSpecificValues::Data {
+                listener_volume,
+                storage: self.resources.storage.clone(),
+                wait_for_namenodes: self
+                    .logging
+                    .for_container(&DataNodeContainer::WaitForNameNodes)
+                    .into_owned(),
+            },
+            logging: RoleGroupLogging { hdfs, vector },
+            _config: PhantomData,
+        })
+    }
+}
diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs
index 154373cb..f77ebf9c 100644
--- a/rust/operator-binary/src/controller/build/resource/config_map.rs
+++ b/rust/operator-binary/src/controller/build/resource/config_map.rs
@@ -1,33 +1,34 @@
 //! Build the per-rolegroup `ConfigMap` for the HdfsCluster.
 
-use snafu::{OptionExt, ResultExt, Snafu};
+use snafu::{ResultExt, Snafu};
 use stackable_operator::{
     builder::configmap::ConfigMapBuilder,
     k8s_openapi::api::core::v1::ConfigMap,
     product_logging::framework::VECTOR_CONFIG_FILE,
     utils::cluster_info::KubernetesClusterInfo,
-    v2::{config_file_writer::PropertiesWriterError, types::operator::RoleGroupName},
+    v2::{
+        config_file_writer::PropertiesWriterError,
+        role_utils::{JavaCommonConfig, RoleGroupConfig},
+        types::operator::RoleGroupName,
+    },
 };
 
 use crate::{
     controller::{
         ValidatedCluster,
         build::{
-            self,
+            self, ResolvedRoleGroup, RoleGroupResolver,
             properties::{
                 ConfigFileName, core_site, hadoop_policy, hdfs_site, product_logging,
                 security_properties, ssl_client, ssl_server,
             },
         },
     },
-    crd::HdfsNodeRole,
+    crd::v1alpha1,
 };
 
 #[derive(Snafu, Debug)]
 pub enum Error {
-    #[snafu(display("the validated cluster has no role group {role_group:?} for role {role:?}"))]
-    MissingRoleGroup { role: String, role_group: String },
-
     #[snafu(display("failed to serialize {} for {rolegroup}", ConfigFileName::Security))]
     JvmSecurityProperties {
         source: PropertiesWriterError,
@@ -44,40 +45,42 @@ pub enum Error {
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
-pub fn build_rolegroup_config_map(
+/// Builds the [`ConfigMap`] of one role group.
+///
+/// Every role-specific value is resolved by the caller into `resolved`. The role comes from
+/// `C::ROLE`, and `C`'s [`RoleGroupResolver`] bound ties it to `resolved`, so this cannot read one
+/// role's `HdfsNodeRole` alongside another role's resolved values. The datanode storage
+/// configuration comes from `resolved` rather than a separate parameter: taking it independently
+/// would let a caller pass a datanode without its storage, which silently drops
+/// `dfs.datanode.data.dir`.
+pub fn build_rolegroup_config_map<C: RoleGroupResolver>(
     cluster: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
-    role: &HdfsNodeRole,
     role_group_name: &RoleGroupName,
+    rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    resolved: &ResolvedRoleGroup<C>,
 ) -> Result<ConfigMap> {
+    let role = C::ROLE;
+
     tracing::info!(
         "Setting up ConfigMap for role {role} role group {role_group_name}",
         role = role.as_ref()
     );
 
-    let metadata = build::rolegroup_metadata(cluster, role, role_group_name);
+    let metadata = build::rolegroup_metadata(cluster, &role, role_group_name);
 
-    let rolegroup_config = cluster
-        .role_groups
-        .get(role)
-        .and_then(|role_groups| role_groups.get(role_group_name))
-        .with_context(|| MissingRoleGroupSnafu {
-            role: role.to_string(),
-            role_group: role_group_name.to_string(),
-        })?;
-    let merged_config = &rolegroup_config.config;
     let config_overrides = &rolegroup_config.config_overrides;
     let cluster_config = &cluster.cluster_config;
 
     let hdfs_site_xml = hdfs_site::build(
         cluster,
         cluster_info,
-        merged_config,
+        resolved.role.datanode_storage().cloned(),
         config_overrides.hdfs_site_xml.clone(),
     );
     let core_site_xml = core_site::build(
         cluster,
-        *role,
+        role,
         cluster_info,
         config_overrides.core_site_xml.clone(),
     );
@@ -108,10 +111,12 @@ pub fn build_rolegroup_config_map(
             )?,
         );
 
-    for (log_config_file, log4j_config) in product_logging::build_log4j_configs(merged_config) {
+    for (log_config_file, log4j_config) in
+        product_logging::build_log4j_configs(&resolved.logging, &resolved.role)
+    {
         builder.add_data(log_config_file, log4j_config);
     }
-    if merged_config.vector_logging_enabled() {
+    if resolved.logging.vector.is_some() {
         builder.add_data(
             VECTOR_CONFIG_FILE,
             product_logging::vector_config_file_content(),
diff --git a/rust/operator-binary/src/controller/build/resource/pdb.rs b/rust/operator-binary/src/controller/build/resource/pdb.rs
index 3823cd43..12136da2 100644
--- a/rust/operator-binary/src/controller/build/resource/pdb.rs
+++ b/rust/operator-binary/src/controller/build/resource/pdb.rs
@@ -13,7 +13,12 @@ use crate::{
 /// Builds the [`PodDisruptionBudget`] for the given `role`, or `None` if the role
 /// has no validated config or PDBs are disabled.
 pub fn build_pdb(cluster: &ValidatedCluster, role: &HdfsNodeRole) -> Option<PodDisruptionBudget> {
-    let pdb = &cluster.role_configs.get(role)?.pdb;
+    let role_config = match role {
+        HdfsNodeRole::Name => cluster.namenode_config.as_ref(),
+        HdfsNodeRole::Data => cluster.datanode_config.as_ref(),
+        HdfsNodeRole::Journal => cluster.journalnode_config.as_ref(),
+    }?;
+    let pdb = &role_config.pdb;
     if !pdb.enabled {
         return None;
     }
diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs
index dfc8e54d..a4988c69 100644
--- a/rust/operator-binary/src/controller/build/resource/statefulset.rs
+++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs
@@ -9,68 +9,66 @@ use stackable_operator::{
         apimachinery::pkg::apis::meta::v1::LabelSelector,
     },
     kube::api::ObjectMeta,
-    kvp::{LabelError, Labels},
     utils::cluster_info::KubernetesClusterInfo,
-    v2::types::operator::RoleGroupName,
+    v2::{
+        role_utils::{JavaCommonConfig, RoleGroupConfig},
+        types::operator::RoleGroupName,
+    },
 };
 
 use crate::{
     controller::{
-        ValidatedCluster, ValidatedRoleGroupConfig,
+        ValidatedCluster,
         build::{
-            self,
+            self, ResolvedRoleGroup, RoleGroupResolver,
             container::{self, ContainerConfig},
             graceful_shutdown::{self, add_graceful_shutdown_config},
         },
     },
-    crd::HdfsNodeRole,
+    crd::v1alpha1,
 };
 
 #[derive(Snafu, Debug)]
 pub enum Error {
-    #[snafu(display("failed to build roleGroup selector labels"))]
-    RoleGroupSelectorLabels { source: LabelError },
-
     #[snafu(display("failed to create container and volume configuration"))]
     FailedToCreateContainerAndVolumeConfiguration { source: container::Error },
 
     #[snafu(display("failed to configure graceful shutdown"))]
     GracefulShutdown { source: graceful_shutdown::Error },
-
-    #[snafu(display("failed to build role-group volume claim templates from config"))]
-    BuildRoleGroupVolumeClaimTemplates { source: container::Error },
 }
 
-pub(crate) fn build_rolegroup_statefulset(
+/// Builds the [`StatefulSet`] of one role group.
+///
+/// Every role-specific value is resolved by the caller into `resolved`. The role comes from
+/// `C::ROLE`, and `resolved` is [`ResolvedRoleGroup<C>`](ResolvedRoleGroup), produced by that same
+/// `C`'s [`RoleGroupResolver::resolve`], so it cannot disagree with `resolved`.
+pub(crate) fn build_rolegroup_statefulset<C: RoleGroupResolver>(
     validated: &ValidatedCluster,
     cluster_info: &KubernetesClusterInfo,
-    role: &HdfsNodeRole,
     role_group_name: &RoleGroupName,
-    rolegroup_config: &ValidatedRoleGroupConfig,
+    rolegroup_config: &RoleGroupConfig<C, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    resolved: &ResolvedRoleGroup<C>,
 ) -> Result<StatefulSet, Error> {
+    let role = &C::ROLE;
+
     tracing::info!(
         "Setting up StatefulSet for role {role} role group {role_group_name}",
         role = role.as_ref()
     );
 
     let image = &validated.image;
-    let merged_config = &rolegroup_config.config;
 
     // PodBuilder for StatefulSet Pod template.
     let mut pb = PodBuilder::new();
 
-    let rolegroup_selector_labels: Labels =
-        build::rolegroup_selector_labels(validated, role, role_group_name)
-            .context(RoleGroupSelectorLabelsSnafu)?;
-
     let pb_metadata = ObjectMeta {
-        labels: Some(rolegroup_selector_labels.clone().into()),
+        labels: Some(resolved.selector_labels.clone().into()),
         ..ObjectMeta::default()
     };
 
     pb.metadata(pb_metadata)
         .image_pull_secrets_from_product_image(image)
-        .affinity(&merged_config.affinity)
+        .affinity(&resolved.common.affinity)
         .service_account_name(
             validated
                 .cluster_resource_names()
@@ -83,39 +81,29 @@ pub(crate) fn build_rolegroup_statefulset(
                 .build(),
         );
 
-    // Adds all containers and volumes to the pod builder
-    // We must use the selector labels ("rolegroup_selector_labels") and not the recommended labels
-    // for the ephemeral listener volumes created by this function.
-    // This is because the recommended set contains a "managed-by" label. This label triggers
-    // the cluster resources to "manage" listeners which is wrong and leads to errors.
-    // The listeners are managed by the listener-operator.
+    // Adds all containers and volumes to the pod builder.
     ContainerConfig::add_containers_and_volumes(
         &mut pb,
         validated,
         cluster_info,
-        role,
         role_group_name,
         rolegroup_config,
-        &rolegroup_selector_labels,
+        resolved,
     )
     .context(FailedToCreateContainerAndVolumeConfigurationSnafu)?;
 
-    add_graceful_shutdown_config(merged_config, &mut pb).context(GracefulShutdownSnafu)?;
+    add_graceful_shutdown_config(&resolved.common, &mut pb).context(GracefulShutdownSnafu)?;
 
     // The `podOverrides` were already merged (role <- role group) during validation
     // by the local-`framework` `with_validated_config`.
     let mut pod_template = pb.build_template();
     pod_template.merge_from(rolegroup_config.pod_overrides.clone());
 
-    // The same comment regarding labels is valid here as it is for the ContainerConfig::add_containers_and_volumes() call above.
-    let pvcs = ContainerConfig::volume_claim_templates(merged_config, &rolegroup_selector_labels)
-        .context(BuildRoleGroupVolumeClaimTemplatesSnafu)?;
-
     let statefulset_spec = StatefulSetSpec {
         pod_management_policy: Some("OrderedReady".to_string()),
         replicas: rolegroup_config.replicas.map(i32::from),
         selector: LabelSelector {
-            match_labels: Some(rolegroup_selector_labels.into()),
+            match_labels: Some(resolved.selector_labels.clone().into()),
             ..LabelSelector::default()
         },
         service_name: Some(
@@ -125,7 +113,7 @@ pub(crate) fn build_rolegroup_statefulset(
         ),
         template: pod_template,
 
-        volume_claim_templates: Some(pvcs),
+        volume_claim_templates: Some(resolved.volume_claim_templates.clone()),
         ..StatefulSetSpec::default()
     };
 
diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs
index afbbfedc..432a7c57 100644
--- a/rust/operator-binary/src/controller/mod.rs
+++ b/rust/operator-binary/src/controller/mod.rs
@@ -14,7 +14,7 @@ use stackable_operator::{
     v2::{
         HasName, HasUid, NameIsValidLabelValue,
         role_group_utils::{QualifiedRoleGroupName, ResourceNames},
-        role_utils::{self, RoleGroupConfig},
+        role_utils::{self, JavaCommonConfig, RoleGroupConfig},
         types::{
             kubernetes::{ConfigMapName, NamespaceName, ServiceName, Uid},
             operator::{
@@ -29,8 +29,8 @@ use crate::{
     HDFS_OPERATOR_NAME,
     controller::build::opa::HdfsOpaConfig,
     crd::{
-        AnyNodeConfig, HdfsNodeRole, UpgradeState, constants::APP_NAME,
-        security::AuthenticationConfig, v1alpha1,
+        DataNodeConfig, HdfsNodeRole, JournalNodeConfig, NameNodeConfig, UpgradeState,
+        constants::APP_NAME, security::AuthenticationConfig, v1alpha1,
     },
     hdfs_controller::HDFS_CONTROLLER_NAME,
 };
@@ -53,9 +53,9 @@ pub struct Applied;
 
 /// Every Kubernetes resource produced by the build step.
 ///
-/// The resources are flat, unordered collections. The reconcile step re-groups the
-/// StatefulSets by role to preserve HDFS's ordered, rollout-gated deployment during
-/// upgrades. The discovery `ConfigMap` is part of `config_maps` whenever it can be built or
+/// The resources are flat collections. `stateful_sets` is ordered by [`HdfsNodeRole`], which
+/// [`apply::Applier::apply`] relies on when rolling out an upgrade.
+/// The discovery `ConfigMap` is part of `config_maps` whenever it can be built or
 /// re-emitted; it is only absent before its first successful build (see
 /// [`build::resource::discovery::build_discovery_config_map`]).
 ///
@@ -72,13 +72,17 @@ pub struct KubernetesResources<T> {
     pub status: PhantomData<T>,
 }
 
-/// The [`RoleGroupConfig`] specialised for HDFS: the validated config is the
-/// per-role [`AnyNodeConfig`],
-pub type ValidatedRoleGroupConfig = RoleGroupConfig<
-    AnyNodeConfig,
-    stackable_operator::v2::role_utils::JavaCommonConfig,
-    v1alpha1::HdfsConfigOverrides,
->;
+/// The [`RoleGroupConfig`] of one namenode role group.
+pub type NameNodeRoleGroupConfig =
+    RoleGroupConfig<NameNodeConfig, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>;
+
+/// The [`RoleGroupConfig`] of one datanode role group.
+pub type DataNodeRoleGroupConfig =
+    RoleGroupConfig<DataNodeConfig, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>;
+
+/// The [`RoleGroupConfig`] of one journalnode role group.
+pub type JournalNodeRoleGroupConfig =
+    RoleGroupConfig<JournalNodeConfig, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>;
 
 /// The validated cluster: proves that config merging and validation succeeded
 /// for every role and role group before any resources are created. Placed in the
@@ -99,8 +103,21 @@ pub struct ValidatedCluster {
     pub product_version: ProductVersion,
     pub image: ResolvedProductImage,
     pub cluster_config: ValidatedClusterConfig,
-    pub role_groups: BTreeMap<HdfsNodeRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
-    pub role_configs: BTreeMap<HdfsNodeRole, ValidatedRoleConfig>,
+    /// The namenode role-level config, or `None` if the role is absent.
+    pub namenode_config: Option<ValidatedRoleConfig>,
+    /// The validated config of every namenode role group, keyed by role group name; empty if the
+    /// role is absent.
+    pub namenode_role_group_configs: BTreeMap<RoleGroupName, NameNodeRoleGroupConfig>,
+    /// The datanode role-level config, or `None` if the role is absent.
+    pub datanode_config: Option<ValidatedRoleConfig>,
+    /// The validated config of every datanode role group, keyed by role group name; empty if the
+    /// role is absent.
+    pub datanode_role_group_configs: BTreeMap<RoleGroupName, DataNodeRoleGroupConfig>,
+    /// The journalnode role-level config, or `None` if the role is absent.
+    pub journalnode_config: Option<ValidatedRoleConfig>,
+    /// The validated config of every journalnode role group, keyed by role group name; empty if
+    /// the role is absent.
+    pub journalnode_role_group_configs: BTreeMap<RoleGroupName, JournalNodeRoleGroupConfig>,
     /// The namenode pod `Listener`s as currently stored in the cluster (see
     /// [`crate::controller::dereference::DereferencedObjects::namenode_listeners`]).
     pub namenode_listeners: Vec<listener::v1alpha1::Listener>,
@@ -113,46 +130,31 @@ pub struct ValidatedCluster {
 }
 
 impl ValidatedCluster {
-    #[allow(clippy::too_many_arguments)]
-    pub fn new(
-        name: ClusterName,
-        namespace: NamespaceName,
-        uid: Uid,
-        image: ResolvedProductImage,
-        cluster_config: ValidatedClusterConfig,
-        role_groups: BTreeMap<HdfsNodeRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
-        role_configs: BTreeMap<HdfsNodeRole, ValidatedRoleConfig>,
-        namenode_listeners: Vec<listener::v1alpha1::Listener>,
-        discovery_config_map: Option<ConfigMap>,
-        status: ValidatedClusterStatus,
-    ) -> Self {
-        // `app_version_label_value` is constructed to be a valid label value, so it is also a valid
-        // `ProductVersion`.
-        let product_version = ProductVersion::from_str(&image.app_version_label_value)
-            .expect("the app version label value is a valid product version");
-        Self {
-            metadata: ObjectMeta {
-                name: Some(name.to_string()),
-                namespace: Some(namespace.to_string()),
-                // The uid is required so this type can produce valid owner references
-                // (Kubernetes rejects owner references without a uid).
-                uid: Some(uid.to_string()),
-                ..ObjectMeta::default()
-            },
-            name,
-            namespace,
-            uid,
-            image,
-            product_version,
-            cluster_config,
-            role_groups,
-            role_configs,
-            namenode_listeners,
-            discovery_config_map,
-            status,
+    /// The `ObjectMeta` a `ValidatedCluster` carries so it can own the objects built from it.
+    ///
+    /// The uid is required: Kubernetes rejects owner references without one.
+    pub(crate) fn object_meta(
+        name: &ClusterName,
+        namespace: &NamespaceName,
+        uid: &Uid,
+    ) -> ObjectMeta {
+        ObjectMeta {
+            name: Some(name.to_string()),
+            namespace: Some(namespace.to_string()),
+            uid: Some(uid.to_string()),
+            ..ObjectMeta::default()
         }
     }
 
+    /// The product version of the resolved image.
+    ///
+    /// `app_version_label_value` is constructed to be a valid label value, so it is also a valid
+    /// `ProductVersion`.
+    pub(crate) fn product_version(image: &ResolvedProductImage) -> ProductVersion {
+        ProductVersion::from_str(&image.app_version_label_value)
+            .expect("the app version label value is a valid product version")
+    }
+
     /// Whether HTTPS is enabled, derived from the validated authentication settings.
     pub fn has_https_enabled(&self) -> bool {
         self.cluster_config.authentication.is_some()
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
index 122bbdd8..b61d90fd 100644
--- a/rust/operator-binary/src/controller/validate.rs
+++ b/rust/operator-binary/src/controller/validate.rs
@@ -9,20 +9,19 @@ use stackable_operator::{
     role_utils::GenericRoleConfig,
     v2::{
         controller_utils::{get_cluster_name, get_namespace, get_uid},
-        role_utils::{JavaCommonConfig, Role, with_validated_config},
+        role_utils::{JavaCommonConfig, Role, RoleGroupConfig, with_validated_config},
         types::operator::RoleGroupName,
     },
 };
-use strum::IntoEnumIterator;
 
 use crate::{
     controller::{
         ValidatedCluster, ValidatedClusterConfig, ValidatedClusterStatus, ValidatedRoleConfig,
-        ValidatedRoleGroupConfig, dereference::DereferencedObjects,
+        dereference::DereferencedObjects,
     },
     crd::{
-        AnyNodeConfig, DataNodeConfigFragment, HdfsNodeRole, JournalNodeConfigFragment,
-        NameNodeConfigFragment, UpgradeStateError, v1alpha1,
+        DataNodeConfigFragment, HdfsNodeRole, JournalNodeConfigFragment, NameNodeConfigFragment,
+        UpgradeStateError, v1alpha1,
     },
 };
 
@@ -88,38 +87,22 @@ pub fn validate_cluster(
         )
         .context(ResolveProductImageSnafu)?;
 
-    let mut role_groups = BTreeMap::new();
-    let mut role_configs = BTreeMap::new();
     let cluster_name = get_cluster_name(hdfs).context(GetClusterNameSnafu)?;
 
-    for hdfs_role in HdfsNodeRole::iter() {
-        if let Some(GenericRoleConfig {
-            pod_disruption_budget: pdb,
-        }) = hdfs.role_config(&hdfs_role)
-        {
-            role_configs.insert(hdfs_role, ValidatedRoleConfig { pdb: pdb.clone() });
-        }
-
-        let group_configs = match hdfs_role {
-            HdfsNodeRole::Name => validate_role_group_configs(
-                hdfs.spec.name_nodes.as_ref(),
-                NameNodeConfigFragment::default_config(cluster_name.as_ref(), &hdfs_role),
-                AnyNodeConfig::Name,
-            )?,
-            HdfsNodeRole::Data => validate_role_group_configs(
-                hdfs.spec.data_nodes.as_ref(),
-                DataNodeConfigFragment::default_config(cluster_name.as_ref(), &hdfs_role),
-                AnyNodeConfig::Data,
-            )?,
-            HdfsNodeRole::Journal => validate_role_group_configs(
-                hdfs.spec.journal_nodes.as_ref(),
-                JournalNodeConfigFragment::default_config(cluster_name.as_ref(), &hdfs_role),
-                AnyNodeConfig::Journal,
-            )?,
-        };
-
-        role_groups.insert(hdfs_role, group_configs);
-    }
+    // The first failure propagates, so this order decides which misconfiguration the user is told
+    // about when more than one role is wrong. It is `HdfsNodeRole`'s variant order.
+    let journalnode_role_group_configs = validate_role_group_configs(
+        hdfs.spec.journal_nodes.as_ref(),
+        JournalNodeConfigFragment::default_config(cluster_name.as_ref(), &HdfsNodeRole::Journal),
+    )?;
+    let namenode_role_group_configs = validate_role_group_configs(
+        hdfs.spec.name_nodes.as_ref(),
+        NameNodeConfigFragment::default_config(cluster_name.as_ref(), &HdfsNodeRole::Name),
+    )?;
+    let datanode_role_group_configs = validate_role_group_configs(
+        hdfs.spec.data_nodes.as_ref(),
+        DataNodeConfigFragment::default_config(cluster_name.as_ref(), &HdfsNodeRole::Data),
+    )?;
 
     let namespace = get_namespace(hdfs).context(GetClusterNamespaceSnafu)?;
     let uid = get_uid(hdfs).context(GetClusterUidSnafu)?;
@@ -135,18 +118,43 @@ pub fn validate_cluster(
             .and_then(|status| status.upgrade_target_product_version.clone()),
     };
 
-    Ok(ValidatedCluster::new(
-        cluster_name,
+    // The three role-level configs share one type, so naming each field is what stops a role
+    // being given another role's PodDisruptionBudget.
+    Ok(ValidatedCluster {
+        metadata: ValidatedCluster::object_meta(&cluster_name, &namespace, &uid),
+        product_version: ValidatedCluster::product_version(&image),
+        name: cluster_name,
         namespace,
         uid,
+        cluster_config: ValidatedClusterConfig::resolve(hdfs, hdfs_opa_config),
         image,
-        ValidatedClusterConfig::resolve(hdfs, hdfs_opa_config),
-        role_groups,
-        role_configs,
+        namenode_config: validated_role_config(hdfs, HdfsNodeRole::Name),
+        namenode_role_group_configs,
+        datanode_config: validated_role_config(hdfs, HdfsNodeRole::Data),
+        datanode_role_group_configs,
+        journalnode_config: validated_role_config(hdfs, HdfsNodeRole::Journal),
+        journalnode_role_group_configs,
         namenode_listeners,
         discovery_config_map,
         status,
-    ))
+    })
+}
+
+/// The validated role-level config of one role, or `None` if the role is absent from the spec.
+///
+/// [`GenericRoleConfig`] is destructured without `..`, so a field added to it upstream fails to
+/// compile here instead of being silently left unvalidated.
+fn validated_role_config(
+    hdfs: &v1alpha1::HdfsCluster,
+    role: HdfsNodeRole,
+) -> Option<ValidatedRoleConfig> {
+    hdfs.role_config(&role).map(
+        |GenericRoleConfig {
+             pod_disruption_budget,
+         }| ValidatedRoleConfig {
+            pdb: pod_disruption_budget.clone(),
+        },
+    )
 }
 
 /// Validates every role group of a role into a map keyed by role group name.
@@ -155,15 +163,19 @@ pub fn validate_cluster(
 /// [`with_validated_config`], which folds the CRD config fragment (default <-
 /// role <- role group) plus the `configOverrides`, `envOverrides`, `cliOverrides`
 /// and `podOverrides` (role group wins) into a single
-/// [`RoleGroupConfig`](stackable_operator::v2::role_utils::RoleGroupConfig). The
-/// concrete per-role validated config is wrapped into [`AnyNodeConfig`] via `wrap`.
+/// [`RoleGroupConfig`].
 ///
 /// Returns an empty map if the role is not configured.
 fn validate_role_group_configs<Config, ValidatedConfig>(
     role: Option<&Role<Config, v1alpha1::HdfsConfigOverrides, GenericRoleConfig, JavaCommonConfig>>,
     default_config: Config,
-    wrap: fn(ValidatedConfig) -> AnyNodeConfig,
-) -> Result<BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>, Error>
+) -> Result<
+    BTreeMap<
+        RoleGroupName,
+        RoleGroupConfig<ValidatedConfig, JavaCommonConfig, v1alpha1::HdfsConfigOverrides>,
+    >,
+    Error,
+>
 where
     Config: Clone + Merge,
     ValidatedConfig: FromFragment<Fragment = Config>,
@@ -184,11 +196,10 @@ where
             >(role_group, role, &default_config)
             .context(ValidateRoleGroupConfigSnafu)?;
 
-            // Re-wrap the per-role validated config into the role-agnostic
-            // `AnyNodeConfig`; the merged overrides carry over unchanged.
-            let validated = ValidatedRoleGroupConfig {
+            // The overrides carry over unchanged.
+            let validated = RoleGroupConfig {
                 replicas: validated.replicas,
-                config: wrap(validated.config.config),
+                config: validated.config.config,
                 config_overrides: validated.config.config_overrides,
                 env_overrides: validated.config.env_overrides.into(),
                 cli_overrides: validated.config.cli_overrides,
diff --git a/rust/operator-binary/src/crd/affinity.rs b/rust/operator-binary/src/crd/affinity.rs
index af13dd45..19f7098b 100644
--- a/rust/operator-binary/src/crd/affinity.rs
+++ b/rust/operator-binary/src/crd/affinity.rs
@@ -43,7 +43,7 @@ mod test {
 
     use crate::{
         crd::HdfsNodeRole,
-        test_support::{anynode_config, deserialize_and_validate_cluster, role_group_name},
+        test_support::{common_config, deserialize_and_validate_cluster, role_group_name},
     };
 
     #[rstest]
@@ -78,7 +78,7 @@ spec:
         "#;
 
         let validated_cluster = deserialize_and_validate_cluster(input);
-        let merged_config = anynode_config(&validated_cluster, &role, &role_group_name("default"));
+        let merged_config = common_config(&validated_cluster, &role, &role_group_name("default"));
 
         assert_eq!(
             merged_config.affinity,
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
index 8c254786..8701670a 100644
--- a/rust/operator-binary/src/crd/mod.rs
+++ b/rust/operator-binary/src/crd/mod.rs
@@ -25,10 +25,7 @@ use stackable_operator::{
     deep_merger::ObjectOverrides,
     k8s_openapi::apimachinery::pkg::api::resource::Quantity,
     kube::CustomResource,
-    product_logging::{
-        self,
-        spec::{ContainerLogConfig, Logging},
-    },
+    product_logging::{self, spec::Logging},
     role_utils::GenericRoleConfig,
     schemars::{self, JsonSchema},
     shared::time::Duration,
@@ -357,94 +354,17 @@ pub struct CommonNodeConfig {
     pub requested_secret_lifetime: Option<Duration>,
 }
 
-/// Configuration for a rolegroup of an unknown type.
-#[derive(Clone, Debug)]
-pub enum AnyNodeConfig {
-    Name(NameNodeConfig),
-    Data(DataNodeConfig),
-    Journal(JournalNodeConfig),
-}
-
-impl Deref for AnyNodeConfig {
-    type Target = CommonNodeConfig;
-
-    fn deref(&self) -> &Self::Target {
-        match self {
-            AnyNodeConfig::Name(node) => &node.common,
-            AnyNodeConfig::Data(node) => &node.common,
-            AnyNodeConfig::Journal(node) => &node.common,
-        }
-    }
-}
-
-impl AnyNodeConfig {
-    // Downcasting helpers for each variant
-    pub fn as_namenode(&self) -> Option<&NameNodeConfig> {
-        if let Self::Name(node) = self {
-            Some(node)
-        } else {
-            None
-        }
-    }
-
-    pub fn as_datanode(&self) -> Option<&DataNodeConfig> {
-        if let Self::Data(node) = self {
-            Some(node)
-        } else {
-            None
-        }
-    }
-
-    #[allow(unused)]
-    pub fn as_journalnode(&self) -> Option<&JournalNodeConfig> {
-        if let Self::Journal(node) = self {
-            Some(node)
-        } else {
-            None
-        }
-    }
-
-    // Logging config is distinct between each role, due to the different enum types,
-    // so provide helpers for containers that are common between all roles.
-    pub fn hdfs_logging(&'_ self) -> Cow<'_, ContainerLogConfig> {
-        match self {
-            AnyNodeConfig::Name(node) => node.logging.for_container(&NameNodeContainer::Hdfs),
-            AnyNodeConfig::Data(node) => node.logging.for_container(&DataNodeContainer::Hdfs),
-            AnyNodeConfig::Journal(node) => node.logging.for_container(&JournalNodeContainer::Hdfs),
-        }
-    }
-
-    pub fn vector_logging(&'_ self) -> Cow<'_, ContainerLogConfig> {
-        match &self {
-            AnyNodeConfig::Name(node) => node.logging.for_container(&NameNodeContainer::Vector),
-            AnyNodeConfig::Data(node) => node.logging.for_container(&DataNodeContainer::Vector),
-            AnyNodeConfig::Journal(node) => {
-                node.logging.for_container(&JournalNodeContainer::Vector)
-            }
-        }
-    }
-
-    pub fn vector_logging_enabled(&self) -> bool {
-        match self {
-            AnyNodeConfig::Name(node) => node.logging.enable_vector_agent,
-            AnyNodeConfig::Data(node) => node.logging.enable_vector_agent,
-            AnyNodeConfig::Journal(node) => node.logging.enable_vector_agent,
-        }
-    }
-
-    pub fn requested_secret_lifetime(&self) -> Option<Duration> {
-        match self {
-            AnyNodeConfig::Name(node) => node.common.requested_secret_lifetime,
-            AnyNodeConfig::Data(node) => node.common.requested_secret_lifetime,
-            AnyNodeConfig::Journal(node) => node.common.requested_secret_lifetime,
-        }
-    }
-}
-
 constant!(JOURNALNODE_ROLE_NAME: RoleName = "journalnode");
 constant!(NAMENODE_ROLE_NAME: RoleName = "namenode");
 constant!(DATANODE_ROLE_NAME: RoleName = "datanode");
 
+/// The HDFS roles, declared in the order they must be rolled out during an upgrade:
+/// journalnodes, then namenodes, then datanodes.
+///
+/// The variant order is load-bearing, because the derived [`Ord`] is what orders the
+/// StatefulSets the apply step rolls out (see
+/// [`crate::controller::build::build`] and [`crate::controller::apply::Applier::apply`]).
+/// Reordering the variants reorders an HDFS upgrade.
 #[derive(
     Clone,
     Copy,
@@ -497,7 +417,11 @@ impl HdfsNodeRole {
         }
     }
 
-    pub fn check_valid_dfs_replication(&self) -> bool {
+    /// Whether this role's replica count is the one `dfs.replication` is compared against.
+    ///
+    /// Only datanodes hold block replicas, so only their count can fall short of the
+    /// replication factor.
+    pub fn replicas_must_cover_dfs_replication(&self) -> bool {
         match self {
             HdfsNodeRole::Name => false,
             HdfsNodeRole::Data => true,
diff --git a/rust/operator-binary/src/event.rs b/rust/operator-binary/src/event.rs
index 4e6520a8..7a891f27 100644
--- a/rust/operator-binary/src/event.rs
+++ b/rust/operator-binary/src/event.rs
@@ -5,7 +5,11 @@ use stackable_operator::{
 };
 use strum::{EnumDiscriminants, IntoStaticStr};
 
-use crate::{controller::ValidatedCluster, crd::HdfsNodeRole, hdfs_controller::Ctx};
+use crate::{
+    controller::{ValidatedCluster, build::total_replicas},
+    crd::HdfsNodeRole,
+    hdfs_controller::Ctx,
+};
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
 #[strum_discriminants(derive(IntoStaticStr))]
@@ -43,13 +47,11 @@ pub fn build_invalid_replica_message(
     validated_cluster: &ValidatedCluster,
     role: &HdfsNodeRole,
 ) -> Option<String> {
-    let replicas: u16 = validated_cluster
-        .role_groups
-        .get(role)
-        .into_iter()
-        .flatten()
-        .map(|(_, role_group)| role_group.replicas.unwrap_or_default())
-        .sum();
+    let replicas = match role {
+        HdfsNodeRole::Name => total_replicas(&validated_cluster.namenode_role_group_configs),
+        HdfsNodeRole::Data => total_replicas(&validated_cluster.datanode_role_group_configs),
+        HdfsNodeRole::Journal => total_replicas(&validated_cluster.journalnode_role_group_configs),
+    };
 
     let dfs_replication = validated_cluster.cluster_config.dfs_replication;
     let role_name = role.to_string();
@@ -63,7 +65,7 @@ pub fn build_invalid_replica_message(
         Some(format!(
             "{role_name}: currently has an even number of replicas [{replicas}], but should always have an odd number to ensure quorum"
         ))
-    } else if !role.replicas_can_be_even() && replicas < dfs_replication as u16 {
+    } else if role.replicas_must_cover_dfs_replication() && replicas < dfs_replication as u16 {
         Some(format!(
             "{role_name}: HDFS replication factor [{dfs_replication}] is configured greater than data node replicas [{replicas}]"
         ))
@@ -71,3 +73,174 @@ pub fn build_invalid_replica_message(
         None
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::test_support::deserialize_and_validate_cluster;
+
+    /// A role group with no explicit `replicas` runs one pod — Kubernetes' default for a
+    /// `StatefulSet` with `replicas: null` — so counting it as zero would warn about a role group
+    /// that is fine.
+    #[test]
+    fn an_unset_replica_count_counts_as_one_datanode() {
+        let cluster = deserialize_and_validate_cluster(
+            "
+---
+apiVersion: hdfs.stackable.tech/v1alpha1
+kind: HdfsCluster
+metadata:
+  name: hdfs
+  namespace: default
+  uid: c2c8c5c0-0b5a-4b1e-9f3e-1a2b3c4d5e6f
+spec:
+  image:
+    productVersion: 3.4.0
+  clusterConfig:
+    zookeeperConfigMapName: hdfs-zk
+    dfsReplication: 1
+  nameNodes:
+    roleGroups:
+      default:
+        replicas: 2
+  journalNodes:
+    roleGroups:
+      default:
+        replicas: 3
+  dataNodes:
+    roleGroups:
+      default: {}
+",
+        );
+
+        assert_eq!(
+            build_invalid_replica_message(&cluster, &HdfsNodeRole::Data),
+            None
+        );
+    }
+
+    /// A role with no role groups at all really does have zero replicas, and still warns.
+    #[test]
+    fn a_role_without_role_groups_still_warns() {
+        let cluster = deserialize_and_validate_cluster(
+            "
+---
+apiVersion: hdfs.stackable.tech/v1alpha1
+kind: HdfsCluster
+metadata:
+  name: hdfs
+  namespace: default
+  uid: c2c8c5c0-0b5a-4b1e-9f3e-1a2b3c4d5e6f
+spec:
+  image:
+    productVersion: 3.4.0
+  clusterConfig:
+    zookeeperConfigMapName: hdfs-zk
+    dfsReplication: 1
+  nameNodes:
+    roleGroups: {}
+  journalNodes:
+    roleGroups:
+      default:
+        replicas: 3
+  dataNodes:
+    roleGroups:
+      default:
+        replicas: 1
+",
+        );
+
+        assert_eq!(
+            build_invalid_replica_message(&cluster, &HdfsNodeRole::Name).as_deref(),
+            Some(
+                "namenode: only has 0 replicas configured, it is strongly recommended to use at \
+                 least [2]"
+            )
+        );
+    }
+
+    /// A `dfsReplication` above the datanode count means HDFS cannot place every replica, so the
+    /// user is warned. [`HdfsNodeRole::replicas_must_cover_dfs_replication`] gates it to datanodes,
+    /// which is what the message is about.
+    #[test]
+    fn fewer_datanodes_than_the_replication_factor_warns() {
+        let cluster = deserialize_and_validate_cluster(
+            "
+---
+apiVersion: hdfs.stackable.tech/v1alpha1
+kind: HdfsCluster
+metadata:
+  name: hdfs
+  namespace: default
+  uid: c2c8c5c0-0b5a-4b1e-9f3e-1a2b3c4d5e6f
+spec:
+  image:
+    productVersion: 3.4.0
+  clusterConfig:
+    zookeeperConfigMapName: hdfs-zk
+    dfsReplication: 3
+  nameNodes:
+    roleGroups:
+      default:
+        replicas: 2
+  journalNodes:
+    roleGroups:
+      default:
+        replicas: 3
+  dataNodes:
+    roleGroups:
+      default:
+        replicas: 2
+",
+        );
+
+        assert_eq!(
+            build_invalid_replica_message(&cluster, &HdfsNodeRole::Data).as_deref(),
+            Some(
+                "datanode: HDFS replication factor [3] is configured greater than data node \
+                 replicas [2]"
+            )
+        );
+    }
+
+    /// The `dfsReplication` warning is worded in terms of datanode replicas, so only datanodes
+    /// are compared against it. A journalnode role group with fewer replicas than
+    /// `dfsReplication` is not a misconfiguration and must stay silent.
+    #[test]
+    fn journalnodes_are_not_compared_to_the_replication_factor() {
+        let cluster = deserialize_and_validate_cluster(
+            "
+---
+apiVersion: hdfs.stackable.tech/v1alpha1
+kind: HdfsCluster
+metadata:
+  name: hdfs
+  namespace: default
+  uid: c2c8c5c0-0b5a-4b1e-9f3e-1a2b3c4d5e6f
+spec:
+  image:
+    productVersion: 3.4.0
+  clusterConfig:
+    zookeeperConfigMapName: hdfs-zk
+    dfsReplication: 5
+  nameNodes:
+    roleGroups:
+      default:
+        replicas: 2
+  journalNodes:
+    roleGroups:
+      default:
+        replicas: 3
+  dataNodes:
+    roleGroups:
+      default:
+        replicas: 5
+",
+        );
+
+        assert_eq!(
+            build_invalid_replica_message(&cluster, &HdfsNodeRole::Journal),
+            None
+        );
+    }
+}
diff --git a/rust/operator-binary/src/hdfs_controller.rs b/rust/operator-binary/src/hdfs_controller.rs
index 99a2eb07..7734aa21 100644
--- a/rust/operator-binary/src/hdfs_controller.rs
+++ b/rust/operator-binary/src/hdfs_controller.rs
@@ -110,10 +110,9 @@ pub async fn reconcile_hdfs(
 
     // Warn about invalid replica counts. This is validation feedback and independent of the
     // resource application below.
+    // Every role is checked, including one that is absent from the spec entirely: a cluster
+    // without journalnodes is exactly the case the warning is for.
     for role in HdfsNodeRole::iter() {
-        if !validated_cluster.role_groups.contains_key(&role) {
-            continue;
-        }
         if let Some(message) = build_invalid_replica_message(&validated_cluster, &role) {
             publish_warning_event(
                 &ctx,
@@ -179,8 +178,10 @@ mod test {
     use super::*;
     use crate::{
         HDFS_FULL_CONTROLLER_NAME,
-        controller::build::container::ContainerConfig,
-        test_support::{deserialize_cluster, role_group_config, validate_cluster},
+        controller::build::{RoleGroupResolver, container::ContainerConfig},
+        test_support::{
+            datanode_config, datanode_role_group_config, deserialize_cluster, validate_cluster,
+        },
     };
 
     #[test]
@@ -222,7 +223,12 @@ spec:
         let hdfs = deserialize_cluster(cr);
         let validated_cluster = validate_cluster(&hdfs);
         let role_group_name = RoleGroupName::from_str("default").unwrap();
-        let role_group_config = role_group_config(&validated_cluster, &role, &role_group_name);
+        let role_group_config = datanode_role_group_config(&validated_cluster, &role_group_name);
+        // Resolved through the production path, so this test cannot drift from what the build
+        // step actually hands the container builder.
+        let resolved = datanode_config(&validated_cluster, &role_group_name)
+            .resolve(&role_group_name, Labels::new())
+            .expect("the datanode role group should resolve");
 
         let mut pb = PodBuilder::new();
         pb.metadata(ObjectMeta::default());
@@ -232,10 +238,9 @@ spec:
             &KubernetesClusterInfo {
                 cluster_domain: DomainName::try_from("cluster.local").unwrap(),
             },
-            &role,
             &role_group_name,
             role_group_config,
-            &Labels::new(),
+            &resolved,
         )
         .unwrap();
         let containers = pb.build().unwrap().spec.unwrap().containers;
diff --git a/rust/operator-binary/src/test_support.rs b/rust/operator-binary/src/test_support.rs
index afec9dfa..ef8c6e1f 100644
--- a/rust/operator-binary/src/test_support.rs
+++ b/rust/operator-binary/src/test_support.rs
@@ -3,8 +3,13 @@ use std::str::FromStr;
 use stackable_operator::v2::types::operator::RoleGroupName;
 
 use crate::{
-    controller::{ValidatedCluster, ValidatedRoleGroupConfig, validate},
-    crd::{AnyNodeConfig, DataNodeConfig, HdfsNodeRole, v1alpha1},
+    controller::{
+        DataNodeRoleGroupConfig, JournalNodeRoleGroupConfig, NameNodeRoleGroupConfig,
+        ValidatedCluster, validate,
+    },
+    crd::{
+        CommonNodeConfig, DataNodeConfig, HdfsNodeRole, JournalNodeConfig, NameNodeConfig, v1alpha1,
+    },
 };
 
 /// The expected `app.kubernetes.io/version` label value for the given product version.
@@ -46,34 +51,69 @@ pub fn role_group_name(name: &str) -> RoleGroupName {
     RoleGroupName::from_str(name).expect("role group name should be valid")
 }
 
-pub fn role_group_config<'a>(
+pub fn namenode_role_group_config<'a>(
     validated_cluster: &'a ValidatedCluster,
-    role: &HdfsNodeRole,
     role_group_name: &RoleGroupName,
-) -> &'a ValidatedRoleGroupConfig {
+) -> &'a NameNodeRoleGroupConfig {
     validated_cluster
-        .role_groups
-        .get(role)
-        .expect("role should be defined")
+        .namenode_role_group_configs
         .get(role_group_name)
-        .expect("role group should be defined")
+        .expect("namenode role group should be defined")
 }
 
-pub fn anynode_config<'a>(
+pub fn datanode_role_group_config<'a>(
+    validated_cluster: &'a ValidatedCluster,
+    role_group_name: &RoleGroupName,
+) -> &'a DataNodeRoleGroupConfig {
+    validated_cluster
+        .datanode_role_group_configs
+        .get(role_group_name)
+        .expect("datanode role group should be defined")
+}
+
+pub fn journalnode_role_group_config<'a>(
     validated_cluster: &'a ValidatedCluster,
-    role: &HdfsNodeRole,
     role_group_name: &RoleGroupName,
-) -> &'a AnyNodeConfig {
-    &role_group_config(validated_cluster, role, role_group_name).config
+) -> &'a JournalNodeRoleGroupConfig {
+    validated_cluster
+        .journalnode_role_group_configs
+        .get(role_group_name)
+        .expect("journalnode role group should be defined")
+}
+
+pub fn namenode_config<'a>(
+    validated_cluster: &'a ValidatedCluster,
+    role_group_name: &RoleGroupName,
+) -> &'a NameNodeConfig {
+    &namenode_role_group_config(validated_cluster, role_group_name).config
 }
 
 pub fn datanode_config<'a>(
     validated_cluster: &'a ValidatedCluster,
     role_group_name: &RoleGroupName,
 ) -> &'a DataNodeConfig {
-    anynode_config(validated_cluster, &HdfsNodeRole::Data, role_group_name)
-        .as_datanode()
-        .expect("should be a DataNode")
+    &datanode_role_group_config(validated_cluster, role_group_name).config
+}
+
+pub fn journalnode_config<'a>(
+    validated_cluster: &'a ValidatedCluster,
+    role_group_name: &RoleGroupName,
+) -> &'a JournalNodeConfig {
+    &journalnode_role_group_config(validated_cluster, role_group_name).config
+}
+
+/// The merged [`CommonNodeConfig`] of one role group, for tests that assert on the settings every
+/// role shares.
+pub fn common_config<'a>(
+    validated_cluster: &'a ValidatedCluster,
+    role: &HdfsNodeRole,
+    role_group_name: &RoleGroupName,
+) -> &'a CommonNodeConfig {
+    match role {
+        HdfsNodeRole::Name => &namenode_config(validated_cluster, role_group_name).common,
+        HdfsNodeRole::Data => &datanode_config(validated_cluster, role_group_name).common,
+        HdfsNodeRole::Journal => &journalnode_config(validated_cluster, role_group_name).common,
+    }
 }
 
 /// A namenode pod `Listener` with a single ingress address, shaped as the dereference step
diff --git a/tests/templates/kuttl/cluster-operation/20-install-hdfs.yaml.j2 b/tests/templates/kuttl/cluster-operation/20-install-hdfs.yaml.j2
index 34297d70..b069afe4 100644
--- a/tests/templates/kuttl/cluster-operation/20-install-hdfs.yaml.j2
+++ b/tests/templates/kuttl/cluster-operation/20-install-hdfs.yaml.j2
@@ -26,6 +26,14 @@ spec:
     stopped: false
     reconciliationPaused: false
   nameNodes:
+    configOverrides:
+      core-site.xml:
+        # After all namenodes go at once, a deleted pod's name keeps resolving to its dead IP for
+        # up to ~30s (CoreDNS cache racing the endpoint update). format-namenodes probes it with
+        # `hdfs haadmin -getServiceState`, the TCP connect black-holes, and the default 45 x 20s
+        # retry budget outlasts this test's asserts. Whether the probe lands in that window is
+        # environment-dependent: local kind loses the race far more often than CI.
+        ipc.client.connect.max.retries.on.timeouts: "3"
     config:
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
@@ -33,6 +41,10 @@ spec:
       default:
         replicas: 2
   dataNodes:
+    configOverrides:
+      core-site.xml:
+        # See the identical nameNodes override: wait-for-namenodes runs the same haadmin probe.
+        ipc.client.connect.max.retries.on.timeouts: "3"
     config:
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
diff --git a/tests/templates/kuttl/kerberos/20-install-hdfs.txt.j2 b/tests/templates/kuttl/kerberos/20-install-hdfs.txt.j2
index 455f9630..ab1be56c 100644
--- a/tests/templates/kuttl/kerberos/20-install-hdfs.txt.j2
+++ b/tests/templates/kuttl/kerberos/20-install-hdfs.txt.j2
@@ -27,6 +27,14 @@ spec:
     vectorAggregatorConfigMapName: vector-aggregator-discovery
 {% endif %}
   nameNodes:
+    configOverrides:
+      core-site.xml:
+        # After all namenodes go at once, a deleted pod's name keeps resolving to its dead IP for
+        # up to ~30s (CoreDNS cache racing the endpoint update). format-namenodes probes it with
+        # `hdfs haadmin -getServiceState`, the TCP connect black-holes, and the default 45 x 20s
+        # retry budget outlasts this test's asserts. Whether the probe lands in that window is
+        # environment-dependent: local kind loses the race far more often than CI.
+        ipc.client.connect.max.retries.on.timeouts: "3"
     config:
       logging:
         enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
@@ -43,6 +51,10 @@ spec:
       default:
         replicas: 2
   dataNodes:
+    configOverrides:
+      core-site.xml:
+        # See the identical nameNodes override: wait-for-namenodes runs the same haadmin probe.
+        ipc.client.connect.max.retries.on.timeouts: "3"
     config:
       requestedSecretLifetime: 2d
       logging:
diff --git a/tests/templates/kuttl/smoke/30-install-hdfs.yaml.j2 b/tests/templates/kuttl/smoke/30-install-hdfs.yaml.j2
index fe2aac56..8f87c514 100644
--- a/tests/templates/kuttl/smoke/30-install-hdfs.yaml.j2
+++ b/tests/templates/kuttl/smoke/30-install-hdfs.yaml.j2
@@ -27,13 +27,11 @@ spec:
   nameNodes:
     configOverrides:
       core-site.xml:
-        # After the chaos monkey (60-unleash-the-chaosmonkey.yaml.j2) force-deletes all pods,
-        # the format-namenodes probe of the other namenode can resolve its name to the stale
-        # pre-delete pod IP (DNS cache) and silently retry a black-hole TCP connect for
-        # 45 x 20s = 15 min by default, exceeding the chaos-monkey step's 10 min `kubectl wait`.
-        # zkfc is unaffected: graceful fencing and health monitoring override their retries via
-        # ha.failover-controller.graceful-fence.connection.retries and
-        # ha.health-monitor.rpc.connect.max.retries respectively.
+        # After all namenodes go at once, a deleted pod's name keeps resolving to its dead IP for
+        # up to ~30s (CoreDNS cache racing the endpoint update). format-namenodes probes it with
+        # `hdfs haadmin -getServiceState`, the TCP connect black-holes, and the default 45 x 20s
+        # retry budget outlasts this test's asserts. Whether the probe lands in that window is
+        # environment-dependent: local kind loses the race far more often than CI.
         ipc.client.connect.max.retries.on.timeouts: "3"
     envOverrides:
       COMMON_VAR: role-value # overridden by role group below
@@ -63,8 +61,7 @@ spec:
   dataNodes:
     configOverrides:
       core-site.xml:
-        # See the identical nameNodes override: wait-for-namenodes runs the same haadmin
-        # probe against namenodes that may resolve to a stale pod IP after the chaos monkey.
+        # See the identical nameNodes override: wait-for-namenodes runs the same haadmin probe.
         ipc.client.connect.max.retries.on.timeouts: "3"
     envOverrides:
       COMMON_VAR: role-value # overridden by role group below