KafkaBackup with SCRAM-SHA-512 authentication tries to connect via TLS

[MishelSirotkin]

Issue Description:

Describe the bug

When configuring a KafkaBackup resource with scram-sha-512 authentication, the operator still generates a job that attempts to connect to the Kafka cluster over TLS using SASL_SSL. Since the user does not have a client certificate (only a password), the broker rejects the connection with a CertificateRequired error.

This makes it impossible to use scram-sha-512 authentication unless the Kafka cluster is configured to allow TLS connections without requiring client certificates.

Expected behavior

When authentication.type is set to scram-sha-512, the operator should generate a job that connects to the Kafka cluster using SASL_PLAINTEXT (or at least provide a way to configure the bootstrap server and security protocol) so that the connection can succeed using only a username and password.

Steps to reproduce

1. Deploy a Strimzi Kafka cluster with TLS enabled.
2. Create a KafkaUser with authentication.type: scram-sha-512.
3. Create a KafkaBackup resource referencing that user, with authentication.type: scram-sha-512 and passwordSecret.
4. Wait for the scheduled backup job to run.
5. Check the backup job logs.

Actual logs

From the kafka-backup job:

2026-07-07 15:03:12 2026-07-07T10:03:12.359791Z INFO kafka_backup::commands::backup: Loading configuration from: /config/backup.yaml
2026-07-07 15:03:12 2026-07-07T10:03:12.360554Z INFO kafka_backup::commands::backup: Starting backup: kafka-daily-backup-scheduled-1783418494
2026-07-07 15:03:12 2026-07-07T10:03:12.360886Z INFO kafka_backup_core::metrics::server: Metrics server listening on http://0.0.0.0:8080
2026-07-07 15:03:12 2026-07-07T10:03:12.372752Z DEBUG kafka_backup_core::kafka::client: TCP keepalive enabled for kafka-kafka-bootstrap.ecpk-dbaas-sirotkinme.svc:9093: time=60s, interval=20s
2026-07-07 15:03:12 2026-07-07T10:03:12.372802Z DEBUG kafka_backup_core::kafka::client: Establishing TLS connection to kafka-kafka-bootstrap.ecpk-dbaas-sirotkinme.svc:9093
2026-07-07 15:03:12 2026-07-07T10:03:12.373234Z DEBUG kafka_backup_core::kafka::tls: Loaded 1 certificate(s) from /certs/cluster-ca/ca.crt
2026-07-07 15:03:12 2026-07-07T10:03:12.373463Z DEBUG kafka_backup_core::kafka::tls: Loaded 1 CA certificate(s) from /certs/cluster-ca/ca.crt
2026-07-07 15:03:12 2026-07-07T10:03:12.373484Z DEBUG kafka_backup_core::kafka::tls: Configuring TLS without client authentication
2026-07-07 15:03:12 2026-07-07T10:03:12.380327Z DEBUG kafka_backup_core::kafka::client: TLS connection established to kafka-kafka-bootstrap.ecpk-dbaas-sirotkinme.svc:9093
2026-07-07 15:03:12 Error: Kafka error: Protocol error: Failed to read response length: received fatal alert: CertificateRequired
2026-07-07 15:03:12
2026-07-07 15:03:12 Caused by:
2026-07-07 15:03:12 Protocol error: Failed to read response length: received fatal alert: CertificateRequired

From the generated ConfigMap:

mode: backup
backup_id: ${BACKUP_ID}
source:
  bootstrap_servers:
  - kafka-kafka-bootstrap.ecpk-dbaas-sirotkinme.svc:9093
  security:
    security_protocol: SASL_SSL 
    ssl_ca_location: /certs/cluster-ca/ca.crt
    sasl_mechanism: SCRAM-SHA512
    sasl_username: kafka-super-user
    sasl_password: ${KAFKA_SASL_PASSWORD}

kind Kafkabackup

apiVersion: kafkabackup.com/v1alpha1
kind: KafkaBackup
metadata:
  annotations:
    meta.helm.sh/release-name: kafka
    meta.helm.sh/release-namespace: ecpk-dbaas-sirotkinme
  labels:
    app.cpaas.io/name: kafka.ecpk-dbaas-sirotkinme
    app.kubernetes.io/instance: kafka
    app.kubernetes.io/managed-by: Helm
    app.kubernetes.io/part-of: strimzi-kafka
    app.kubernetes.io/version: 0.1.0
    helm.sh/chart: kafka-helm-ecpk-1.0.0-test
    strimzi.io/cluster: kafka
  name: kafka-daily-backup
spec:
  authentication:
    passwordSecret:
      key: password
      name: kafka-super-user
    type: scram-sha-512
    username: kafka-super-user
  backup:
    compression: zstd
    parallelism: 4
    segmentSize: 268435456
    stopAtCurrentOffsets: true
  env:
    - name: RUST_LOG
      value: kafka_backup=debug,rdkafka=debug
  image: kafka-backup:v0.15.8
  logging:
    format: json
    level: debug
    modules:
      kafka_backup: debug
      rdkafka: debug
  resources:
    limits:
      cpu: 500m
      memory: 1Gi
    requests:
      cpu: 500m
      memory: 1Gi
  retention:
    maxAge: 3d
    maxBackups: 5
    pruneOnSchedule: true
  schedule:
    cron: 0 2 * * *
    suspend: false
    timezone: UTC
  storage:
    s3:
      accessKeySecret:
        key: access-key-id
        name: aws-credentials
      allowHttp: true
      bucket: kafka-b-dev-backup
      endpoint: http://172.30.49.126:8080
      forcePathStyle: true
      prefix: kafka-backup
      region: eu-west-1
      secretKeySecret:
        key: secret-access-key
        name: aws-credentials
    type: s3
  strimziClusterRef:
    name: kafka
  template:
    pod:
      serviceAccountName: kafka-kafka
  topics:
    exclude:
      - "*-internal"
      - __*
    include:
      - "*"

Cronjob

apiVersion: batch/v1
kind: CronJob
metadata:
  labels:
    app.kubernetes.io/instance: kafka-daily-backup
    app.kubernetes.io/managed-by: kafka-backup-operator
    app.kubernetes.io/name: kafka-backup-operator
    app.kubernetes.io/part-of: kafka-backup
    kafkabackup.com/backup: kafka-daily-backup
    kafkabackup.com/type: backup
    strimzi.io/cluster: kafka
  name: kafka-daily-backup-scheduled
