How to Use an External Secrets Manager for Kafka Credentials in Ververica Platform 3 with Kubernetes Secrets

In short: Keep Kafka credentials in your external secrets manager, whether that is Azure Key Vault, HashiCorp Vault, AWS Secrets Manager, Google Cloud Secret Manager, Doppler or another one, and use them in Flink SQL jobs on Ververica Platform 3 without storing them in the platform. Sync the username and password into a Kubernetes Secret, mount it through the JobManager and TaskManager pod templates, and let the Kafka client read the files with a config provider. Verified end to end on Ververica Platform 3.1.5.

Credentials belong in your secrets manager

Enterprises do not keep credentials in the tools that use them. They keep them in an external secrets manager: Azure Key Vault, HashiCorp Vault, AWS Secrets Manager, Google Cloud Secret Manager, Doppler or a comparable service. Access policies, audit trails and rotation live there, under the control of the security team.

On Kubernetes, these secrets managers deliver secrets to workloads the same way. A sync operator, such as External Secrets Operator, the Doppler Kubernetes Operator or the Vault Secrets Operator, writes the entry into a Kubernetes Secret, and the pod reads it from there. The Kubernetes Secret is the common contract, whatever the secrets manager behind it.

Flink SQL on Ververica Platform 3 does not consume that contract directly. The platform has its own credential store, secret values, referenced as ${secret_values.<key>}. Using it means copying every credential from the secrets manager into the platform and keeping two copies in sync. Flink SQL has no syntax to reference a Kubernetes Secret in the WITH clause, and an environment variable built from one does not reach it either.

This post closes that gap for Kafka. Any credential that exists as a Kubernetes Secret in the job namespace, synced from any secrets manager, authenticates Flink SQL Kafka sources and sinks, and Ververica Platform never stores the password. We verified every step end to end on Ververica Platform 3.1.5 with the engine vera-4.5-jdk11-flink-1.20 (VVR 4.5.4) against a SASL-protected Apache Kafka broker.

Prerequisites

• Ververica Platform 3.1.5 or later.
• A Kafka cluster that authenticates clients with SASL.
• A sync operator of your secrets manager that writes Kubernetes Secrets into the namespace of your deployment target. For a first test, permission to create a Secret there is enough.
• Edit rights on the deployment YAML in Ververica Platform.

How it works

The credential travels from the secrets manager, whichever one you use, to the Kafka client on its own path. Ververica Platform deploys the pods and renders the pod templates, but it never holds or forwards the password: the Kubernetes Secret is the hand-off point between your secrets setup and Flink.

The pod templates mount the Secret into the Flink container on the JobManager (Kafka source enumerator) and on the TaskManagers (readers and writers). The Kafka client resolves the ${dir:...} placeholders once, when it starts.

Ververica Platform does not touch the placeholder. Its own substitution only matches plain names such as ${secret_values.my_password}, so ${dir:...} reaches the Kafka client unchanged. The platform stores and renders only the reference, never the password. Platform secret values and file placeholders can live side by side in the same job.

Step 1: Let your secrets manager create the Kubernetes Secret

You do not create this Secret by hand. Your secrets manager already holds the Kafka username and password, and its Kubernetes operator creates the Secret from them and keeps it in sync. The table further down lists the operator for each secrets manager. The pattern needs one result: a Secret in the Kubernetes namespace where your Flink jobs run, which is the namespace of your deployment target, with the username and the password as separate keys. The secrets manager holds plain values only, nothing Kafka-specific.

With Doppler, for example, a DopplerSecret resource tells the Doppler Kubernetes Operator which project and config to sync and where to create the Secret:

apiVersion: secrets.doppler.com/v1alpha1
kind: DopplerSecret
metadata:
  name: kafka-credentials
  namespace: doppler-operator-system
spec:
  tokenSecret:
    name: doppler-token-secret
    namespace: doppler-operator-system
  project: <Doppler project>
  config: <Doppler config>
  managedSecret:
    name: kafka-credentials
    namespace: <job namespace>
    type: Opaque

