9.1 – Security in Kafka Connect

Overview

This module is the authoritative guide to securing Kafka Connect clusters. By the end, you will understand:

• TLS/SSL and SASL authentication for Connect-to-Kafka communication.
• Securing connector credentials and sensitive configurations.
• Secret handling with externalized config providers.
• RBAC (Role-Based Access Control) patterns for connectors.
• REST API security.

All Connect security concepts live here. Other modules will reference this when discussing production deployments or compliance requirements, but will not re-explain these concepts.

Security Architecture

Threat Model

┌─────────────────────────────────────────────────┐
│           Kafka Connect Security Layers         │
│                                                 │
│  1. Connect ↔ Kafka: TLS + SASL                │
│  2. Connect ↔ External System: Credentials      │
│  3. REST API: Authentication + Authorization    │
│  4. Config Storage: Encrypted Secrets           │
│  5. Worker-to-Worker: TLS (distributed mode)    │
│                                                 │
└─────────────────────────────────────────────────┘

Key Security Goals

1. Confidentiality: Encrypt data in transit (TLS) and at rest (external system credentials).
2. Authentication: Verify identity of workers, connectors, and clients.
3. Authorization: Control which connectors can access which Kafka topics.
4. Auditability: Log all connector operations for compliance.

Securing Connect-to-Kafka Communication

1. TLS/SSL Encryption

Purpose: Encrypt data in transit between Connect workers and Kafka brokers.

Worker Configuration (connect-distributed.properties):

# Kafka broker connection
bootstrap.servers=kafka1:9093,kafka2:9093,kafka3:9093

# Enable TLS
security.protocol=SSL

# TLS truststore (CA certificate)
ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
ssl.truststore.password=truststore-password

# TLS keystore (client certificate, optional for mutual TLS)
ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
ssl.keystore.password=keystore-password
ssl.key.password=key-password

Verification:

openssl s_client -connect kafka1:9093 -showcerts

2. SASL Authentication

Purpose: Authenticate Connect workers to Kafka brokers.

SASL/PLAIN (Username/Password)

Worker Configuration:

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect-worker" \
  password="worker-secret";

SASL/SCRAM (More Secure)

Create User (on Kafka broker):

kafka-configs --bootstrap-server localhost:9092 \
  --alter \
  --add-config 'SCRAM-SHA-512=[password=worker-secret]' \
  --entity-type users \
  --entity-name connect-worker

Worker Configuration:

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
  username="connect-worker" \
  password="worker-secret";

SASL/GSSAPI (Kerberos)

Worker Configuration:

security.protocol=SASL_SSL
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka
sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
  useKeyTab=true \
  storeKey=true \
  keyTab="/etc/security/keytabs/connect.keytab" \
  principal="connect/hostname@REALM.COM";

3. Consumer and Producer Security Overrides

Purpose: Apply security configs to internal Kafka consumer/producer used by Connect.

Worker Configuration:

# Consumer security (for sink connectors)
consumer.security.protocol=SASL_SSL
consumer.sasl.mechanism=SCRAM-SHA-512
consumer.sasl.jaas.config=...

# Producer security (for source connectors)
producer.security.protocol=SASL_SSL
producer.sasl.mechanism=SCRAM-SHA-512
producer.sasl.jaas.config=...

Securing Connector Credentials

The Problem

Connectors often require credentials for external systems (e.g., database passwords, API keys). Hardcoding them in connector configs is insecure:

{
  "connection.url": "jdbc:postgresql://db.example.com/mydb",
  "connection.password": "super-secret-password"  ❌ BAD
}

Risks:

• Credentials stored in plaintext in Kafka internal topics (connect-configs).
• Visible in REST API responses (GET /connectors/{name}/config).
• Accessible to anyone with access to the Connect cluster.

Externalized Secrets (Config Providers)

What Are Config Providers?

Config Providers are plugins that fetch secrets from external systems (e.g., Vault, AWS Secrets Manager, Kubernetes Secrets) and inject them into connector configs at runtime.

Syntax:

${provider:path/to/secret:key}

Example: File Config Provider

Create Secrets File (/etc/kafka-connect/secrets.properties):

