labrenbe · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026 · Sep 24, 2026
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -13,6 +13,12 @@ All notable changes to this project will be documented in this file.
 - Support for Kafka 4.3.1 ([#1022]).
 - Add `/ready` endpoint to the operator Deployment, which reports the CRD installation status ([#1029]).
 - Brokers now have a default affinity to the OPA Pods when OPA authorization is configured ([#1035]).
+- Add management access via `spec.clusterConfig.managementAccess`. When enabled, the operator
+  deploys an agent per KafkaCluster, which authenticates anonymously or with a TLS client
+  certificate ([#1028]).
+- Add the KafkaTopic CRD to manage topics declaratively. The agent creates and adopts topics,
+  replaces their configs with `configOverrides`, increases their partitions and optionally deletes
+  them. The operator removes the finalizers of KafkaTopics whose KafkaCluster is deleted ([#1050]).
 - The OPA authorizer is now also configured on KRaft controllers. Previously, requests that brokers
   forward to the controllers (e.g. creating or deleting topics) were not authorized at all.
   The Kafka nodes are now configured as super users, so that traffic between them does not hit OPA.
@@ -57,6 +63,9 @@ All notable changes to this project will be documented in this file.
 - Support Vector 0.58.0 ([#1039], [#1044], [#1046]).
 - test: Update OPA to 1.20.2 in integration tests and docs example ([#1042]).
 - The operator Deployment runs with a restricted security context and a read-only root filesystem, so it can be deployed into namespaces that enforce the `restricted` Pod Security Standard ([#1049]).
+- The `/ready` endpoint of the operator now also waits for the KafkaTopic CRD. If the CRDs are not
+  maintained by the operator (`maintenance.customResourceDefinitions.maintain: false`), install the
+  KafkaTopic CRD together with the KafkaCluster CRD ([#1050]).
 
 ### Fixed
 
@@ -93,6 +102,7 @@ All notable changes to this project will be documented in this file.
 [#1021]: https://github.com/stackabletech/kafka-operator/pull/1021
 [#1022]: https://github.com/stackabletech/kafka-operator/pull/1022
 [#1024]: https://github.com/stackabletech/kafka-operator/pull/1024
+[#1028]: https://github.com/stackabletech/kafka-operator/pull/1028
 [#1029]: https://github.com/stackabletech/kafka-operator/pull/1029
 [#1034]: https://github.com/stackabletech/kafka-operator/pull/1034
 [#1035]: https://github.com/stackabletech/kafka-operator/pull/1035
@@ -104,6 +114,7 @@ All notable changes to this project will be documented in this file.
 [#1046]: https://github.com/stackabletech/kafka-operator/pull/1046
 [#1047]: https://github.com/stackabletech/kafka-operator/pull/1047
 [#1049]: https://github.com/stackabletech/kafka-operator/pull/1049
+[#1050]: https://github.com/stackabletech/kafka-operator/pull/1050
 [#1051]: https://github.com/stackabletech/kafka-operator/pull/1051
 [#1052]: https://github.com/stackabletech/kafka-operator/pull/1052
 

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.nix b/Cargo.nix
diff --git a/Cargo.toml b/Cargo.toml
@@ -11,6 +11,7 @@ repository = "https://github.com/stackabletech/kafka-operator"
 
 [workspace.dependencies]
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.122.0", features = ["agent", "crds", "webhook"] }
+stackable-kafka-admin = { path = "rust/kafka-admin" }
 stackable-kafka-shared = { path = "rust/shared" }
 
 anyhow = "1.0"
@@ -23,6 +24,7 @@ const_format = "0.2"
 futures = "0.3"
 indoc = "2.0"
 kafka-protocol = { version = "0.18", default-features = false, features = ["broker", "client"] }
+regex = "1.13"
 rstest = "0.28"
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "std", "tls12"] }
 semver = "1.0"

diff --git a/deploy/helm/kafka-operator/Chart.yaml b/deploy/helm/kafka-operator/Chart.yaml
@@ -42,3 +42,8 @@ annotations:
       name: kafkaclusters.kafka.stackable.tech
       displayName: KafkaCluster
       description: A Kafka cluster stacklet. This resource is managed by the Stackable operator for Apache Kafka.
+    - kind: KafkaTopic
+      version: v1alpha1
+      name: kafkatopics.kafka.stackable.tech
+      displayName: KafkaTopic
+      description: A Kafka topic, created and updated by the agent of the referenced Kafka cluster.
diff --git a/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml b/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml
@@ -8,6 +8,30 @@ metadata:
   labels:
   {{- include "kafka-operator.labels" . | nindent 4 }}
 rules:
+  # KafkaTopics of the agent's cluster: watched, and their finalizer added and removed.
+  - apiGroups:
+      - {{ include "kafka-operator.name" . }}.stackable.tech
+    resources:
+      - kafkatopics
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+  - apiGroups:
+      - {{ include "kafka-operator.name" . }}.stackable.tech
+    resources:
+      - kafkatopics/status
+    verbs:
+      - patch
+  # Required to report reconciliation errors as events on the KafkaTopic objects.
+  - apiGroups:
+      - events.k8s.io
+    resources:
+      - events
+    verbs:
+      - create
+      - patch
 {{ if .Capabilities.APIVersions.Has "security.openshift.io/v1" }}
   # On OpenShift, the agent pod must be allowed to use the nonroot-v2 SCC
   - apiGroups:

diff --git a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml
@@ -128,6 +128,24 @@ rules:
       - {{ include "kafka-operator.name" . }}clusters/status
     verbs:
       - patch
+  # KafkaTopics: watched to take over while no agent is available, and their finalizer removed
+  # once the KafkaCluster is gone.
+  - apiGroups:
+      - {{ include "kafka-operator.name" . }}.stackable.tech
+    resources:
+      - kafkatopics
+    verbs:
+      - get
+      - list
+      - watch
+      - patch
+  # Status subresource: conditions set while the agent is unavailable.
+  - apiGroups:
+      - {{ include "kafka-operator.name" . }}.stackable.tech
+    resources:
+      - kafkatopics/status
+    verbs:
+      - patch
   # Read authentication class configuration referenced in the KafkaCluster spec.
   - apiGroups:
       - authentication.stackable.tech

diff --git a/docs/modules/kafka/pages/index.adoc b/docs/modules/kafka/pages/index.adoc
@@ -5,6 +5,7 @@
 :github: https://github.com/stackabletech/kafka-operator/
 :crd: {crd-ref}{crd-ref-release}&operator=kafka-operator
 :crd-kafkacluster: {crd-ref}/kafka.stackable.tech/v1alpha1/kafkacluster{crd-ref-release}
+:crd-kafkatopic: {crd-ref}/kafka.stackable.tech/v1alpha1/kafkatopic{crd-ref-release}
 :feature-tracker: https://features.stackable.tech/unified
 :metadata-quorum: https://cwiki.apache.org/confluence/display/KAFKA/KIP-500%3A+Replace+ZooKeeper+with+a+Self-Managed+Metadata+Quorum
 :pegelonline: https://www.pegelonline.wsv.de/webservice/ueberblick
@@ -75,3 +76,4 @@ include::partial$supported-versions.adoc[]
 * The {github}[kafka-operator {external-link-icon}^] GitHub repository
 * The operator feature overview in the {feature-tracker}[feature tracker {external-link-icon}^]
 * The {crd-kafkacluster}[KafkaCluster {external-link-icon}^] CRD documentation
+* The {crd-kafkatopic}[KafkaTopic {external-link-icon}^] CRD documentation
diff --git a/docs/modules/kafka/pages/usage-guide/management-access.adoc b/docs/modules/kafka/pages/usage-guide/management-access.adoc
@@ -0,0 +1,132 @@
+= Management access
+:description: Let the Stackable Operator for Apache Kafka manage resources inside a Kafka cluster, such as topics, through a per-cluster agent.
+
+Management access allows the Stackable Data Platform (SDP) to manage resources inside an Apache Kafka cluster, such as xref:usage-guide/topics.adoc[topics].
+
+== Agents
+
+The agent is an optional Deployment next to the Kafka cluster that manages resources inside Kafka, such as topics.
+The operator deploys it when management access is enabled.
+
+The agent runs the image of the operator, pulled with the pull secrets of the Kafka image in `spec.image.pullSecrets`.
+If the operator image is not public, these pull secrets must grant access to its registry as well, or the Deployment of the agent gets its own pull secrets with <<agent-resources,object overrides>>.
+
+== Limitations
+
+Management access is not available for all clusters:
+
+* Clusters that use ZooKeeper are not supported, only clusters in xref:usage-guide/kraft-controller.adoc[KRaft mode].
+* Clusters with Kerberos authentication are not supported.
+
+== Configuration
+
+Enable management access with `spec.clusterConfig.managementAccess` and choose how the agent authenticates to Kafka:
+
+[source,yaml]
+----
+apiVersion: kafka.stackable.tech/v1alpha1
+kind: KafkaCluster
+metadata:
+  name: simple-kafka
+spec:
+  image:
+    productVersion: 4.3.1
+  clusterConfig:
+    metadataManager: KRaft
+    tls:
+      serverSecretClass: tls
+    authentication:
+      - authenticationClass: kafka-client-tls
+    managementAccess:
+      enabled: true # <1>
+      authentication:
+        userProvided:
+          tls:
+            secretClassName: kafka-client-auth-secret # <2>
+  controllers:
+    roleGroups:
+      default:
+        replicas: 3
+  brokers:
+    roleGroups:
+      default:
+        replicas: 3
+----
+<1> Deploys the agent. Setting it to `false` removes the agent again.
+<2> The SecretClass that provisions the client certificate of the agent, here the same SecretClass that the brokers use to verify client certificates, see xref:usage-guide/security.adoc[Security].
+
+The agent supports the following authentication methods:
+
+`anonymous: {}`:: The agent connects without a client certificate.
+This is only possible if the cluster does not require client authentication.
+`tls.secretClassName`:: A client certificate provisioned by the xref:secret-operator:index.adoc[Secret Operator] from the given SecretClass.
+`tls.secretName`:: A static Secret holding the client certificate in the keys `tls.crt` and `tls.key` (PEM), for example a Secret of type `kubernetes.io/tls`.
+
+The agent verifies the brokers with the CA of the SecretClass used by the client listener, that is `spec.clusterConfig.tls.serverSecretClass` or the `clientCertSecretClass` of the TLS AuthenticationClass.
+
+== Authorization
+
+The principal of the agent must be allowed to manage topics in Kafka.
+The principal is derived from the subject of the client certificate.
+The agent needs the following permissions:
+
+[cols="2,3", options="header"]
+|===
+| Kafka API | Operation
+
+| Metadata | `DESCRIBE` on the topic
+| CreateTopics | `CREATE` on the cluster or the topic, and `DESCRIBE_CONFIGS` on the topic
+| CreatePartitions | `ALTER` on the topic
+| DescribeConfigs | `DESCRIBE_CONFIGS` on the topic
+| IncrementalAlterConfigs | `ALTER_CONFIGS` on the topic
+| DeleteTopics | `DELETE` on the topic
+| ListPartitionReassignments | `DESCRIBE` on the cluster
+|===
+
+NOTE: Without `DESCRIBE_CONFIGS`, Kafka still creates a topic, but the agent cannot observe it.
+The KafkaTopic then reports `KafkaRejected`.
+
+== Network
+
+The agent connects to the headless Services of the broker role groups, which resolve to the broker Pods.
+The traffic of the agent therefore stays within the Kubernetes cluster, regardless of the xref:usage-guide/listenerclass.adoc[ListenerClasses] of the cluster.
+
+[[agent-resources]]
+== Agent resources
+
+The operator creates the following resources for the agent, named after the KafkaCluster:
+
+* A Deployment `<cluster>-agent` with a single replica, limited to 500m CPU and 128Mi memory by default.
+* A ServiceAccount `<cluster>-agent`.
+* A RoleBinding `<cluster>-agent`, which binds the ClusterRole `kafka-agent-clusterrole` deployed by the Helm chart.
+
+The agent reports not ready while it cannot list the KafkaTopics of its cluster, and the operator then reports `AgentUnavailable` after a minute.
+
+The Deployment can be adjusted with xref:concepts:overrides.adoc#object-overrides[object overrides], for example to change the resources of the agent:
+
+[source,yaml]
+----
+spec:
+  objectOverrides:
+    - apiVersion: apps/v1
+      kind: Deployment
+      metadata:
+        name: simple-kafka-agent
+      spec:
+        template:
+          spec:
+            containers:
+              - name: agent
+                resources:
+                  limits:
+                    memory: 256Mi
+----
+
+The default resources of the agent are sufficient for a few thousand topics.
+The operator keeps all KafkaTopics of all clusters it watches in memory and reconciles them whenever it starts, so with many thousands of topics, increase its CPU and memory in the Helm values of the operator.
+
+== Disabling management access
+
+Setting `enabled` to `false` or removing `managementAccess` removes the agent.
+The KafkaTopics and the topics in Kafka stay as they are.
+While no agent is available, the operator reports `AgentUnavailable` in the conditions of the KafkaTopics of the cluster, see xref:usage-guide/topics.adoc#deletion[Deletion] for the effect on deleted KafkaTopics.