Secure Flink on Kubernetes with HashiCorp Vault: The Complete mTLS Integration Guide for Confluent Flink

In the world of real-time data streaming, securing Apache Flink workloads on Kubernetes , especially when they interact with Confluent Platform and Kafka, it requires a thoughtful approach to secrets management and mutual TLS (mTLS) authentication. If you’re operating Flink pipelines that process high-volume data streams and connect to Kafka brokers over a secure, mutually authenticated channel, managing certificates securely is non-negotiable.

In my previous post, Beyond the Broker: A 101 on Deploying Confluent Platform Flink on Kubernetes, we covered the foundational deployment of Confluent Manager for Apache Flink (CMF) on Kubernetes including installing the Flink Kubernetes Operator, deploying CMF via Helm, creating compute pools, and integrating with Confluent Control Center for visibility and management. That setup gives you a scalable, declarative way to run Flink workloads.

However, once your Flink jobs are live and talking to Kafka, the next critical layer is security: ensuring strong encryption and mutual authentication without exposing secrets in Helm values, Git, or Kubernetes etcd. This follow-up guide builds directly on that deployment by showing how to integrate HashiCorp Vault for dynamic, secure delivery of Kafka client certificates (keystore and truststore) to Flink pods using the Secrets Store CSI Driver , enabling production-grade mTLS end-to-end.

This article provides a comprehensive, step-by-step implementation for integrating HashiCorp Vault with Confluent Platform Flink on Kubernetes. It covers Vault installation, certificate management, Kubernetes authentication configuration, and secrets delivery to Flink workloads (JobManagers, TaskManagers, and compute pools).

Let’s begin with a Quick Architecture Overview

This integration follows a common production pattern for external Vault + Kubernetes workloads:

• Vault Server A dedicated, external Vault instance (running outside the Kubernetes cluster) typically on a separate VM, bare-metal server, or managed Vault cluster. Vault serves as the single source of truth for sensitive material (in this case, Kafka client certificates).
• Kubernetes Cluster The runtime environment for Confluent Platform components (including Flink JobManagers, TaskManagers, and compute pools). Vault communicates with the Kubernetes API server to validate pod identities during the authentication process.
• Target Namespace The Kubernetes namespace where Confluent Platform Flink workloads are deployed.
• Secret Storage Paths in Vault Logical paths inside Vault where the Kafka client keystore and truststore are stored.

The end-to-end flow is elegant and secure: A Flink pod authenticates to Vault using its Kubernetes ServiceAccount token. Vault validates this identity against the Kubernetes API, applies policies, and via the CSI Driver delivers decoded JKS files as mounted volumes. These files enable mTLS for Kafka connections, ensuring both encryption and bidirectional authentication.

Prerequisites

Before we start, ensure you have the following in place:

• A Dedicated server for Vault installation, with network connectivity to your Kubernetes cluster.
• A Kubernetes cluster running version 1.21 or higher, with administrative access.
• Helm 3.x installed and configured.
• The kubectl CLI tool properly set up to interact with your cluster.
• SSL certificates in the form of keystore and truststore JKS files for Kafka connectivity (these are the client certificates for mTLS).

With these ready, we can move forward confidently.

Phase 1: Installing and Initializing Vault

Step 1: Install Vault on a dedicated server

sudo dnf update -y
sudo dnf install -y yum-utils
sudo yum-config-manager --add-repo https://rpm.releases.hashicorp.com/RHEL/hashicorp.repo
sudo dnf install vault -y

Step 2: Basic Configuration & Service

sudo vi /etc/vault.d/vault.hcl
# Note: Configure the listener address, storage backend, and other 
# parameters as per yourenvironment requirements.
sudo systemctl start vault
sudo systemctl enable vault
sudo systemctl status vault

Step 3: Initialize Vault (Do this ONCE)

export VAULT_ADDR='http://<ip-address:vault_port>'
vault operator init

Save the 5 unseal keys and the root token in a password manager. This is critical.

Why Shamir’s Secret Sharing? Vault splits the master key into 5 pieces using Shamir’s algorithm. Any 3 can unseal. One person can’t compromise the whole vault.

Step 4: Unseal

vault operator unseal <key1>
vault operator unseal <key2>
vault operator unseal <key3>
vault status   
# Should show Sealed: false

Step 5: Login

export VAULT_TOKEN="hvs.your-root-token"
vault login

A Note on External Vault Placement: Running Vault on a separate VM (rather than inside Kubernetes) is a common and recommended pattern for production. It provides physical and logical separation, reduces the blast radius if the cluster is compromised, and allows for easier scaling and backup strategies independent of Kubernetes.

Phase 2: Storing Your mTLS Certificates in Vault

Enable KV v2 Secrets Engine

vault secrets enable -path=test kv-v2
vault secrets list

Why KV v2? Versioning, soft deletes, and metadata. Perfect for certs that rotate.

Store the Certificates (Base64 is key!)

base64 -w 0 client-keystore.jks > client-keystore.b64
base64 -w 0 client-truststore.jks > client-truststore.b64

vault kv put test/flink/client-keystore keystore="$(cat client-keystore.b64)"
vault kv put test/flink/client-truststore truststore="$(cat client-truststore.b64)"

Verify:

vault kv get test/flink/client-keystore
vault kv list test/flink/

Pro tip: Store under test/flink/ so you can have prod/flink/, dev/flink/, etc.

Understanding Certificate Storage in Vault: The base64 encoding step is essential because JKS files are binary. Storing them directly as text would corrupt them. KV v2’s versioning is particularly valuable here , when you rotate certificates (a best practice for security), Vault retains previous versions, allowing rollback if needed. This hierarchical path structure (test/flink/) also simplifies policy management later, as we can scope permissions to specific environments.

Phase 3: Kubernetes Authentication (The Secure Bridge)

This phase establishes the trusted relationship between Vault and Kubernetes. It is one of the most critical parts of the integration, as it governs how workloads prove their identity.

Enable Kubernetes Auth Method

vault auth enable kubernetes

Create Service Account & Non-Expiring Token

kubectl create serviceaccount vault-auth -n confluent

cat > vault-auth-token-secret.yaml <<EOF
apiVersion: v1
kind: Secret
metadata:
  name: vault-auth-token-secret
  namespace: confluent
  annotations:
    kubernetes.io/service-account.name: vault-auth
type: kubernetes.io/service-account-token
EOF

kubectl apply -f vault-auth-token-secret.yaml
sleep 5

kubectl get secret vault-auth-token-secret -n confluent \
  -o jsonpath='{.data.token}' | base64 -d > /opt/vault-auth-token.jwt

Why a Secret-based token instead of kubectl create token? The token from a Secret never expires. No rotation headaches.

While the non-expiring reviewer token is a long-lived credential on the Vault server. If compromised, an attacker could use it to validate arbitrary service account tokens for an extended period, increasing the potential blast radius.

Mitigation recommendations Rotate the token periodically (every 30–90 days) via automated Kubernetes CronJob that regenerates it, updates Vault’s auth/kubernetes/config, and securely replaces the file. Combine with: strict filesystem permissions (SELinux/AppArmor), encrypted storage if feasible, and auditing/alerting on unusual TokenReview activity in Vault and Kubernetes audit logs.

This approach preserves operational simplicity while meaningfully reducing long-term exposure.

Extract K8s CA Certificate

kubectl config view --raw --minify --flatten \
  -o jsonpath='{.clusters[0].cluster.certificate-authority-data}' \
  | base64 -d > /opt/ca.crt

Grant Token Review Permissions

cat <<EOF | kubectl apply -f -
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: vault-auth-delegator
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: system:auth-delegator
subjects:
- kind: ServiceAccount
  name: vault-auth
  namespace: confluent
EOF

Deep Dive into the Vault-Kubernetes Authentication Flow: To fully appreciate this setup, let’s walk through what happens when a Flink pod requests secrets from Vault:

1. The pod’s container (or the CSI provider on its behalf) presents the ServiceAccount JWT token (automatically mounted at /var/run/secrets/kubernetes.io/serviceaccount/token).
2. Vault forwards this token to the Kubernetes API server using the token_reviewer_jwt we provided.
3. The API server validates the token against the CA certificate and confirms the ServiceAccount’s identity and namespace.
4. Vault then checks the configured role (flink-role) to see if the ServiceAccount is authorized.
5. If everything matches, Vault issues a short-lived Vault token and applies the associated policy.

This flow leverages Kubernetes’ native identity system, eliminating the need for long-lived Vault tokens in pods. The non-expiring reviewer token is a deliberate choice for external Vault deployments , it avoids the operational burden of token refresh while maintaining security through the least-privilege ClusterRoleBinding.

Phase 4: Policies & Roles (Enforcing Least Privilege)

Create Policy

cat > /opt/flink-policy.hcl <<EOF
path "test/data/flink/*" {
  capabilities = ["read", "list"]
}
path "test/metadata/flink/*" {
  capabilities = ["read", "list"]
}
EOF