db.password=super-secret-password
api.key=my-api-key-12345

Restrict Permissions:

chmod 600 /etc/kafka-connect/secrets.properties
chown kafka-connect:kafka-connect /etc/kafka-connect/secrets.properties

Worker Configuration:

# Enable FileConfigProvider
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

Connector Configuration:

{
  "connection.url": "jdbc:postgresql://db.example.com/mydb",
  "connection.user": "myuser",
  "connection.password": "${file:/etc/kafka-connect/secrets.properties:db.password}"
}

Result: At runtime, ${file:...} is replaced with the actual password from the file.

Example: HashiCorp Vault Config Provider

Dependencies (Maven):

<dependency>
  <groupId>com.github.jcustenborder.kafka.config</groupId>
  <artifactId>kafka-config-provider-vault</artifactId>
  <version>0.1.2</version>
</dependency>

Worker Configuration:

config.providers=vault
config.providers.vault.class=com.github.jcustenborder.kafka.config.VaultConfigProvider
config.providers.vault.param.vault.address=https://vault.example.com
config.providers.vault.param.vault.token=s.XXXXXXXXXXXXXXXXXXXX

Connector Configuration:

{
  "connection.password": "${vault:secret/database:password}"
}

Benefit: Centralized secret management with Vault's audit logs, rotation, and access control.

Example: AWS Secrets Manager Config Provider

Dependencies (Maven):

<dependency>
  <groupId>com.github.jcustenborder.kafka.config</groupId>
  <artifactId>kafka-config-provider-aws</artifactId>
  <version>0.1.2</version>
</dependency>

Worker Configuration:

config.providers=aws
config.providers.aws.class=com.github.jcustenborder.kafka.config.AwsSecretsManagerConfigProvider
config.providers.aws.param.aws.region=us-east-1

Store Secret in AWS Secrets Manager:

aws secretsmanager create-secret \
  --name kafka-connect/db-password \
  --secret-string "super-secret-password"

Connector Configuration:

{
  "connection.password": "${aws:kafka-connect/db-password}"
}

Example: Kubernetes Secrets Config Provider

Create Kubernetes Secret:

kubectl create secret generic connect-secrets \
  --from-literal=db-password=super-secret-password

Mount Secret as Volume (in Connect deployment):

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-connect
spec:
  template:
    spec:
      containers:
      - name: connect
        volumeMounts:
        - name: secrets
          mountPath: /etc/kafka-connect/secrets
          readOnly: true
      volumes:
      - name: secrets
        secret:
          secretName: connect-secrets

Worker Configuration:

config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProvider

Connector Configuration:

{
  "connection.password": "${file:/etc/kafka-connect/secrets/db-password}"
}

RBAC (Role-Based Access Control)

Kafka ACLs for Connectors

Purpose: Control which connectors can produce/consume from which Kafka topics.

Example: Grant Source Connector Permissions

Source Connector (produces to orders topic):

# Allow WRITE to topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation WRITE \
  --topic orders

# Allow DESCRIBE on topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation DESCRIBE \
  --topic orders

# Allow READ/WRITE to offset storage topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation ALL \
  --topic connect-offsets

# Allow READ/WRITE to config storage topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation ALL \
  --topic connect-configs

# Allow READ/WRITE to status storage topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation ALL \
  --topic connect-status

Example: Grant Sink Connector Permissions

Sink Connector (consumes from logs topic):

# Allow READ from topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation READ \
  --topic logs

# Allow DESCRIBE on topic
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation DESCRIBE \
  --topic logs

# Allow READ on consumer group
kafka-acls --bootstrap-server localhost:9092 \
  --add \
  --allow-principal User:connect-worker \
  --operation READ \
  --group connect-my-sink-connector

# Allow READ/WRITE to internal topics (same as source)
# (connect-offsets, connect-configs, connect-status)

REST API Security

1. Basic Authentication

Purpose: Require username/password for REST API access.

Worker Configuration:

rest.extension.classes=org.apache.kafka.connect.rest.basic.auth.extension.BasicAuthSecurityRestExtension

rest.extension.basic.auth.credentials.file=/etc/kafka-connect/rest-credentials.properties
rest.extension.basic.auth.credentials.reload=true

