Kafka Writer Initial Setup

Mutual TLS (mTLS) requires both the Schema Registry server and the client to present certificates. For steps to generate certificates, see Using Mutual TLS under Setting up Connection to Kafka.

For local (self-managed) Schema Registry deployment, configure the schema-registry.properties:

listeners=https://0.0.0.0:8081
ssl.client.auth=required
ssl.keystore.location=/path/to/server.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/path/to/server.truststore.jks
ssl.truststore.password=changeit

Configure the Schema Registry Connection Profile with mTLS settings:

SASL/PLAIN is a simple authentication mechanism that sends the username and password (base64-encoded). It is the simplest SASL method but requires SSL/TLS for security over the network.

Configure the Kafka Connection Profile with PLAIN authentication:

JAAS Config string format:

org.apache.kafka.common.security.plain.PlainLoginModule required
username="<username>"
password="<password>";

Replace <username> and <password> with valid credentials configured on the Kafka broker.

For Confluent Cloud Kafka, use the API Key as the username and API Secret as the password:

SASL/GSSAPI uses Kerberos for authentication --- an enterprise-grade, ticket-based authentication protocol. It provides strong security without sending passwords over the network.

JAAS Config string format:

com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/etc/security/keytabs/kafka-client.keytab"
principal="kafka-client@EXAMPLE.COM";

Replace the keyTab and principal values with your actual client keytab path and Kerberos principal.

SASL/SCRAM-SHA-256 (Salted Challenge Response Authentication Mechanism) is a secure password-based authentication that does not send passwords over the network. It is more secure than SASL/PLAIN and simpler than Kerberos.

JAAS Config string format:

org.apache.kafka.common.security.scram.ScramLoginModule required
username="<username>"
password="<password>";

Replace <username> and <password> with valid credentials configured on the Kafka broker.

SASL/SCRAM-SHA-512 is identical to SCRAM-SHA-256 but uses the SHA-512 hashing algorithm, providing stronger cryptographic security with longer hash outputs.

JAAS Config string format:

org.apache.kafka.common.security.scram.ScramLoginModule required
username="<username>"
password="<password>";

Replace <username> and <password> with valid credentials configured on the Kafka broker.

Kafka supports combining SASL-based authentication with SSL/TLS encryption to ensure both secure identity verification and encrypted communication. Enable the Use SSL toggle along with any SASL mechanism (PLAIN, SCRAM_SHA_256, SCRAM_SHA_512, or GSSAPI).

Step 1: Create a Certificate Authority (CA)

# Generate CA private key
openssl genrsa -aes256 -out ca-key.pem 4096
# Create self-signed CA certificate
openssl req -x509 -new -key ca-key.pem -sha256 -days 1826 -out ca-cert.pem

Step 2: Create and sign a Server Certificate

openssl genpkey -algorithm RSA -out server.key -aes256
openssl req -new -key server.key -out server.csr -config server.conf
openssl x509 -req -in server.csr -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial \
-out server.crt -days 825 -sha256 -extfile server.ext

Step 3: Create Server Keystore and Truststore

# Keystore
openssl pkcs12 -export -in server.crt -inkey server.key -certfile ca-cert.pem \
-out server.keystore.p12 -name kafka-server
keytool -importkeystore -deststorepass <pass> -destkeypass <pass> \
-destkeystore server.keystore.jks -srckeystore server.keystore.p12 \
-srcstoretype PKCS12 -srcstorepass <p12-pass> -alias kafka-server
# Truststore
keytool -import -alias myCA -file ca-cert.pem -keystore server.truststore.jks

Step 4: Configure Kafka Broker for mTLS (server.properties):

listeners=SSL://localhost:9093
ssl.keystore.location=/path/to/kafka.server.keystore.jks
ssl.keystore.password=serverpass
ssl.key.password=serverkeypass
ssl.truststore.location=/path/to/kafka.server.truststore.jks
ssl.truststore.password=truststorepass
ssl.client.auth=required

Step 5: Create a Client Certificate for Striim and configure the Kafka Connection Profile:

openssl genpkey -algorithm RSA -out client.key -aes256
openssl req -new -key client.key -out client.csr
openssl x509 -req -in client.csr -CA ca-cert.pem -CAkey ca-key.pem \
-CAcreateserial -out client.crt -days 365 -sha256

Rotated certificates are used by adapters when the adapter encounters an exception and retries, or on application restart. To rotate certificates:

Prerequisites:

Steps:

These are the AWS IAM-based equivalents of the Required ACLs for self-managed Kafka. The IAM policies required depend on whether E1P (exactly-once processing) or A1P (at-least-once processing) is used.

IAM Policy for KafkaWriter with E1P:

{
"Version": "<versionNo>",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"kafka:GetBootstrapBrokers",
"kafka:ListTopics",
"kafka:DescribeTopic",
"kafka:DescribeTopicPartitions"
],
"Resource": "*"
},
{
"Sid": "VisualEditor1",
"Effect": "Allow",
"Action": [
"kafka-cluster:DeleteGroup",
"kafka-cluster:WriteDataIdempotently",
"kafka-cluster:DescribeCluster",
"kafka-cluster:ReadData",
"kafka-cluster:DescribeTransactionalId",
"kafka-cluster:AlterTransactionalId",
"kafka-cluster:DescribeTopicDynamicConfiguration",
"kafka-cluster:AlterTopicDynamicConfiguration",
"kafka-cluster:AlterGroup",
"kafka-cluster:AlterClusterDynamicConfiguration",
"kafka-cluster:AlterTopic",
"kafka-cluster:CreateTopic",
"kafka-cluster:DescribeTopic",
"kafka-cluster:AlterCluster",
"kafka-cluster:DescribeGroup",
"kafka-cluster:DescribeClusterDynamicConfiguration",
"kafka-cluster:Connect",
"kafka-cluster:DeleteTopic",
"kafka-cluster:WriteData"
],
"Resource": [
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:transactional-id/*/*/*",
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:cluster/*/*",
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:topic/*/*/*",
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:group/*/*/*"
]
}
]
}

IAM Policy for KafkaWriter with A1P (at-least-once processing):

{
"Version": "<versionNo>",
"Statement": [
{
"Sid": "VisualEditor0",
"Effect": "Allow",
"Action": [
"kafka:GetBootstrapBrokers",
"kafka:ListTopics",
"kafka:DescribeTopic",
"kafka:DescribeTopicPartitions"
],
"Resource": "*"
},
{
"Sid": "VisualEditor1",
"Effect": "Allow",
"Action": [
"kafka-cluster:DeleteGroup",
"kafka-cluster:DescribeCluster",
"kafka-cluster:ReadData",
"kafka-cluster:DescribeTopicDynamicConfiguration",
"kafka-cluster:AlterTopicDynamicConfiguration",
"kafka-cluster:AlterGroup",
"kafka-cluster:AlterClusterDynamicConfiguration",
"kafka-cluster:AlterTopic",
"kafka-cluster:CreateTopic",
"kafka-cluster:DescribeTopic",
"kafka-cluster:AlterCluster",
"kafka-cluster:DescribeGroup",
"kafka-cluster:DescribeClusterDynamicConfiguration",
"kafka-cluster:Connect",
"kafka-cluster:DeleteTopic",
"kafka-cluster:WriteData"
],
"Resource": [
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:cluster/*/*",
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:topic/*/*/*",
"arn:aws:kafka:*:{AWS_ACCOUNT_ID}:group/*/*/*"
]
}
]
}

Connecting with None Authentication Type (plaintext):

Connecting with SCRAM-SHA-512 Authentication:

Connecting with AWS_MSK_IAM Authentication: