---
title: "Connect to a secured Kafka cluster"
description: "Configure monedula-gitops for SASL/SCRAM, SASL_SSL, mutual TLS or OAUTHBEARER, and connect it to Schema Registry and Confluent MDS."
url: https://monedula.dev/flock/docs/gitops/how-to/connect-to-a-secured-cluster/
---

A `KafkaCluster` tells monedula-gitops how to reach your brokers and, when you use them, Schema Registry and Confluent MDS. The CLI reads it from the file you pass with `--cluster-config-file`, and the operator reads it as a custom resource.

## Before you start

- The bootstrap address of the listener you connect to, the authentication method it expects, and credentials for the principal that monedula-gitops connects as.
- For TLS with a private certificate authority, the CA certificate in PEM format.

A super-user principal is the quickest start, because it bypasses ACL enforcement. A principal without super-user rights needs the ACLs for every kind your manifests declare, and the operator's principal also needs `Describe` on `Cluster`; [Least-privilege connecting principal](/flock/docs/gitops/reference/connecting-principal-acls/) lists them. Such a principal can also lock itself out: on a broker with `allow.everyone.if.no.acl.found=true`, a topic is open only while it has no ACLs, so once monedula-gitops creates a topic's ACLs, a connecting principal they do not list loses access to that topic. The CLI and the operator warn when the desired ACLs leave out the connecting principal, but they cannot see the broker's `super.users`.

## Describe the broker connection

Every `KafkaCluster` needs `spec.bootstrapServers`, a comma-separated list of `host:port` pairs. TLS and authentication are optional. Your manifests refer to the cluster by its `metadata.name`, in `clusterRef.name`.

This example connects over TLS and authenticates with SCRAM-SHA-512, the SASL_SSL setup. The CLI and the operator read the credentials from different places:

<Tabs group="connect-to-a-secured-cluster-mode" labels={['CLI', 'Operator']} label="Where you run it">

```yaml
apiVersion: gitops.monedula.dev/v1alpha1
kind: KafkaCluster
metadata:
  name: prod
spec:
  bootstrapServers: broker1:9093,broker2:9093
  tls:
    enabled: true
    # Private CA? Point caCert at the CA bundle (omit to use the system trust store):
    caCert:
      valueFrom:
        file: certs/ca.crt
  auth:
    mechanism: SCRAM-SHA-512   # or SCRAM-SHA-256, PLAIN, None
    scram:
      username:
        valueFrom:
          env: KAFKA_USER
      password:
        valueFrom:
          file: /etc/secrets/kafka-password
```

```yaml
apiVersion: gitops.monedula.dev/v1alpha1
kind: KafkaCluster
metadata:
  name: prod
  namespace: monedula-system
spec:
  bootstrapServers: kafka-bootstrap.kafka.svc:9093
  tls:
    enabled: true
    # caCert:
    #   valueFrom:
    #     secretKeyRef:
    #       name: kafka-ca
    #       key: ca.crt
  auth:
    mechanism: SCRAM-SHA-512
    scram:
      username:
        valueFrom:
          secretKeyRef:
            name: kafka-admin-credentials
            key: username
      password:
        valueFrom:
          secretKeyRef:
            name: kafka-admin-credentials
            key: password
```

</Tabs>

- `tls.enabled: true` turns on TLS and verifies the brokers against the system trust store. Set `tls.caCert` to trust a private CA instead. `insecureSkipVerify: true` switches off server certificate verification; keep it to development.
- `auth.mechanism` is one of `None`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`, `OAUTHBEARER` or `mTLS`. The `scram` block holds the username and password for the SCRAM mechanisms and for `PLAIN`.
- For a SASL_PLAINTEXT listener, leave out the `tls` block.

Every credential and certificate field is a `valueFrom` reference: `env` or `file` for the CLI, `secretKeyRef` for the operator, and `inline` for both, as [One model, two runtimes](/flock/docs/gitops/concepts/one-model-two-runtimes/) lists. Neither runtime logs resolved values. With the operator, create the `KafkaCluster` in the namespace of the resources that reference it, or in the [`clusterNamespace`](/flock/docs/gitops/how-to/configure-the-operator/).

## Authenticate with a client certificate instead

For mutual TLS, set `auth.mechanism: mTLS` and give the client certificate and key under `tls`:

```yaml
apiVersion: gitops.monedula.dev/v1alpha1
kind: KafkaCluster
metadata:
  name: shared
spec:
  bootstrapServers: localhost:9092
  tls:
    enabled: true
    caCert:     { valueFrom: { file: certs/ca.crt } }
    clientCert: { valueFrom: { file: certs/client.crt } }
    clientKey:  { valueFrom: { file: certs/client.key } }
  auth:
    mechanism: mTLS
