feat: restore topic selection and per-CR job service account

sionsmith · claude · sionsmith · commit 937535337c3a · 2026-06-11T06:40:35.000+01:00
Two gaps reported against multi-purpose use of the operator: KafkaRestore had no way to restore a subset of topics even though the kafka-backup CLI honours target.topics include/exclude filtering. Add spec.topics (shared TopicSelection: glob or ~-prefixed regex patterns, matched against source topic names before topicMapping renames) and emit target.topics in the generated restore config. Backup/restore Jobs run in the CR's namespace but always referenced the operator-wide service account (BACKUP_JOB_SERVICE_ACCOUNT), which is namespace-scoped — so any CR outside the operator's namespace produced Jobs whose pods could never be created. Add spec.template.pod.serviceAccountName to both CRDs, applied as a pod template override that takes precedence over the operator-wide default. Sync the regenerated CRDs into the Helm chart, document the multi-namespace service account behaviour, and add builder- and config-level regression tests for both features. Release 0.2.8. Fixes #26, #27. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,13 @@
 
 All notable changes to this project will be documented in this file.
 
+## 0.2.8 - 2026-06-11
+
+### Added
+
+- Add `spec.topics` with include/exclude glob (or `~`-prefixed regex) patterns to `KafkaRestore`, allowing specific topics to be restored from a backup. Fixes [#26](https://github.com/osodevops/strimzi-backup-operator/issues/26).
+- Add `spec.template.pod.serviceAccountName` to `KafkaBackup` and `KafkaRestore` so job pods in namespaces other than the operator's can run with a service account that exists there. Fixes [#27](https://github.com/osodevops/strimzi-backup-operator/issues/27).
+
 ## 0.2.7 - 2026-06-10
 
 ### Fixed
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "kafka-backup-operator"
-version = "0.2.7"
+version = "0.2.8"
 edition = "2021"
 rust-version = "1.88"
 license = "Apache-2.0"
diff --git a/README.md b/README.md
@@ -26,13 +26,13 @@ The **Kafka Backup Operator** solves these problems with a purpose-built Kuberne
 - **Scheduled backups** — cron-based scheduling with timezone support via Kubernetes CronJobs
 - **Point-in-time recovery (PITR)** — restore your Kafka cluster to any millisecond-precision timestamp
 - **Multi-cloud storage** — back up to Amazon S3, Azure Blob Storage, Google Cloud Storage, or any S3-compatible store (MinIO, Ceph RGW)
-- **Topic filtering** — include/exclude topics using regex patterns
+- **Topic filtering** — include/exclude topics using glob or regex patterns, for both backup and restore
 - **Topic mapping** — rename topics during restore for migration or testing scenarios
 - **Consumer group offset restore** — restore consumer group offsets with optional group remapping
 - **Retention policies** — automatic pruning of old backups by count or age
 - **Compression** — gzip, snappy, lz4, or zstd compression for storage efficiency
 - **Prometheus metrics** — built-in observability with backup/restore counters, duration histograms, and storage gauges
-- **Pod template customisation** — full control over backup/restore Job pods (affinity, tolerations, host aliases, security context, environment variables)
+- **Pod template customisation** — full control over backup/restore Job pods (affinity, tolerations, host aliases, service account, security context, environment variables)
 - **Azure Workload Identity** — native support for passwordless Azure authentication
 
 ## Architecture
@@ -148,6 +148,11 @@ spec:
     name: my-cluster
   backupRef:
     name: my-cluster-backup
+  topics:
+    include:
+      - "orders-*"          # glob, or "~orders-\d+" for regex
+    exclude:
+      - "*-internal"
   pointInTime:
     timestamp: "2026-02-12T14:30:00.000Z"
   logging:
@@ -294,6 +299,26 @@ spec:
 | `resources.limits.cpu` | CPU limit | `500m` |
 | `resources.limits.memory` | Memory limit | `512Mi` |
 
+### Job service accounts across namespaces
+
+Backup and restore Jobs run in the namespace of the `KafkaBackup`/`KafkaRestore`
+resource, and by default reference the service account named by
+`backupJobs.serviceAccountName` (falling back to the operator's own service
+account). Service accounts are namespace-scoped, so for resources created
+outside the operator's namespace, set `spec.template.pod.serviceAccountName`
+to a service account that exists in that namespace:
+
+```yaml
+spec:
+  template:
+    pod:
+      serviceAccountName: kafka-backup-jobs
+```
+
+The service account only needs to exist (job pods don't call the Kubernetes
+API), but it should carry any workload-identity annotations (IRSA, Azure
+Workload Identity) your storage backend requires.
+
 ## Logging
 
 The operator deployment and the backup/restore job pods are configured separately.
diff --git a/config/examples/kafka-restore-pitr.yaml b/config/examples/kafka-restore-pitr.yaml
@@ -18,6 +18,14 @@ spec:
     name: daily-backup
     backupId: "backup-20260213-020000"
 
+  # Restore only selected topics from the backup. Glob patterns by default;
+  # prefix with ~ for regex. Matched against source topic names, before
+  # topicMapping renames. Omit to restore every topic in the backup.
+  topics:
+    include:
+      - "orders"
+      - "payments"
+
   pointInTime:
     timestamp: "2026-02-13T01:30:00.000Z"
 
@@ -56,6 +64,9 @@ spec:
 
   template:
     pod:
+      # Service account for the restore job pod; must exist in this namespace.
+      # Defaults to the operator-wide backupJobs.serviceAccountName.
+      serviceAccountName: kafka-backup-jobs
       hostAliases:
         - ip: "10.10.0.5"
           hostnames:
diff --git a/deploy/crds/kafkabackups.yaml b/deploy/crds/kafkabackups.yaml
@@ -718,6 +718,10 @@ spec:
                         description: Pod security context (pass-through to k8s PodSecurityContext)
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
+                      serviceAccountName:
+                        description: Service account for job pods. Must exist in the namespace of this resource. Overrides the operator-wide default (BACKUP_JOB_SERVICE_ACCOUNT).
+                        nullable: true
+                        type: string
                       tolerations:
                         description: Pod tolerations (pass-through to k8s Tolerations)
                         items:
diff --git a/deploy/crds/kafkarestores.yaml b/deploy/crds/kafkarestores.yaml
@@ -516,6 +516,10 @@ spec:
                         description: Pod security context (pass-through to k8s PodSecurityContext)
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
+                      serviceAccountName:
+                        description: Service account for job pods. Must exist in the namespace of this resource. Overrides the operator-wide default (BACKUP_JOB_SERVICE_ACCOUNT).
+                        nullable: true
+                        type: string
                       tolerations:
                         description: Pod tolerations (pass-through to k8s Tolerations)
                         items:
@@ -540,6 +544,21 @@ spec:
                   - targetTopic
                   type: object
                 type: array
+              topics:
+                description: Topics to restore, selected by include/exclude glob (or `~`-prefixed regex) patterns matched against source topic names in the backup. All topics are restored when omitted. Filtering is applied before topicMapping renames.
+                nullable: true
+                properties:
+                  exclude:
+                    description: Glob patterns for topics to exclude
+                    items:
+                      type: string
+                    type: array
+                  include:
+                    description: Glob patterns for topics to include
+                    items:
+                      type: string
+                    type: array
+                type: object
             required:
             - backupRef
             - strimziClusterRef
diff --git a/deploy/helm/strimzi-backup-operator/Chart.yaml b/deploy/helm/strimzi-backup-operator/Chart.yaml
@@ -2,8 +2,8 @@ apiVersion: v2
 name: strimzi-backup-operator
 description: Kubernetes operator for Kafka backup and restore, integrated with Strimzi
 type: application
-version: 0.2.7
-appVersion: "0.2.7"
+version: 0.2.8
+appVersion: "0.2.8"
 home: https://github.com/osodevops/strimzi-backup-operator
 sources:
   - https://github.com/osodevops/strimzi-backup-operator
diff --git a/deploy/helm/strimzi-backup-operator/crds/kafkabackups.yaml b/deploy/helm/strimzi-backup-operator/crds/kafkabackups.yaml
@@ -718,6 +718,10 @@ spec:
                         description: Pod security context (pass-through to k8s PodSecurityContext)
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
+                      serviceAccountName:
+                        description: Service account for job pods. Must exist in the namespace of this resource. Overrides the operator-wide default (BACKUP_JOB_SERVICE_ACCOUNT).
+                        nullable: true
+                        type: string
                       tolerations:
                         description: Pod tolerations (pass-through to k8s Tolerations)
                         items:
diff --git a/deploy/helm/strimzi-backup-operator/crds/kafkarestores.yaml b/deploy/helm/strimzi-backup-operator/crds/kafkarestores.yaml
@@ -516,6 +516,10 @@ spec:
                         description: Pod security context (pass-through to k8s PodSecurityContext)
                         type: object
                         x-kubernetes-preserve-unknown-fields: true
+                      serviceAccountName:
+                        description: Service account for job pods. Must exist in the namespace of this resource. Overrides the operator-wide default (BACKUP_JOB_SERVICE_ACCOUNT).
+                        nullable: true
+                        type: string
                       tolerations:
                         description: Pod tolerations (pass-through to k8s Tolerations)
                         items:
@@ -540,6 +544,21 @@ spec:
                   - targetTopic
                   type: object
                 type: array
+              topics:
+                description: Topics to restore, selected by include/exclude glob (or `~`-prefixed regex) patterns matched against source topic names in the backup. All topics are restored when omitted. Filtering is applied before topicMapping renames.
+                nullable: true
+                properties:
+                  exclude:
+                    description: Glob patterns for topics to exclude
+                    items:
+                      type: string
+                    type: array
+                  include:
+                    description: Glob patterns for topics to include
+                    items:
+                      type: string
+                    type: array
+                type: object
             required:
             - backupRef
             - strimziClusterRef
diff --git a/deploy/helm/strimzi-backup-operator/values.yaml b/deploy/helm/strimzi-backup-operator/values.yaml
@@ -25,8 +25,11 @@ serviceAccount:
   name: ""
 
 backupJobs:
-  # Service account used by backup and restore job pods.
+  # Default service account used by backup and restore job pods.
   # If empty, the chart passes the operator service account name.
+  # The account must exist in each namespace where KafkaBackup/KafkaRestore
+  # resources are created; individual resources can override it with
+  # spec.template.pod.serviceAccountName.
   serviceAccountName: ""
 
 # Azure Workload Identity configuration
diff --git a/src/adapters/backup_config.rs b/src/adapters/backup_config.rs
@@ -286,7 +286,7 @@ fn build_backup_options(opts: &crate::crd::kafka_backup::BackupOptionsSpec) -> R
     Ok(Value::Mapping(config))
 }
 
