How to Prepare the Kafka Broker for Kafka Connect

This guide shows you how to give a Kafka Connect cluster the topics and Access Control List (ACL) entries it needs on the broker, either by letting the chart’s bootstrap Job create them or by creating them yourself.

Type

How-to guide

Goal

Prepare the Apache Kafka broker so a Kafka Connect cluster can start.

Audience

Platform Operator who can create topics and ACLs on the target Kafka cluster.

When to use

Use this guide before deploying a Kafka Connect cluster.

Contents

The sections below cover each task in this guide:

Where this guide fits

Complete this guide before running How to Deploy a Kafka Connect Cluster.

The worker principal cannot grant ACLs to itself. A superuser identity is required. See Why the worker cannot grant its own Access Control Lists for the reason.

Prerequisites

Confirm the following before you begin.

Access and permissions required

You need the following access and permissions:

  • A Kafka superuser identity, either:

    • A mutual TLS (mTLS) certificate whose Distinguished Name (DN) is listed in the broker’s super.users config, or

    • A Simple Authentication and Security Layer (SASL)/SCRAM-512 username listed in super.users.

  • The superuser identity must differ from the Connect worker identity.

  • On Strimzi-managed brokers, the worker principal must already have a KafkaUser Custom Resource (CR) or an mTLS certificate before starting this guide.

Tools and versions

You need the following tools and versions:

  • kubectl >= 1.28, with access to the target namespace.

  • kafka-topics.sh and kafka-acls.sh from the Kafka CLI (manual path only).

Resources that must exist before starting

The following resources must already exist before you start:

  • The Kafka broker is running and reachable.

  • The worker’s principal (mTLS cert DN or SASL username) exists on the broker.

  • The tenant, instance, and clusterName values are decided: the tenant and instance short names, and the Kafka Connect cluster name as you will register it in Self-Service. The chart derives the group ID and the three internal topic names from them: _<tenant>-<instance>-<clusterName>-connect for groupId, plus -configs, -offsets, and -status for the topics. If you set groupId, offsetStorageTopic, configStorageTopic, or statusStorageTopic explicitly, use those values instead. See Kafka Connect Chart Values Reference.

Self-Service provisions connector target topics (source/sink) when an App Owner creates a connector. This guide covers only the Connect-internal topics.
Never grant the worker principal Create on connector target topics. Self-Service and Platform Manager provision those topics under governance. A worker with Create lets a connector auto-create topics through Connect’s topic.creation.* settings and bypass that governance. The worker needs only Write (sinks: Read) and Describe on target topics. The internal-topic ACLs in this guide never include Create.

Choose the preparation path

Choose one of two paths:

  • Automated (recommended): The chart’s aclBootstrap Job runs as a superuser Helm hook before Strimzi starts the worker. It creates the topics and grants ACLs in a single idempotent operation. Choose this path if you can supply a Kubernetes Secret with the superuser credentials.

  • Manual: Create the topics and apply ACLs using the Kafka CLI. Choose this path when the Helm hook is not suitable for your environment (for example, in an air-gapped environment with no Helm access to the broker).

Neither path is undone by uninstalling the chart. The topics, the ACLs and the automated path’s hook resources are state Helm never manages, so Stage 1: Undo the broker preparation removes them when the cluster goes away.

Automated path: enable the aclBootstrap Helm hook Job

The chart’s aclBootstrap Job creates the internal topics and grants the worker ACLs during install.

Step 1: Create a Kubernetes Secret with the superuser credentials

The bootstrap Job authenticates to the broker as a superuser, and it reads those credentials from a Secret you create first.

Replace <VALUE> placeholders with your actual values before running any command.
Superuser mTLS certificate
kubectl create secret generic kafka-superuser \
  --from-file=tls.crt=superuser.crt \
  --from-file=tls.key=superuser.key \
  --namespace <ns>
Superuser SASL/SCRAM-512 password
kubectl create secret generic kafka-superuser \
  --from-literal=password=<superuser-password> \
  --namespace <ns>
This step is not idempotent. If the Secret already exists from a previous run, skip it or delete and recreate it.

Step 2: Add aclBootstrap values to your connect-values.yaml

Add the block that matches your superuser authentication type.

mTLS superuser
aclBootstrap:
  enabled: true
  auth:
    type: tls
    tls:
      secretName: kafka-superuser   (1)
      certificate: tls.crt
      key: tls.key
  principal: "User:CN=<worker-cert-cn>,O=<org>,C=<country>"  (2)
  replicationFactor: 3     (3)
  minInsyncReplicas: 2
1 The Secret created in step 1.
2 The full DN of the worker’s mTLS certificate. Must match exactly.
3 Match your broker’s configured replication factor. Use 1 for a single-broker test cluster. Both keys default to 1, and the chart refuses a minInsyncReplicas higher than replicationFactor.
SASL/SCRAM-512 superuser
aclBootstrap:
  enabled: true
  auth:
    type: scram-sha-512
    scram:
      username: <superuser-username>
      passwordSecret:
        secretName: kafka-superuser   (1)
        password: password            (2)
  principal: "User:<worker-scram-username>"   (3)
  replicationFactor: 3
  minInsyncReplicas: 2