```

`mTLS` requires `tls.enabled: true`, `clientCert` and `clientKey` must be set together, and `auth.scram` and `auth.oauth` must be absent. With the operator, point the three fields at a Secret with `secretKeyRef`.

## Authenticate with OAUTHBEARER instead

monedula-gitops uses the OIDC client-credentials flow: it requests a token from `tokenEndpoint` with the client ID and secret.

This example is the CLI cluster config of the runnable scenarios' local development profile, which issues tokens from a mock OIDC server on `localhost:8080` over plain HTTP. For a real cluster, set `tokenEndpoint` to your identity provider's HTTPS token endpoint. With the operator, `clientId` and `clientSecret` come from a Secret through `secretKeyRef`.

```yaml
apiVersion: gitops.monedula.dev/v1alpha1
kind: KafkaCluster
metadata:
  name: shared
spec:
  bootstrapServers: localhost:9092
  auth:
    mechanism: OAUTHBEARER
    oauth:
      tokenEndpoint: http://localhost:8080/default/token
      clientId:     { valueFrom: { env: OAUTH_CLIENT_ID } }
      clientSecret: { valueFrom: { env: OAUTH_CLIENT_SECRET } }
      scope: kafka
```

`tokenEndpoint` is required and `scope` is optional. If the identity provider serves the token endpoint with a certificate from a private CA, set `oauth.tokenEndpointCA` to that CA. `tls.caCert` is never used for the identity provider. `auth.scram` must be absent.

## Add Schema Registry

If any `KafkaTopic` on this cluster declares a schema, add a `spec.schemaRegistry` block:

```yaml
spec:
  schemaRegistry:
    endpoint: "https://sr.example.com:8081"
    auth:
      type: basic
      username: { valueFrom: { env: SR_USER } }
      password: { valueFrom: { env: SR_PASSWORD } }
    tls:
      enabled: true
      # Private CA? Point caCert at the CA bundle (omit to use the system trust store):
      caCert:
        valueFrom:
          file: certs/sr-ca.crt
      # Optional TLS client certificate (set clientCert and clientKey together):
      # clientCert: { valueFrom: { file: certs/sr-client.crt } }
      # clientKey:  { valueFrom: { file: certs/sr-client.key } }
```

`endpoint` is required when the block is present, and `basic` is the only `auth.type`. `schemaRegistry.tls` works like `spec.tls`.

## Add Confluent MDS

To manage Confluent RBAC role bindings, add the Metadata Service connection under `spec.authorization`:

```yaml
authorization:
  mds:
    endpoint: "https://mds.example.com:8090"
    auth:
      type: basic
      username: { valueFrom: { env: MDS_USERNAME } }
      password: { valueFrom: { env: MDS_PASSWORD } }
    clusters:
      kafkaCluster: "abc123"
      schemaRegistryCluster: "sr-abc"   # required for scope.type: schema-registry
      connectCluster: "connect-1"       # required for scope.type: connect
      ksqlCluster: "ksql-1"             # required for scope.type: ksql
```

`endpoint` and `clusters.kafkaCluster`, the Kafka cluster ID, are required. `auth.type` is `basic`, `bearer` with a `token`, or `mtls` with the client certificate from `authorization.mds.tls`; the MDS connection is independent of `spec.auth`. [Manage Confluent RBAC role bindings](/flock/docs/gitops/how-to/manage-confluent-rbac/) continues from here.

## Check it worked

<Tabs group="connect-to-a-secured-cluster-check" labels={['CLI', 'Operator']} label="Where you run it">

```bash
monedula-gitops doctor --cluster-config-file ./clusters/prod.yaml
```

```bash
kubectl -n monedula-system get kafkacluster prod -o jsonpath='{.status.conditions}'
```

</Tabs>

`doctor` exits 0 when every check passes and never changes anything; [Check connectivity with doctor](/flock/docs/gitops/how-to/check-connectivity-with-doctor/) explains each check. When the file defines several clusters, select one with `--cluster <name>`.

With the operator, the `KafkaCluster` reports the conditions `Ready`, `ClusterReachable` and `Authenticated` with status `True`, plus `SchemaRegistryReachable` when it configures a registry, and reaches the phase `Ready`. If a Secret reference does not resolve, the phase is `Error`, `ClusterReachable` is `False`, and its message names the problem.

## Related

- The [KafkaCluster reference](/flock/docs/gitops/reference/resources/kafka-cluster/) lists every connection field.
- [Supported platforms and versions](/flock/docs/gitops/concepts/supported-platforms/) covers Confluent Cloud and the tested broker versions.
- Runnable versions: [scenarios/04-sasl-ssl](https://github.com/monedula-dev/monedula-gitops/tree/main/scenarios/04-sasl-ssl), [scenarios/20-mtls](https://github.com/monedula-dev/monedula-gitops/tree/main/scenarios/20-mtls) and [scenarios/21-oauthbearer](https://github.com/monedula-dev/monedula-gitops/tree/main/scenarios/21-oauthbearer), with their cluster configs in [scenarios/clusters](https://github.com/monedula-dev/monedula-gitops/tree/main/scenarios/clusters).