vault policy write flink-policy /opt/flink-policy.hcl

Create Role

vault write auth/kubernetes/role/flink-role \
  bound_service_account_names=vault-auth,confluent-sa,flink-sa,default \
  bound_service_account_namespaces=confluent \
  policies=flink-policy \
  ttl=1h \
  max_ttl=24h

Why TTLs? Defense in depth. Even if a token leaks, it dies quickly.

The Role of Policies and Roles in Security: Vault policies act as the “what” (permissions to specific paths), while roles act as the “who” (mapping Kubernetes identities to policies). The data/ and metadata/ prefixes are KV v2-specific: data/ holds the actual secret values, and metadata/ provides audit information. This granular control ensures that Flink workloads can only read their designated secrets — nothing more. The TTL settings add an extra layer: tokens are automatically revoked after a short period, reducing exposure windows.

Phase 5: CSI Driver — The Secret Delivery System

Install Repos & Drivers

helm repo add hashicorp https://helm.releases.hashicorp.com
helm repo add secrets-store-csi-driver https://kubernetes-sigs.github.io/secrets-store-csi-driver/charts
helm repo update

helm install csi-secrets-store secrets-store-csi-driver/secrets-store-csi-driver \
  --namespace kube-system

helm install vault hashicorp/vault \
  --namespace confluent \
  --set "csi.enabled=true" \
  --set "server.dev.enabled=true" \
  --set "injector.enabled=false" \
  --set "vault.address=http://<IP-ADDRESS:PORT"

Why CSI over Agent Injector?

• No sidecar → less resource usage
• Native volume mounts → zero code change in Flink
• Works with any workload (Deployments, StatefulSets, Flink compute pools)

engineering.clearroute.io

How the Secrets Store CSI Driver Works in Depth: The CSI (Container Storage Interface) framework allows Kubernetes to treat secrets as volumes without custom code. The generic Secrets Store CSI Driver (in kube-system) interfaces with the kubelet to handle volume mounting. The Vault-specific provider (in the confluent namespace) handles Vault communication. When a pod starts:

1. The kubelet detects the CSI volume in the pod spec.
2. It calls the CSI driver.
3. The driver invokes the Vault provider.
4. The provider authenticates using the pod’s ServiceAccount.
5. Vault validates and returns secrets.
6. The provider decodes (base64) and writes files to a tmpfs volume.
7. The files are mounted into the pod.

This architecture is lightweight, scalable, and supports multiple backends (Vault, AWS Secrets Manager, etc.). Disabling the injector avoids resource overhead from sidecar containers.

Phase 6: Testing the Full Flow

Create a test pod:

kubectl run vault-test --rm -it --image=vault:latest \
  --serviceaccount=vault-auth -n confluent -- sh

Inside the pod:

export VAULT_ADDR=http://<IP-ADDRESS:PORT>
vault login -method=kubernetes role=flink-role
vault kv get test/flink/client-keystore

If you see the secret → you did it.

Validating the Integration: This test pod simulates a real Flink workload. A successful authentication and secret retrieval confirms network reachability, certificate validation, role binding, and policy enforcement. It’s a powerful diagnostic tool — if it fails at any point, you can pinpoint the exact component.

Creating SecretProviderClass (The Mount Blueprint)

# keystore-spc.yaml
apiVersion: secrets-store.csi.x-k8s.io/v1
kind: SecretProviderClass
metadata:
  name: cmf-client-keystore-spc
  namespace: confluent
spec:
  provider: vault
  parameters:
    vaultAddress: "http://<IP-ADDRESS:PORT"
    roleName: "flink-role"
    vaultKubernetesMountPath: "kubernetes"
    vaultSkipTLSVerify: "true"
    objects: |
      - objectName: "client.keystore.jks"
        secretPath: "test/data/flink/client-keystore"
        secretKey: "keystore"
        encoding: "base64"

Do the same for truststore.

Apply:

kubectl apply -f keystore-spc.yaml
kubectl apply -f truststore-spc.yaml

The SecretProviderClass in Detail: This resource is the blueprint for secret retrieval. Each object entry tells the CSI provider exactly what to fetch, how to name the file in the pod, and how to transform it (base64 decoding). The roleName links back to our Vault role, ensuring only authorized pods can use it. In production, set vaultSkipTLSVerify: “false” and provide a proper CA for Vault’s TLS endpoint.