spec:
  concurrencyPolicy: Forbid
  failedJobsHistoryLimit: 3
  jobTemplate:
    metadata:
      creationTimestamp: null
      labels:
        app.kubernetes.io/instance: kafka-daily-backup
        app.kubernetes.io/managed-by: kafka-backup-operator
        app.kubernetes.io/name: kafka-backup-operator
        app.kubernetes.io/part-of: kafka-backup
        kafkabackup.com/backup: kafka-daily-backup
        kafkabackup.com/type: backup
        strimzi.io/cluster: kafka
    spec:
      backoffLimit: 3
      template:
        metadata:
          creationTimestamp: null
          labels:
            app.kubernetes.io/instance: kafka-daily-backup
            app.kubernetes.io/managed-by: kafka-backup-operator
            app.kubernetes.io/name: kafka-backup-operator
            app.kubernetes.io/part-of: kafka-backup
            kafkabackup.com/backup: kafka-daily-backup
            kafkabackup.com/type: backup
            strimzi.io/cluster: kafka
        spec:
          containers:
            - args:
                - backup
                - --config
                - /config/backup.yaml
              command:
                - kafka-backup
              env:
                - name: KAFKA_SASL_PASSWORD
                  valueFrom:
                    secretKeyRef:
                      key: password
                      name: kafka-super-user
                - name: AWS_ACCESS_KEY_ID
                  valueFrom:
                    secretKeyRef:
                      key: access-key-id
                      name: aws-credentials
                - name: AWS_SECRET_ACCESS_KEY
                  valueFrom:
                    secretKeyRef:
                      key: secret-access-key
                      name: aws-credentials
                - name: BACKUP_ID
                  valueFrom:
                    fieldRef:
                      apiVersion: v1
                      fieldPath: metadata.labels['batch.kubernetes.io/job-name']
                - name: RUST_LOG
                  value: kafka_backup=debug,rdkafka=debug
              image: kafka-backup:v0.15.8
              imagePullPolicy: IfNotPresent
              name: backup
              resources:
                limits:
                  cpu: 500m
                  memory: 1Gi
                requests:
                  cpu: 500m
                  memory: 1Gi
              terminationMessagePath: /dev/termination-log
              terminationMessagePolicy: File
              volumeMounts:
                - mountPath: /config
                  name: config
                  readOnly: true
                - mountPath: /certs/cluster-ca
                  name: cluster-ca
                  readOnly: true
                - mountPath: /certs/user
                  name: user-certs
                  readOnly: true
          dnsPolicy: ClusterFirst
          restartPolicy: Never
          schedulerName: default-scheduler
          securityContext: {}
          serviceAccount: kafka-kafka
          serviceAccountName: kafka-kafka
          terminationGracePeriodSeconds: 30
          volumes:
            - configMap:
                defaultMode: 420
                name: kafka-daily-backup-config
              name: config
            - name: cluster-ca
              secret:
                defaultMode: 420
                items:
                  - key: ca.crt
                    path: ca.crt
                secretName: kafka-cluster-ca-cert
            - name: user-certs
              secret:
                defaultMode: 420
                items:
                  - key: password
                    path: password
                secretName: kafka-super-user
  schedule: 0 2 * * *
  successfulJobsHistoryLimit: 3
  suspend: false
  timeZone: UTC

Environment (please complete the following information):

• strimzi-backup-operator version: [e.g., v0.2.11]
• Strimzi version: [e.g., 0.51.0]
• Kubernetes distribution: [e.g., OpenShift, Vanilla K8s]
• Storage backend: S3 (MinIO)

---

[sionsmith]

@MishelSirotkin This is fixed in v0.2.12 (PR #38) — could you upgrade and retest?

What changed: the operator now picks the Kafka listener whose authentication.type matches your KafkaBackup/KafkaRestore authentication, instead of always preferring the first TLS listener. With type: scram-sha-512, your jobs will connect through your SCRAM listener (SASL_PLAINTEXT for a plaintext listener, SASL_SSL for a SCRAM-over-TLS one — no client certificate needed in either case). Among matching listeners it prefers in-cluster, TLS-encrypted ones.

You also get explicit control if you ever need it:

spec:
  strimziClusterRef:
    name: kafka
    listener: <listener-name>   # optional: force a specific listener from your Kafka CR

When upgrading, please also apply the updated CRDs — helm upgrade does not update CRDs, and the new spec.strimziClusterRef.listener field needs them:

kubectl apply -f https://raw.githubusercontent.com/osodevops/strimzi-backup-operator/v0.2.12/deploy/crds/kafkabackups.yaml
kubectl apply -f https://raw.githubusercontent.com/osodevops/strimzi-backup-operator/v0.2.12/deploy/crds/kafkarestores.yaml

Then delete the failed Job so the next scheduled run picks up the regenerated config. One thing to check: your Kafka CR needs a listener with authentication.type: scram-sha-512 — if none exists, the operator now reports a clear NoCompatibleListener condition on the KafkaBackup status listing your cluster's listeners instead of generating a config that can't work.

Verified end-to-end against a cluster with the same topology as yours (an mTLS listener alongside a SCRAM listener): backup and restore both complete, and the exact CertificateRequired failure from your logs is gone. Please reopen if you still hit trouble on v0.2.12.

[MishelSirotkin]

@sionsmith, thank you very much for the quick fix! I really appreciate your time and effort on this.

I'll test the changes in our environment over the next few days and let you know if everything works as expected. If I run into any issues, I'll follow up here.

Thanks again for the great support!

[MishelSirotkin]

All ok. Thanks !