-fn build_topic_selection(topics: &crate::crd::common::TopicSelection) -> Value {
+pub(crate) fn build_topic_selection(topics: &crate::crd::common::TopicSelection) -> Value {
     let mut topic_config = serde_yaml::Mapping::new();
     if !topics.include.is_empty() {
         topic_config.insert(
diff --git a/src/adapters/restore_config.rs b/src/adapters/restore_config.rs
@@ -6,6 +6,7 @@ use crate::strimzi::kafka_cr::ResolvedKafkaCluster;
 use crate::strimzi::kafka_user::ResolvedAuth;
 use crate::strimzi::tls::ResolvedTlsCerts;
 
+use super::backup_config::build_topic_selection;
 use super::logging_config::build_logging_config;
 use super::storage_config::build_storage_config;
 
@@ -32,7 +33,13 @@ pub fn build_restore_config_yaml(
     );
 
     // Target (Kafka cluster)
-    let target = build_kafka_config(cluster, tls_certs, auth, restore.spec.connection.as_ref())?;
+    let target = build_kafka_config(
+        cluster,
+        tls_certs,
+        auth,
+        restore.spec.topics.as_ref(),
+        restore.spec.connection.as_ref(),
+    )?;
     config.insert(Value::String("target".to_string()), target);
 
     // Storage (from source backup CR)
@@ -75,6 +82,7 @@ fn build_kafka_config(
     cluster: &ResolvedKafkaCluster,
     _tls_certs: &Option<ResolvedTlsCerts>,
     auth: &ResolvedAuth,
+    topics: Option<&crate::crd::common::TopicSelection>,
     connection: Option<&crate::crd::common::KafkaConnectionSpec>,
 ) -> Result<Value> {
     let mut kafka = serde_yaml::Mapping::new();
@@ -137,6 +145,15 @@ fn build_kafka_config(
         Value::Mapping(security),
     );
 
+    if let Some(topics) = topics {
+        let topic_config = build_topic_selection(topics);
+        if let Value::Mapping(ref m) = topic_config {
+            if !m.is_empty() {
+                kafka.insert(Value::String("topics".to_string()), topic_config);
+            }
+        }
+    }
+
     if let Some(connection) = connection {
         let connection_config = build_connection_config(connection);
         if let Value::Mapping(ref m) = connection_config {
diff --git a/src/crd/common.rs b/src/crd/common.rs
@@ -407,6 +407,10 @@ pub struct PodOverrides {
     /// Pod host aliases (pass-through to k8s HostAlias)
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub host_aliases: Vec<HostAlias>,
+    /// Service account for job pods. Must exist in the namespace of this resource.
+    /// Overrides the operator-wide default (BACKUP_JOB_SERVICE_ACCOUNT).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub service_account_name: Option<String>,
 }
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize, JsonSchema)]
diff --git a/src/crd/kafka_restore.rs b/src/crd/kafka_restore.rs
@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 
 use super::common::{
     AuthenticationSpec, Condition, KafkaConnectionSpec, LoggingSpec, MetricsSpec, PodTemplateSpec,
-    ResourceRequirementsSpec, RestoreInfo, StrimziClusterRef,
+    ResourceRequirementsSpec, RestoreInfo, StrimziClusterRef, TopicSelection,
 };
 
 /// KafkaRestore defines a restore operation from a KafkaBackup to a Strimzi-managed Kafka cluster.
@@ -35,6 +35,12 @@ pub struct KafkaRestoreSpec {
     /// Reference to the source backup
     pub backup_ref: BackupRef,
 
+    /// Topics to restore, selected by include/exclude glob (or `~`-prefixed regex)
+    /// patterns matched against source topic names in the backup. All topics are
+    /// restored when omitted. Filtering is applied before topicMapping renames.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub topics: Option<TopicSelection>,
+
     /// Point-in-time recovery settings
     #[serde(skip_serializing_if = "Option::is_none")]
     pub point_in_time: Option<PointInTimeSpec>,
diff --git a/src/jobs/templates.rs b/src/jobs/templates.rs
@@ -353,6 +353,13 @@ pub fn apply_pod_template(pod_spec: &mut PodSpec, template: Option<&CrdPodTempla
         if !pod_overrides.host_aliases.is_empty() {
             pod_spec.host_aliases = Some(pod_overrides.host_aliases.clone());
         }
+
+        if let Some(sa) = &pod_overrides.service_account_name {
+            let sa = sa.trim();
+            if !sa.is_empty() {
+                pod_spec.service_account_name = Some(sa.to_string());
+            }
+        }
     }
 
     if let Some(container_overrides) = &tmpl.container {
diff --git a/tests/integration/backup_test.rs b/tests/integration/backup_test.rs
diff --git a/tests/integration/restore_test.rs b/tests/integration/restore_test.rs