Credentials File (rest-credentials.properties):

admin=admin-password
operator=operator-password

API Call:

curl -u admin:admin-password http://localhost:8083/connectors

2. TLS for REST API

Purpose: Encrypt REST API traffic (HTTPS).

Worker Configuration:

listeners=https://0.0.0.0:8083

rest.advertised.listener=https
rest.advertised.host.name=connect.example.com
rest.advertised.port=8083

# TLS keystore
listeners.https.ssl.keystore.location=/var/private/ssl/rest-keystore.jks
listeners.https.ssl.keystore.password=keystore-password
listeners.https.ssl.key.password=key-password

# TLS truststore (for mutual TLS, optional)
listeners.https.ssl.truststore.location=/var/private/ssl/rest-truststore.jks
listeners.https.ssl.truststore.password=truststore-password
listeners.https.ssl.client.auth=required

API Call:

curl --cacert ca-cert.pem https://connect.example.com:8083/connectors

3. OAuth 2.0 / JWT (Advanced)

Purpose: Integrate with enterprise identity providers (e.g., Okta, Auth0, Keycloak).

Implementation: Use a custom REST extension or API gateway (e.g., Kong, Apigee) to validate JWT tokens.

Audit Logging

Enable Audit Logs

Worker Configuration (log4j.properties):

# Log all connector operations
log4j.logger.org.apache.kafka.connect.runtime.WorkerConnector=INFO, auditLog
log4j.additivity.org.apache.kafka.connect.runtime.WorkerConnector=false

# Audit log appender
log4j.appender.auditLog=org.apache.log4j.RollingFileAppender
log4j.appender.auditLog.File=/var/log/kafka-connect/audit.log
log4j.appender.auditLog.MaxFileSize=100MB
log4j.appender.auditLog.MaxBackupIndex=10
log4j.appender.auditLog.layout=org.apache.log4j.PatternLayout
log4j.appender.auditLog.layout.ConversionPattern=[%d] %p %m (%c)%n

Logged Events:

• Connector creation/deletion.
• Connector configuration changes.
• Task start/stop.
• Connector failures.

Security Best Practices

1. Never Hardcode Secrets

❌ Bad:

{"connection.password": "super-secret-password"}

✅ Good:

{"connection.password": "${vault:secret/db:password}"}

2. Use Mutual TLS for Production

Why: Ensures both client (Connect) and server (Kafka) authenticate each other.

Configuration:

ssl.keystore.location=/var/private/ssl/kafka.client.keystore.jks
ssl.keystore.password=...
ssl.key.password=...

3. Rotate Credentials Regularly

Automated Rotation:

• Use Vault's dynamic secrets (credentials expire and regenerate).
• Use AWS Secrets Manager's automatic rotation.

Manual Rotation:

1. Update secret in config provider (Vault, AWS Secrets Manager).
2. Restart connectors (Connect auto-reloads secrets).

4. Restrict REST API Access

Options:

• Basic Auth (development/testing).
• TLS + mutual TLS (production).
• API Gateway + OAuth 2.0 (enterprise).

5. Audit Everything

What to Audit:

• Connector creation/deletion.
• Configuration changes.
• Failed authentication attempts (REST API).
• Task failures.

Tools: Splunk, ELK Stack, Datadog Logs.

RBAC via Confluent Platform CLI

For users of Confluent Platform, you can manage RBAC (Role-Based Access Control) centrally using the Confluent CLI instead of managing individual Kafka ACLs.

Prerequisites

• Confluent Platform 5.4.0+ with RBAC enabled
• Confluent CLI installed
• Admin access to Metadata Service (MDS)

Step 1: Create a Role Binding for a Connector

# Grant connector the 'DeveloperWrite' role on topic 'orders'
confluent iam rolebinding create \
  --principal User:connect-source-user \
  --role DeveloperWrite \
  --resource Topic:orders \
  --kafka-cluster-id <cluster-id>

# Grant connector the 'DeveloperRead' role on topic 'users' (for sink)
confluent iam rolebinding create \
  --principal User:connect-sink-user \
  --role DeveloperRead \
  --resource Topic:users \
  --kafka-cluster-id <cluster-id>