The operator creates the Secret kafka-credentials in the job namespace if it does not exist, and checks Doppler for changes every 60 seconds by default. The DopplerSecret sits in the operator namespace doppler-operator-system, because only resources there may manage Secrets in other namespaces. This example follows Syncing secrets in the Doppler documentation; the operator setup was not part of our test.

Mind the key names. The Doppler operator uses the Doppler secret names as keys, upper case by default, for example KAFKA_USERNAME and KAFKA_PASSWORD. Either use those names in the placeholders in Step 3, or rename them during the sync with the name transformer or the processors that Doppler documents. Other operators let you choose the key names in their own resource. Sync only what the job needs, because every key of the Secret becomes a file in the Flink pods.

Whatever operator you use, the Secret that arrives in the job namespace has this shape. This post uses the keys username and password:

apiVersion: v1
kind: Secret
metadata:
  name: kafka-credentials
type: Opaque
stringData:
  username: <user>
  password: <password>

Each key becomes one file in the pod, and the Kafka client inserts the file content exactly as it is. The values must not end with a newline. A trailing newline in the password breaks the JAAS line, and the Kafka client fails with Value not specified for key ';' in JAAS config. If the username is not sensitive, keep only the password in the Secret and write the username in the SQL.

For a first test without a sync operator, you can apply this Secret yourself with kubectl apply. That is what we did in our test. Ververica Platform and the Kafka client only ever see the resulting Secret, however it was created.

Step 2: Mount the Secret on the JobManager and the TaskManagers

Add the mount to the deployment through the JobManager and TaskManager pod templates. In the deployment YAML this sits under spec.template.spec:

kubernetes:
  jobManagerPodTemplate:
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
            - name: kafka-secrets
              mountPath: /etc/kafka-secrets
              readOnly: true
      volumes:
        - name: kafka-secrets
          secret:
            secretName: kafka-credentials
  taskManagerPodTemplate:
    spec:
      containers:
        - name: flink-main-container
          volumeMounts:
            - name: kafka-secrets
              mountPath: /etc/kafka-secrets
              readOnly: true
      volumes:
        - name: kafka-secrets
          secret:
            secretName: kafka-credentials

The mounts marked platform are a selection of those Ververica Platform 3.1.5 rendered in our test; /etc/kafka-secrets is the only one the template adds.

Three details matter:

• The container name must be exactly flink-main-container. Ververica Platform merges a template container with that name into the Flink container it generates, so your mount is added next to the platform's own mounts. Any other name creates a separate sidecar, and the Flink process would not see the file.
• Set both templates. The Kafka source enumerator runs on the JobManager and also logs in to the broker. A mount on the TaskManagers alone is not enough.
• Use the pod templates, not kubernetes.pods.volumeMounts. The pod-template route is the one we verified for this pattern on Ververica Platform 3.1.5.

You only add the volume and the mount. The image, resources and the platform's volumes stay as Ververica Platform renders them.

Step 3: Reference the file in the Kafka table

Add three options to every Kafka table that should use the credential. The JAAS line stays in the SQL; only the username and the password come from the mounted files. The same options work for sources and sinks:

CREATE TABLE orders (
  order_id BIGINT,
  amount   DECIMAL(18, 2)
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'broker-1:9092',
  'properties.security.protocol' = 'SASL_PLAINTEXT',
  'properties.sasl.mechanism' = 'PLAIN',
  'properties.config.providers' = 'dir',
  'properties.config.providers.dir.class' =
    'org.apache.flink.kafka.shaded.org.apache.kafka.common.config.provider.DirectoryConfigProvider',
  'properties.sasl.jaas.config' =
    'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="${dir:/etc/kafka-secrets:username}" password="${dir:/etc/kafka-secrets:password}";',
  'properties.group.id' = 'orders-reader',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

• properties.config.providers gives the provider an alias, here dir.
• properties.config.providers.dir.class names the provider class, again with the shaded prefix.
• ${dir:/etc/kafka-secrets:password} reads the file password in the folder /etc/kafka-secrets, which is the mount path plus the Secret key from Steps 1 and 2. The Kafka client replaces each placeholder inside the JAAS line.

The provider class and the login module carry the prefix org.apache.flink.kafka.shaded., because the Kafka client inside the VVR Kafka connector is shaded. If your jobs already set sasl.jaas.config today, keep the login module class you use there.

The Kafka connector of VVR 4.5.4 bundles Kafka client 3.4.0. That client ships two config providers: DirectoryConfigProvider, which reads one file per key and fits Secret volumes, and FileConfigProvider, which reads keys from one properties file. It does not include Kafka's EnvVarConfigProvider, so environment variables are not an alternative on this engine.

This example uses SASL_PLAINTEXT with the PLAIN mechanism. SASL_SSL, SCRAM and OAUTHBEARER use the same Secret mount and the same placeholders. The section Other security protocols and SASL mechanisms shows the options we verified for each of them.

Step 4: Verify that it works

Start the deployment and run these checks. All of them passed in our test on Ververica Platform 3.1.5.

Other security protocols and SASL mechanisms

The Secret, the pod templates and the config provider do not depend on the security protocol or the SASL mechanism. Only a few table options change, and every secret value still comes from a mounted file. We ran each of the following combinations on Ververica Platform 3.1.5, on the source and on the sink, with records flowing end to end:

TLS with SASL_SSL

Add the CA certificate that signed the broker certificate to the same Secret as one more key, in PEM format:

stringData:
 username: <user>
 password: <password>
 ca.crt: |
   -----BEGIN CERTIFICATE-----
   <CA certificate>
   -----END CERTIFICATE-----

Then switch the protocol and point the truststore to the mounted file. The truststore location is a normal file path, so it needs no placeholder:

'properties.security.protocol' = 'SASL_SSL',
'properties.ssl.truststore.type' = 'PEM',
'properties.ssl.truststore.location' = '/etc/kafka-secrets/ca.crt',

The JAAS line stays as in Step 3. We tested the PEM file from the Secret. Without the two truststore options the Kafka client falls back to the default truststore of the JVM, which is standard Kafka client behavior and fits brokers with a certificate from a public certificate authority; that route was not part of our test.

SCRAM

SCRAM uses the same username and password files. Change the mechanism and the login module:

'properties.sasl.mechanism' = 'SCRAM-SHA-512',
'properties.sasl.jaas.config' =
 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="${dir:/etc/kafka-secrets:username}" password="${dir:/etc/kafka-secrets:password}";',

We tested SCRAM-SHA-512. SCRAM-SHA-256 uses the same login module with the mechanism name SCRAM-SHA-256.

OAUTHBEARER

With OAUTHBEARER the Secret holds the client ID and the client secret of an OAuth client instead of a Kafka username and password. The Kafka client uses them to request a token from your identity provider with the client credentials grant, and renews the token before it expires:

'properties.sasl.mechanism' = 'OAUTHBEARER',
'properties.sasl.login.callback.handler.class' =
  'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler',
'properties.sasl.oauthbearer.token.endpoint.url' = '<token endpoint of your identity provider>',
'properties.sasl.jaas.config' =
  'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="${dir:/etc/kafka-secrets:client_id}" clientSecret="${dir:/etc/kafka-secrets:client_secret}";',

• The token endpoint must be reachable from the JobManager and the TaskManager pods.
• In our test the identity provider was Keycloak and the tokens were valid for five minutes. The Kafka clients renewed their tokens on their own, and the job kept processing records after the first token had expired.
• Our token endpoint used plain HTTP inside the cluster. An HTTPS token endpoint with a certificate from a private CA was not part of our test.
• The client secret follows the same rules as a password: no trailing newline in the Secret, and a job restart after a rotation.

Mutual TLS with client certificates and Kerberos (GSSAPI) were not part of our test. They build on the same mechanism, files from the mounted Secret, so the same approach should work for them as well: a PEM keystore for mutual TLS, and a keytab that the JAAS line references by its path for Kerberos. Verify them in your environment before you rely on them. Cloud IAM mechanisms such as AWS MSK IAM are a different case: they use no static secret and need an additional login library, so they are outside this pattern.

Secrets managers, rotation and security

Any secrets manager works. Ververica Platform only needs a Kubernetes Secret in the job namespace. Which secrets manager holds the master copy does not matter, so you keep the one you have and its access policies. Most ship an operator that syncs their entries into Kubernetes Secrets, and External Secrets Operator covers many of them with one tool:

Whatever the secrets manager, Steps 2 to 4 stay exactly the same. Our test covered the Kubernetes Secret part; the sync operators are documented by their vendors and were not part of our test. A csi volume follows the same pod-template structure but was not tested either.

Rotation. The Kafka client resolves the placeholder when it is created, and a running client keeps the credential it started with. In our test we changed the Secret while the job ran. Kubernetes refreshed the mounted file in the running pods within about 30 seconds, without a pod restart, and the next task restart created Kafka clients with the new credential. The refresh delay depends on your cluster. For a controlled rotation, update the credential in your secrets manager, let the sync operator update the Secret, then restart the job from a savepoint. Keep the old credential valid on the broker until the restart is done.

Security notes.

• Anyone who can read Secrets or exec into pods in the job namespace can read the credential. Restrict get secrets and pods/exec there.
• Only the config providers built into the connector's shaded Kafka client are available. Third-party Kafka config providers for Vault or Key Vault implement the unshaded Kafka interface and will not load as they are.

Limits

This pattern covers Kafka only. Config providers are a Kafka client feature, so they work for every table that uses the Kafka client, but not for other connectors.

• JDBC databases, object storage (S3, Azure Blob) and catalogs have no file-based mechanism in Flink SQL. Keep using Ververica Platform secret values for them, ideally filled from your secrets manager by an automated sync step.
• One identity per table. Each table can point to its own file, so different tables in one job can use different Kafka users. We verified this with a source and a sink that logged in as two different users in one job.
• Not covered by our test: mutual TLS with client certificates, Kerberos (GSSAPI), csi volumes, the sync operators themselves, and Validate in the SQL Editor with the ${dir:...} value. Mutual TLS, Kerberos and csi volumes use the same mounted-file mechanism, so the same approach should work for them as well. Test them in your own environment before you rely on them.

Summary

An external secrets manager, a Kubernetes Secret and three Kafka table options are enough to keep Kafka credentials out of Ververica Platform. The checklist:

• Sync operator writes a Secret with username and password into the job namespace, values without a trailing newline
• jobManagerPodTemplate and taskManagerPodTemplate both mount it, container name flink-main-container
• Kafka table sets properties.config.providers, the shaded DirectoryConfigProvider class and a JAAS line with ${dir:<mount path>:<key>} placeholders
• Password absent from JobManager and TaskManager logs
• Restart planned after each credential rotation

Tested on Ververica Platform 3.1.5, engine vera-4.5-jdk11-flink-1.20 (VVR 4.5.4, Kafka client 3.4.0 in the connector), against an Apache Kafka 3.9.1 broker with SASL_PLAINTEXT and SASL_SSL listeners, the mechanisms PLAIN, SCRAM-SHA-512 and OAUTHBEARER, and Keycloak 26.0.8 as the identity provider for OAUTHBEARER.

Further reading

• Pod-level customization in Ververica Platform 3: jobManagerPodTemplate, taskManagerPodTemplate and flink-main-container
• Secret values in Ververica Platform: the platform's own credential store
• KIP-297: Externalizing Secrets for Connect Configurations: the Kafka proposal that introduced config providers and the ${provider:path:key} syntax
• Doppler Kubernetes Operator and External Secrets Operator with Doppler
• Secrets Store CSI driver: secret auto rotation