1 The Secret created in step 1.
2 The key within the Secret holding the password.
3 The exact SASL username of the worker.
The aclBootstrap Job selects the bootstrap server (bootstrapServers.tls or bootstrapServers.sasl) based on auth.type. Set the matching key even when the worker uses a different listener. For example, if the worker uses SASL but you have an mTLS superuser cert, set aclBootstrap.auth.type: tls. The Job then uses bootstrapServers.tls and requires no SASL superuser account.

The Job verifies the broker TLS hostname against the certificate Subject Alternative Name (SAN). Set aclBootstrap.verifyHostname: false only when the broker certificate SAN does not include the bootstrap address, for example an internal Service name.

Prefer a superuser identity already listed in the broker super.users configuration, so the bootstrap Job needs no broker restart. Adding a new SASL/SCRAM superuser means editing super.users, and that change rolls every broker. An existing mTLS superuser certificate avoids the roll.

Step 3: Verify the Job after the chart is installed

The aclBootstrap Job runs automatically as a Helm pre-install/pre-upgrade hook. It executes when the chart is installed during deployment, which is a later procedure. Steps 1 and 2 above are the only preparation you do before deploying. Being a pre-upgrade hook, it runs again on every helm upgrade, even when nothing changed. The completed Job and its pod are kept until the next run replaces them, so the logs stay readable. helm upgrade --install blocks on the hook but does not stream its logs; run kubectl logs -f job/<cluster-name>-acl-bootstrap --namespace <ns> from a second terminal to watch it live.

After the chart is installed, you can confirm the Job succeeded from its logs:

kubectl logs job/<cluster-name>-acl-bootstrap \
  --namespace <ns> \
  --tail=60

Expected output in the final lines:

[acl-bootstrap ...] STEP 3/3: granting consumer-group ACLs (Read/Describe) on '<groupId>'
[acl-bootstrap ...] resulting ACLs for the principal(s):
...
[acl-bootstrap ...] === ACL bootstrap complete: SUCCESS ===
The Job is idempotent (--create --if-not-exists, ACL --add). Re-running the chart install with aclBootstrap.enabled: true is safe.

If the Job fails, aclBootstrap Job fails with FAILED (exit …​) lists the causes and fixes.

The Job and its ConfigMap are Helm hook resources the release does not track, so they are not removed by setting aclBootstrap.enabled: false or by helm uninstall, and neither are the topics and ACLs the Job created. See Stage 1: Undo the broker preparation for removing all of it.

Manual path: create topics and apply ACLs without the Helm hook

Run these steps yourself when you cannot use the Helm hook.

Step 1: Create the three internal topics

Create all three topics as compacted topics before the first helm install. Replace <bootstrap>, <rf>, <min-isr>, and the topic names with your values. The settings match what the aclBootstrap Job applies. The client.properties file must contain the superuser authentication configuration for the broker.

kafka-topics.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --create --if-not-exists \
  --topic <configStorageTopic> \
  --partitions 1 \
  --replication-factor <rf> \
  --config cleanup.policy=compact \
  --config min.insync.replicas=<min-isr> \
  --config segment.ms=3600000

kafka-topics.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --create --if-not-exists \
  --topic <offsetStorageTopic> \
  --partitions 25 \
  --replication-factor <rf> \
  --config cleanup.policy=compact \
  --config min.insync.replicas=<min-isr> \
  --config segment.ms=3600000

kafka-topics.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --create --if-not-exists \
  --topic <statusStorageTopic> \
  --partitions 5 \
  --replication-factor <rf> \
  --config cleanup.policy=compact \
  --config min.insync.replicas=<min-isr> \
  --config segment.ms=3600000

Each command prints Created topic <name>., or nothing when the topic already exists (--if-not-exists).

The config topic must have exactly 1 partition. This is a Kafka Connect requirement.

Step 2: Grant ACLs on the internal topics

Grant the worker read and write access to the three internal topics.

kafka-acls.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --add \
  --allow-principal "User:<worker-principal>" \
  --resource-pattern-type literal \
  --topic <configStorageTopic> \
  --topic <offsetStorageTopic> \
  --topic <statusStorageTopic> \
  --operation Read --operation Write \
  --operation Describe --operation DescribeConfigs

Each --add prints Adding ACLs for resource …​ followed by the resulting ACL list. --add is idempotent: re-adding an existing ACL is a no-op.

Step 3: Grant ACLs on the consumer group

Grant the worker access to its consumer group.

kafka-acls.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --add \
  --allow-principal "User:<worker-principal>" \
  --resource-pattern-type literal \
  --group <groupId> \
  --operation Read --operation Describe
On Strimzi-managed brokers, add spec.authorization to the KafkaUser CR instead of using the Kafka CLI directly.

Step 4: Verify the ACLs are in place

List the ACLs to confirm they were created.

kafka-acls.sh --bootstrap-server <bootstrap> \
  --command-config client.properties \
  --list \
  --principal "User:<worker-principal>"

Confirm the output includes:

  • Read, Write, Describe, and DescribeConfigs for the three internal topics.

  • Read and Describe for the consumer group.

Next steps

Return to How to Deploy a Kafka Connect Cluster and continue from the Vault setup step.