# Grant connector access to consumer group
confluent iam rolebinding create \
  --principal User:connect-sink-user \
  --role DeveloperRead \
  --resource Group:connect-my-sink \
  --kafka-cluster-id <cluster-id>

Step 2: Configure Connect Worker to Use RBAC

Worker Configuration (connect-distributed.properties):

# SASL/OAUTHBEARER for RBAC
security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  username="connect-worker" \
  password="..." \
  metadataServerUrls="https://mds:8090";

# SSL settings
ssl.truststore.location=/var/private/ssl/kafka.client.truststore.jks
ssl.truststore.password=...

# Enable RBAC for REST API
rest.extension.classes=io.confluent.connect.security.ConnectSecurityExtension

Step 3: Verify Role Bindings

# List all role bindings for a principal
confluent iam rolebinding list \
  --principal User:connect-source-user \
  --kafka-cluster-id <cluster-id>

Expected Output:

Principal         | Role            | Resource Type | Resource Name | Pattern Type
------------------|-----------------|---------------|---------------|-------------
User:connect-...  | DeveloperWrite  | Topic         | orders        | LITERAL

Benefits of RBAC over ACLs:

1. Centralized Management: Single point of control via MDS.
2. Role Reusability: Define roles once, apply to multiple principals.
3. Audit Trail: All RBAC changes are logged in MDS.
4. Integration: Works with LDAP, Active Directory, SSO.

Network Security: Firewall Rules

Kafka Connect requires specific network ports to be open for communication with Kafka, external systems, and monitoring tools. Here's a reference table for firewall configuration.

Required Ports for Kafka Connect

Example: AWS Security Group (Inbound)

Example: AWS Security Group (Outbound)

Best Practices:

1. Least Privilege: Only open ports required for your connectors.
2. Internal-Only REST API: Do not expose port 8083 to the internet.
3. Use Security Groups/VPCs: Segment Connect workers from public networks.
4. TLS Everywhere: Use TLS for Kafka (9093), Schema Registry (8081), external systems.

Security Troubleshooting Guide

Use this table to diagnose common security issues in Kafka Connect.

General Debugging Steps:

1. Enable DEBUG logging for security components:

   log4j.logger.org.apache.kafka.common.security=DEBUG
   

2. Test Kafka connectivity separately with kafka-console-producer or kafka-console-consumer using the same credentials as Connect.
3. Use openssl s_client to test TLS connections:

   openssl s_client -connect kafka-broker:9093 -showcerts
   

4. Verify ACLs with kafka-acls:

   kafka-acls --authorizer-properties bootstrap-server=localhost:9092 --list
   

Security Checklist

Pre-Production

• Enable TLS for Connect-to-Kafka communication.
• Enable SASL authentication (SCRAM or Kerberos).
• Externalize all connector credentials (Vault, AWS Secrets Manager).
• Restrict REST API access (Basic Auth or TLS).
• Configure Kafka ACLs for each connector.
• Enable audit logging.

Production

• Use mutual TLS for Connect-to-Kafka.
• Rotate credentials every 90 days.
• Monitor for failed authentication attempts.
• Encrypt internal Kafka topics (connect-configs, connect-offsets, connect-status) at rest.
• Use network segmentation (firewalls, VPCs).

Key Takeaways

1. Connect-to-Kafka: Enable TLS + SASL (SCRAM, Kerberos) for authentication.
2. Connector Credentials: Use externalized secrets (Vault, AWS Secrets Manager, Kubernetes Secrets).
3. RBAC: Use Kafka ACLs to control connector topic access.
4. REST API: Secure with Basic Auth, TLS, or OAuth 2.0.
5. Audit Logging: Log all connector operations for compliance.
6. Best Practices: Never hardcode secrets, use mutual TLS, rotate credentials regularly.

What's Next?

• Module 8.1 covers monitoring (use security metrics to detect intrusions).
• Module 3.2 explains distributed mode (secure internal topic replication).
• Module 7.4 discusses packaging connectors (include security in deployment docs).

Further Reading

• Kafka Connect Security (Confluent)
• Config Providers (KIP-297)
• Kafka Security Documentation

End of Module 9.1