Configuring Confluent Manager for Flink (CMF)

Add this to your CMF values.yaml:

mountedVolumes:
  volumeMounts:
    - name: client-truststore
      mountPath: /opt/client-truststore
      readOnly: true
    - name: client-keystore
      mountPath: /opt/client-keystore
      readOnly: true
  volumes:
    - name: client-truststore
      csi:
        driver: secrets-store.csi.k8s.io
        readOnly: true
        volumeAttributes:
          secretProviderClass: "cmf-client-truststore-spc"
    - name: client-keystore
      csi:
        driver: secrets-store.csi.k8s.io
        readOnly: true
        volumeAttributes:
          secretProviderClass: "cmf-client-keystore-spc"

Then upgrade:

helm upgrade <cmf-release> <chart> --namespace confluent --values values.yaml

Vault-Flink Integration via CMF: Confluent Manager for Flink (CMF) manages the Flink deployments on Kubernetes. By injecting these volume configurations at the Helm level, every Flink JobManager and TaskManager pod automatically receives the mounted secrets. This centralizes the configuration, making it consistent across the platform.

Configuring Flink Compute Pool

apiVersion: cmf.confluent.io/v1
kind: ComputePool
metadata:
  name: test-pool
spec:
  type: DEDICATED
  clusterSpec:
    flinkVersion: v1_19
    image: confluentinc/cp-flink-sql:1.19-cp1
    flinkConfiguration:
      pipeline.operator-chaining.enabled: "false"
      execution.checkpointing.interval: "10s"
    taskManager:
      resource:
        cpu: 2.0
        memory: "2048m"
    jobManager:
      resource:
        cpu: 2.0
        memory: "2048m"
    podTemplate:
      spec:
        serviceAccountName: confluent-sa  # ← important: must match a SA bound in Vault role
        containers:
        - name: flink-main-container
          volumeMounts:
          - name: client-truststore
            mountPath: /opt/client-truststore
            readOnly: true
          - name: client-keystore
            mountPath: /opt/client-keystore
            readOnly: true
        volumes:
        - name: client-truststore
          csi:
            driver: secrets-store.csi.k8s.io
            readOnly: true
            volumeAttributes:
              secretProviderClass: cmf-client-truststore-spc
        - name: client-keystore
          csi:
            driver: secrets-store.csi.k8s.io
            readOnly: true
            volumeAttributes:
              secretProviderClass: cmf-client-keystore-spc

With this compute pool definition, create the Compute pool with the newly created secret provider class so that Flink Jobmanager and Taskmanager Pods can use the Keystore and Truststore.

Verify Compute Pool Deployment

After creating the compute pool, verify secrets are mounted:

Get Flink task manager pod

kubectl get pods -n confluent | grep taskmanager

Check if secrets are mounted

kubectl exec -it <taskmanager-pod> -n confluent - ls -la /opt/clientkeystore
kubectl exec -it <taskmanager-pod> -n confluent - ls -la /opt/clienttruststore

Expected: You should see client.keystore.jks and client.truststore.jks files.

Understanding mTLS in the Flink-Kafka Context: Mutual TLS (mTLS) goes beyond one-way TLS by requiring both client (Flink) and server (Kafka) to present valid certificates. The keystore contains Flink’s private key and certificate (for client authentication), while the truststore holds the Kafka brokers’ CA certificates (for server validation). This bidirectional verification prevents man-in-the-middle attacks and ensures only trusted clients can produce or consume data. In a Confluent Platform environment, this is often enforced at the broker level via SSL listener configurations.

Why This Matters for Flink Compute Pools: Flink compute pools in Confluent Platform create dynamic TaskManager pods for job execution. Configuring the pod template ensures every new pod inherits the secure volume mounts. The ServiceAccount must be explicitly bound in the Vault role to maintain authorization.

Conclusion

By following this comprehensive guide, you now have a secure integration of HashiCorp Vault with Confluent Platform Flink on Kubernetes. Sensitive Kafka client certificates (keystore and truststore) are stored centrally in Vault, dynamically delivered to Flink JobManager, TaskManager, and compute pool pods via the Secrets Store CSI Driver, and used for mutual TLS (mTLS) authentication to Kafka brokers with zero secrets hardcoded, committed to Git, or stored in etcd.

This architecture represents current best practice for enterprise streaming platforms: secure by design, operationally efficient, and scalable. With proper monitoring, periodic token rotation for the Vault reviewer JWT, and regular certificate updates, your Flink workloads are well protected against credential leakage and unauthorized access.