Apache Kafka is a distributed event streaming platform that handles high-throughput, fault-tolerant messaging with persistent storage and replay capabilities. Combined with Kafka Connect, it provides a connector framework for reliably moving data between Kafka and external systems. Strimzi is a Kubernetes operator that simplifies deploying, managing, and scaling Kafka clusters through Custom Resource Definitions (CRDs). This combination provides a self-hosted alternative to Amazon Data Firehose, a managed service that automatically loads streaming data into destinations like Amazon S3, Amazon Redshift, Amazon OpenSearch Service, and Splunk with built-in transformations and auto-scaling.

This article explains how to deploy Apache Kafka with Strimzi on Kubernetes as an alternative to Amazon Data Firehose. It covers installing the Strimzi operator, deploying a Kafka cluster in KRaft mode, and configuring Kafka Connect with sink connectors for Object Storage. It also covers stream transformations with Single Message Transforms (SMTs), authentication and encryption, monitoring with Prometheus and Grafana, and migration from Amazon Data Firehose.

Understanding Kafka and Kafka Connect Architecture

Apache Kafka and Kafka Connect map directly to Amazon Data Firehose components, providing equivalent functionality through a self-hosted Kafka and Kafka Connect deployment.

The Kafka with Strimzi architecture consists of:

• Kafka Brokers: Distributed servers that store and replicate topic partitions. Strimzi deploys brokers as StatefulSets with persistent volumes.
• KRaft mode: Kafka's built-in consensus protocol replacing ZooKeeper for metadata management, simplifying operations and reducing component count.
• Kafka Connect: Framework for running source and sink connectors that move data between Kafka and external systems.
• Strimzi Operator: Kubernetes operator that manages Kafka, KafkaConnect, KafkaTopic, and KafkaUser resources through declarative CRDs.
• Schema Registry: Optional component for managing Avro, JSON Schema, and Protobuf schemas when using structured data formats.

Prerequisites

Before you begin, you need to:

• Deploy a Kubernetes cluster with at least 3 worker nodes and enough memory to run Kafka, Kafka Connect, Prometheus, and Grafana concurrently (8 GB RAM per node recommended).
• Install and configure kubectl with cluster access.
• Install Helm 3.
• Provision object storage with an S3-compatible API and create a bucket for the connector to write to.
• Provision a container registry to host the custom Kafka Connect image.

Install the Strimzi Kafka Operator

The Strimzi operator watches for Kafka-related Custom Resources and manages the underlying Kubernetes resources (StatefulSets, Services, ConfigMaps) automatically.

1. Add the Strimzi Helm repository.

   console Copy

   $ helm repo add strimzi https://strimzi.io/charts/
   

   Explain Code

2. Update the Helm repository cache.

   console Copy

   $ helm repo update
   

   Explain Code

3. Create a dedicated namespace for Kafka resources.

   console Copy

   $ kubectl create namespace kafka
   

   Explain Code

4. Install the Strimzi Kafka operator. The --version flag pins the operator to a release whose CRDs still serve the kafka.strimzi.io/v1beta2 API used throughout this article. Starting with Strimzi 1.0.0, Strimzi supports only the v1 CRD API, so manifests that use v1beta2 are not compatible with Strimzi 1.0.0 or newer.

   console Copy

   $ helm install strimzi-kafka-operator strimzi/strimzi-kafka-operator \
       --namespace kafka \
       --version 0.46.0 \
       --set watchAnyNamespace=true
   

   Explain Code

5. Verify that the operator pod is running.

   console Copy

   $ kubectl get pods -n kafka
   

   Explain Code

   The output displays a pod named strimzi-cluster-operator-* with status Running.

6. Verify that the CRDs are installed.

   console Copy

   $ kubectl get crds | grep strimzi
   

   Explain Code

   The output lists CRDs including kafkas.kafka.strimzi.io, kafkanodepools.kafka.strimzi.io, kafkatopics.kafka.strimzi.io, kafkaconnects.kafka.strimzi.io, and kafkausers.kafka.strimzi.io.

Deploy Kafka Cluster

In KRaft mode (Kafka Raft), Kafka manages its own metadata quorum and no longer depends on ZooKeeper. Strimzi models this with KafkaNodePool resources that define groups of nodes acting as brokers, controllers, or both.

1. Create the KafkaNodePool manifest for the controller and broker nodes.

   console Copy

   $ nano kafka-node-pools.yaml
   

   Explain Code

2. Add the following configuration.

   yaml Copy

   ---
   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: controller
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     replicas: 3
     roles:
       - controller
     storage:
       type: persistent-claim
       size: 40Gi
       deleteClaim: false
     resources:
       requests:
         memory: 1Gi
         cpu: 250m
       limits:
         memory: 2Gi
         cpu: 1000m
   ---
   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaNodePool
   metadata:
     name: broker
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     replicas: 3
     roles:
       - broker
     storage:
       type: persistent-claim
       size: 50Gi
       deleteClaim: false
     resources:
       requests:
         memory: 2Gi
         cpu: 500m
       limits:
         memory: 4Gi
         cpu: 2000m
   

   Explain Code

   Save and close the file.

   In this configuration:

   • controller pool: Three nodes that participate in the KRaft quorum for metadata management.
   • broker pool: Three nodes that store topic partitions and serve client traffic.
   • roles: Defines whether each node acts as a controller, broker, or both. Separating roles is recommended for production.
   • storage.type: persistent-claim: Uses Persistent Volume Claims for data durability across pod restarts. The sizes above (40 GB for controllers, 50 GB for brokers) meet common block storage minimum volume sizes. Increase the broker volumes to match your topic retention and partition footprint.

3. Apply the KafkaNodePool manifest.

   console Copy

   $ kubectl apply -f kafka-node-pools.yaml
   

   Explain Code

4. Create the Kafka cluster manifest.

   console Copy

   $ nano kafka-cluster.yaml
   

   Explain Code

5. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: kafka-cluster
     namespace: kafka
     annotations:
       strimzi.io/node-pools: enabled
       strimzi.io/kraft: enabled
   spec:
     kafka:
       version: 3.9.0
       metadataVersion: 3.9-IV0
       listeners:
         - name: plain
           port: 9092
           type: internal
           tls: false
         - name: tls
           port: 9093
           type: internal
           tls: true
       config:
         offsets.topic.replication.factor: 3
         transaction.state.log.replication.factor: 3
         transaction.state.log.min.isr: 2
         default.replication.factor: 3
         min.insync.replicas: 2
     entityOperator:
       topicOperator: {}
       userOperator: {}
   

   Explain Code

   Save and close the file.

   In this configuration:

   • strimzi.io/node-pools: enabled: Enables the KafkaNodePool resources defined in the previous step.
   • strimzi.io/kraft: enabled: Activates KRaft mode, removing the ZooKeeper dependency.
   • metadataVersion: Sets the KRaft metadata version compatible with the configured Kafka version.
   • listeners: Configures both plaintext (port 9092) and Transport Layer Security (TLS) encrypted (port 9093) internal listeners.
   • min.insync.replicas: 2: Requires at least two replicas to acknowledge writes for durability.
   • entityOperator: Deploys the Topic and User operators for managing KafkaTopic and KafkaUser resources.

6. Apply the Kafka cluster manifest.

   console Copy

   $ kubectl apply -f kafka-cluster.yaml
   

   Explain Code

7. Monitor the cluster deployment progress.

   console Copy

   $ kubectl get kafka -n kafka -w
   

   Explain Code

   Wait until the READY column shows True. This may take several minutes as Strimzi provisions the StatefulSets and waits for all pods to become ready.

8. Verify that all Kafka pods are running.

   console Copy

   $ kubectl get pods -n kafka -l strimzi.io/cluster=kafka-cluster
   

   Explain Code

   The output displays three controller pods, three broker pods, and entity operator pods, all with status Running.

Create Ingestion Topics

Kafka topics are the primary mechanism for organizing and storing messages. Sink connectors consume from these topics, so the topics must exist before the connectors are configured.

1. Create the KafkaTopic manifest for the main data ingestion topic.

   console Copy

   $ nano events-topic.yaml
   

   Explain Code

2. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: events
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     partitions: 12
     replicas: 3
     config:
       retention.ms: 604800000
       cleanup.policy: delete
       segment.bytes: 1073741824
       min.insync.replicas: 2
   

   Explain Code

   Save and close the file.

   In this configuration:

   • partitions: 12: Creates 12 partitions for parallel processing. Choose a partition count based on expected throughput and consumer parallelism.
   • replicas: 3: Replicates each partition across all three brokers for fault tolerance.
   • retention.ms: 604800000: Retains messages for 7 days (604800000 milliseconds).
   • cleanup.policy: delete: Removes messages after the retention period expires.
   • segment.bytes: 1073741824: Sets segment file size to 1 GB for efficient disk I/O.

3. Apply the topic manifest.

   console Copy

   $ kubectl apply -f events-topic.yaml
   

   Explain Code

4. Create additional topics for different data streams.

   console Copy

   $ nano additional-topics.yaml
   

   Explain Code

5. Add the following configuration.

   yaml Copy

   ---
   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: logs
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     partitions: 6
     replicas: 3
     config:
       retention.ms: 259200000
       cleanup.policy: delete
   ---
   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: metrics
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     partitions: 6
     replicas: 3
     config:
       retention.ms: 86400000
       cleanup.policy: delete
   

   Explain Code

   Save and close the file.

6. Apply the additional topics.

   console Copy

   $ kubectl apply -f additional-topics.yaml
   

   Explain Code

7. Verify that all topics are created.

   console Copy

   $ kubectl get kafkatopics -n kafka
   

   Explain Code

   The output lists the events, logs, and metrics topics with their partition and replica counts.

Deploy Kafka Connect

Kafka Connect provides a framework for running connectors that stream data between Kafka and external systems. Strimzi builds a custom Connect image with the S3 sink connector plugin and runs it as a KafkaConnect resource.

1. Create a Kubernetes Secret that holds your container registry credentials. Strimzi uses this Secret to push the custom Connect image, and the cluster's worker nodes use it to pull the image. Replace REGISTRY-HOSTNAME, REGISTRY-USERNAME, and REGISTRY-PASSWORD with values from your container registry (for example, the hostname, username, and API key from the Overview tab). Set REGISTRY-HOSTNAME to the registry host only, such as ams.vultrcr.com, without the https:// scheme and without the registry name path. The registry name belongs in the image path in a later step, not in the server value.

   console Copy

   $ kubectl create secret docker-registry registry-credentials \
       --namespace kafka \
       --docker-server=REGISTRY-HOSTNAME \
       --docker-username=REGISTRY-USERNAME \
       --docker-password=REGISTRY-PASSWORD
   

   Explain Code

2. Create the KafkaConnect manifest with the S3 sink connector plugin.

   console Copy

   $ nano kafka-connect.yaml
   

   Explain Code

3. Add the following configuration, replacing REGISTRY-HOSTNAME with your registry hostname and REGISTRY-NAMESPACE with your registry namespace or project path.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: kafka-connect
     namespace: kafka
     annotations:
       strimzi.io/use-connector-resources: "true"
   spec:
     version: 3.9.0
     replicas: 2
     bootstrapServers: kafka-cluster-kafka-bootstrap:9092
     config:
       group.id: connect-cluster
       offset.storage.topic: connect-offsets
       config.storage.topic: connect-configs
       status.storage.topic: connect-status
       offset.storage.replication.factor: 3
       config.storage.replication.factor: 3
       status.storage.replication.factor: 3
       key.converter: org.apache.kafka.connect.storage.StringConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter.schemas.enable: false
     resources:
       requests:
         memory: 1Gi
         cpu: 500m
       limits:
         memory: 2Gi
         cpu: 1000m
     template:
       pod:
         imagePullSecrets:
           - name: registry-credentials
     build:
       output:
         type: docker
         image: REGISTRY-HOSTNAME/REGISTRY-NAMESPACE/kafka-connect-s3:latest
         pushSecret: registry-credentials
       plugins:
         - name: kafka-connect-s3
           artifacts:
             - type: zip
               url: https://hub-downloads.confluent.io/api/plugins/confluentinc/kafka-connect-s3/versions/10.6.7/confluentinc-kafka-connect-s3-10.6.7.zip
   

   Explain Code

   Save and close the file.

   In this configuration:

   • annotations.strimzi.io/use-connector-resources: Enables managing connectors via KafkaConnector CRDs instead of the REST API.
   • replicas: 2: Deploys two Connect workers for high availability and load distribution.
   • bootstrapServers: Points to the Kafka cluster's internal bootstrap service.
   • offset.storage.topic, config.storage.topic, status.storage.topic: Internal topics for storing connector state.
   • template.pod.imagePullSecrets: Lets the worker pods pull the built image from the private container registry.
   • build: Builds a custom Connect image with the S3 sink connector plugin using Strimzi's build capability and pushes it to the configured container registry.

4. Apply the KafkaConnect manifest.

   console Copy

   $ kubectl apply -f kafka-connect.yaml
   

   Explain Code

5. Monitor the build and deployment progress.

   console Copy

   $ kubectl get kafkaconnect -n kafka -w
   

   Explain Code

   Wait until the READY column shows True. The initial build may take several minutes.

6. Verify that the Connect worker pods are running.

   console Copy

   $ kubectl get pods -n kafka -l strimzi.io/cluster=kafka-connect
   

   Explain Code

   The output displays two Connect worker pods with status Running.

7. Check the available connector plugins.

   console Copy

   $ kubectl exec -n kafka kafka-connect-connect-0 -- curl -s localhost:8083/connector-plugins | jq '.[].class'
   

   Explain Code

   The output includes io.confluent.connect.s3.S3SinkConnector confirming the S3 plugin is installed.

Configure S3-Compatible Sink Connector

A sink connector delivers Kafka messages to S3-compatible object storage. Storing credentials in a Kubernetes Secret and non-sensitive settings (bucket name, region, endpoint) in a ConfigMap keeps the connector manifests free of environment-specific values.

1. Create a Kubernetes Secret with object storage credentials. Replace OBJECT-STORAGE-ACCESS-KEY and OBJECT-STORAGE-SECRET-KEY with your S3-compatible access credentials.

   console Copy

   $ kubectl create secret generic s3-credentials \
       --namespace kafka \
       --from-literal=aws.access.key.id=OBJECT-STORAGE-ACCESS-KEY \
       --from-literal=aws.secret.access.key=OBJECT-STORAGE-SECRET-KEY
   

   Explain Code

2. Create a Kubernetes ConfigMap with the bucket and endpoint settings. Replace OBJECT-STORAGE-BUCKET, OBJECT-STORAGE-REGION, and OBJECT-STORAGE-ENDPOINT with values from your object storage provider. Set OBJECT-STORAGE-ENDPOINT to the hostname only (without https://).

   console Copy

   $ kubectl create configmap s3-config \
       --namespace kafka \
       --from-literal=bucket=OBJECT-STORAGE-BUCKET \
       --from-literal=region=OBJECT-STORAGE-REGION \
       --from-literal=endpoint=https://OBJECT-STORAGE-ENDPOINT
   

   Explain Code

3. Update the KafkaConnect resource to expose the Secret and ConfigMap values as environment variables on the Connect worker pods.

   console Copy

   $ nano kafka-connect.yaml
   

   Explain Code

4. Add the following connectContainer block under the existing spec.template block, at the same indentation as the pod block already present there.

   yaml Copy

   connectContainer:
         env:
           - name: AWS_ACCESS_KEY_ID
             valueFrom:
               secretKeyRef:
                 name: s3-credentials
                 key: aws.access.key.id
           - name: AWS_SECRET_ACCESS_KEY
             valueFrom:
               secretKeyRef:
                 name: s3-credentials
                 key: aws.secret.access.key
           - name: S3_BUCKET
             valueFrom:
               configMapKeyRef:
                 name: s3-config
                 key: bucket
           - name: S3_REGION
             valueFrom:
               configMapKeyRef:
                 name: s3-config
                 key: region
           - name: S3_ENDPOINT
             valueFrom:
               configMapKeyRef:
                 name: s3-config
                 key: endpoint
   

   Explain Code

   Save and close the file.

   The Strimzi operator injects each value as an environment variable into the Connect worker container. Connector configurations reference these variables using the ${strimzienv:VAR_NAME} substitution syntax, which Strimzi auto-registers via the EnvVarConfigProvider on every Connect worker.

5. Apply the updated KafkaConnect configuration.

   console Copy

   $ kubectl apply -f kafka-connect.yaml
   

   Explain Code

6. Wait for the Connect cluster to roll the worker pods.

   console Copy

   $ kubectl wait --for=condition=Ready kafkaconnect/kafka-connect -n kafka --timeout=300s
   

   Explain Code

7. Create the S3 sink connector manifest.

   console Copy

   $ nano s3-sink-connector.yaml
   

   Explain Code

8. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: s3-sink-events
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-connect
   spec:
     class: io.confluent.connect.s3.S3SinkConnector
     tasksMax: 3
     config:
       topics: events
       s3.bucket.name: ${strimzienv:S3_BUCKET}
       s3.region: ${strimzienv:S3_REGION}
       store.url: ${strimzienv:S3_ENDPOINT}
       aws.access.key.id: ${strimzienv:AWS_ACCESS_KEY_ID}
       aws.secret.access.key: ${strimzienv:AWS_SECRET_ACCESS_KEY}
       storage.class: io.confluent.connect.s3.storage.S3Storage
       format.class: io.confluent.connect.s3.format.json.JsonFormat
       partitioner.class: io.confluent.connect.storage.partitioner.TimeBasedPartitioner
       path.format: "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH"
       partition.duration.ms: 3600000
       locale: en-US
       timezone: UTC
       flush.size: 1000
       rotate.interval.ms: 600000
       s3.part.size: 5242880
       topics.dir: kafka-events
       behavior.on.null.values: ignore
   

   Explain Code

   Save and close the file.

   In this configuration:

   • tasksMax: 3: Creates three connector tasks for parallel processing across topic partitions.
   • s3.bucket.name, s3.region, store.url: Resolved at runtime from the S3_BUCKET, S3_REGION, and S3_ENDPOINT environment variables sourced from the s3-config ConfigMap.
   • aws.access.key.id, aws.secret.access.key: Resolved at runtime from the AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables sourced from the s3-credentials Secret.
   • format.class: Writes records as JSON files. Alternatives include AvroFormat and ParquetFormat.
   • partitioner.class: TimeBasedPartitioner: Organizes files by time-based directory structure.
   • path.format: Creates directories like year=2024/month=06/day=15/hour=14.
   • flush.size: 1000: Writes a file after receiving 1000 records.
   • rotate.interval.ms: 600000: Forces file rotation every 10 minutes regardless of record count.

9. Apply the S3 sink connector.

   console Copy

   $ kubectl apply -f s3-sink-connector.yaml
   

   Explain Code

10. Verify that the connector is running.

    console Copy

    $ kubectl get kafkaconnector -n kafka
    

    Explain Code

    The output shows the s3-sink-events connector with READY status True.

11. Check the connector status for any errors.

    console Copy

    $ kubectl describe kafkaconnector s3-sink-events -n kafka
    

    Explain Code

Configure Additional Sink Connectors

Configure additional connectors to deliver data to multiple destinations from the same Kafka topics, similar to Amazon Data Firehose's multi-destination capability. Each collapsible below is optional and requires a backing service deployed in the cluster before you configure the connector.

Set Up Stream Transformations

Single Message Transforms (SMTs) modify records as they flow through Kafka Connect, providing functionality similar to Amazon Data Firehose Lambda transforms.

1. Create a connector with SMTs for data enrichment.

   console Copy

   $ nano s3-sink-transformed.yaml
   

   Explain Code

2. Add the following configuration with transformation chains.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: s3-sink-transformed
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-connect
   spec:
     class: io.confluent.connect.s3.S3SinkConnector
     tasksMax: 2
     config:
       topics: events
       s3.bucket.name: ${strimzienv:S3_BUCKET}
       s3.region: ${strimzienv:S3_REGION}
       store.url: ${strimzienv:S3_ENDPOINT}
       aws.access.key.id: ${strimzienv:AWS_ACCESS_KEY_ID}
       aws.secret.access.key: ${strimzienv:AWS_SECRET_ACCESS_KEY}
       storage.class: io.confluent.connect.s3.storage.S3Storage
       format.class: io.confluent.connect.s3.format.json.JsonFormat
       flush.size: 500
       rotate.interval.ms: 300000
       topics.dir: transformed-events
       transforms: addTimestamp,addSource,filterNull,maskSensitive
       transforms.addTimestamp.type: org.apache.kafka.connect.transforms.InsertField$Value
       transforms.addTimestamp.timestamp.field: processed_at
       transforms.addSource.type: org.apache.kafka.connect.transforms.InsertField$Value
       transforms.addSource.static.field: source
       transforms.addSource.static.value: kafka-connect
       transforms.filterNull.type: org.apache.kafka.connect.transforms.Filter
       transforms.filterNull.predicate: isNullValue
       transforms.maskSensitive.type: org.apache.kafka.connect.transforms.MaskField$Value
       transforms.maskSensitive.fields: email,ssn
       transforms.maskSensitive.replacement: "[REDACTED]"
       predicates: isNullValue
       predicates.isNullValue.type: org.apache.kafka.connect.transforms.predicates.RecordIsTombstone
   

   Explain Code

   Save and close the file.

   The configuration applies a chain of four transforms to every record:

   • addTimestamp: Inserts a processed_at field with the time the connector handled the record.
   • addSource: Inserts a static source field set to kafka-connect for downstream lineage.
   • filterNull: Drops records that match the isNullValue predicate, which uses RecordIsTombstone to identify Kafka tombstone records.
   • maskSensitive: Replaces the values of the email and ssn fields with [REDACTED] to prevent personally identifiable information from reaching the sink.

3. Apply the transformed sink connector.

   console Copy

   $ kubectl apply -f s3-sink-transformed.yaml
   

   Explain Code

Configure Buffering and Batching

Tune connector flush intervals and batch sizes to balance latency versus throughput, similar to Amazon Data Firehose buffering configuration.

1. Create a connector with optimized batching for high-throughput scenarios.

   console Copy

   $ nano s3-sink-batched.yaml
   

   Explain Code

2. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnector
   metadata:
     name: s3-sink-batched
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-connect
   spec:
     class: io.confluent.connect.s3.S3SinkConnector
     tasksMax: 6
     config:
       topics: events,logs,metrics
       s3.bucket.name: ${strimzienv:S3_BUCKET}
       s3.region: ${strimzienv:S3_REGION}
       store.url: ${strimzienv:S3_ENDPOINT}
       aws.access.key.id: ${strimzienv:AWS_ACCESS_KEY_ID}
       aws.secret.access.key: ${strimzienv:AWS_SECRET_ACCESS_KEY}
       storage.class: io.confluent.connect.s3.storage.S3Storage
       format.class: io.confluent.connect.s3.format.json.JsonFormat
       flush.size: 10000
       rotate.interval.ms: 900000
       rotate.schedule.interval.ms: 3600000
       s3.part.size: 26214400
       s3.retry.backoff.ms: 200
       s3.retries: 3
       partitioner.class: io.confluent.connect.storage.partitioner.TimeBasedPartitioner
       path.format: "'year'=YYYY/'month'=MM/'day'=dd"
       partition.duration.ms: 86400000
       locale: en-US
       timezone: UTC
       errors.tolerance: all
       errors.deadletterqueue.topic.name: dlq-s3-sink
       errors.deadletterqueue.topic.replication.factor: 3
       errors.log.enable: true
       errors.log.include.messages: true
   

   Explain Code

   Save and close the file.

   In this configuration:

   • flush.size: 10000: Writes a file after receiving 10,000 records.
   • rotate.interval.ms: 900000: Forces file rotation every 15 minutes.
   • rotate.schedule.interval.ms: 3600000: Schedules rotation at hourly boundaries.
   • s3.part.size: 26214400: Uses 25 MB parts for multipart uploads.
   • errors.deadletterqueue.topic.name: Routes failed records to a dead-letter queue topic for later analysis.

3. Apply the batched sink connector.

   console Copy

   $ kubectl apply -f s3-sink-batched.yaml
   

   Explain Code

4. Create the dead-letter queue topic.

   console Copy

   $ nano dlq-topic.yaml
   

   Explain Code

5. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaTopic
   metadata:
     name: dlq-s3-sink
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     partitions: 3
     replicas: 3
     config:
       retention.ms: 2592000000
   

   Explain Code

   Save and close the file.

6. Apply the dead-letter queue topic.

   console Copy

   $ kubectl apply -f dlq-topic.yaml
   

   Explain Code

Set Up Authentication and Encryption

Configure Simple Authentication and Security Layer (SASL) authentication with the Salted Challenge Response Authentication Mechanism (SCRAM) and TLS encryption to secure Kafka communication, replacing Amazon Data Firehose's IAM-based security model. Enable the authorizer on the Kafka cluster first so that KafkaUser Access Control List (ACL) rules are accepted by the operator.

1. Update the Kafka cluster to add a TLS-protected SASL listener and enable simple authorization.

   console Copy

   $ nano kafka-cluster.yaml
   

   Explain Code

2. Add an authorization block under spec.kafka, at the same indentation as the existing listeners and config keys.

   yaml Copy

   authorization:
         type: simple
   

   Explain Code

3. Add the following sasl listener entry to the existing spec.kafka.listeners list, below the tls entry.

   yaml Copy

   - name: sasl
           port: 9094
           type: internal
           tls: true
           authentication:
             type: scram-sha-512
   

   Explain Code

   Save and close the file.

   In this configuration:

   • authorization.type: simple: Enables Kafka's built-in ACL authorizer. Without this block, the User Operator rejects any KafkaUser that declares ACL rules.
   • sasl listener: Adds a TLS-encrypted listener on port 9094 that authenticates clients with SCRAM-SHA-512 credentials.

4. Apply the updated Kafka cluster configuration and wait for the rolling update to finish.

   console Copy

   $ kubectl apply -f kafka-cluster.yaml
   

   Explain Code
   console Copy

   $ kubectl wait --for=condition=Ready kafka/kafka-cluster -n kafka --timeout=600s
   

   Explain Code

5. Create the KafkaUser manifest.

   console Copy

   $ nano kafka-user.yaml
   

   Explain Code

6. Add the following configuration.

   yaml Copy

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: connect-user
     namespace: kafka
     labels:
       strimzi.io/cluster: kafka-cluster
   spec:
     authentication:
       type: scram-sha-512
     authorization:
       type: simple
       acls:
         - resource:
             type: topic
             name: events
             patternType: literal
           operations:
             - Read
             - Write
             - Describe
         - resource:
             type: topic
             name: logs
             patternType: literal
           operations:
             - Read
             - Write
             - Describe
         - resource:
             type: topic
             name: metrics
             patternType: literal
           operations:
             - Read
             - Write
             - Describe
         - resource:
             type: topic
             name: connect-
             patternType: prefix
           operations:
             - Read
             - Write
             - Describe
             - Create
         - resource:
             type: topic
             name: dlq-s3-sink
             patternType: literal
           operations:
             - Read
             - Write
             - Describe
         - resource:
             type: group
             name: connect-cluster
             patternType: literal
           operations:
             - Read
             - Describe
         - resource:
             type: group
             name: connect-
             patternType: prefix
           operations:
             - Read
             - Describe
         - resource:
             type: group
             name: verify-group
             patternType: literal
           operations:
             - Read
             - Describe
   

   Explain Code

   Save and close the file.

   The connect- prefix ACLs on topics and consumer groups let each sink connector create its internal offset topic and join a dedicated consumer group named after the connector (for example, connect-s3-sink-events).

7. Apply the KafkaUser.

   console Copy

   $ kubectl apply -f kafka-user.yaml
   

   Explain Code

8. Verify that the user is Ready and that a credentials Secret was generated.

   console Copy

   $ kubectl get kafkauser connect-user -n kafka
   

   Explain Code

   The output shows True under the READY column once the User Operator has reconciled.

   Note

   To connect a client that runs outside the cluster, retrieve the plaintext SCRAM password from the generated Secret with kubectl get secret connect-user -n kafka -o jsonpath='{.data.password}' | base64 -d. In-cluster components such as KafkaConnect read the password directly from the Secret, so this value is not needed elsewhere in this guide.

9. Update KafkaConnect to use SASL authentication.

   console Copy

   $ nano kafka-connect.yaml
   

   Explain Code

10. Change the existing spec.bootstrapServers value from port 9092 to the SASL listener on port 9094.

    yaml Copy

    bootstrapServers: kafka-cluster-kafka-bootstrap:9094
    

    Explain Code

11. Add the following tls and authentication blocks under spec, at the same indentation as bootstrapServers.

    yaml Copy

    tls:
        trustedCertificates:
          - secretName: kafka-cluster-cluster-ca-cert
            certificate: ca.crt
      authentication:
        type: scram-sha-512
        username: connect-user
        passwordSecret:
          secretName: connect-user
          password: password
    

    Explain Code

    Save and close the file.

12. Apply the updated KafkaConnect configuration and wait for the worker pods to roll onto the new SASL/TLS configuration.

    console Copy

    $ kubectl apply -f kafka-connect.yaml
    

    Explain Code
    console Copy

    $ kubectl wait --for=condition=Ready kafkaconnect/kafka-connect -n kafka --timeout=300s
    

    Explain Code
    Note

    The first worker pod to restart under the new configuration can briefly crash loop with NoSuchFileException: /tmp/kafka/cluster.truststore.p12, since the mounted CA certificate Secret volume is not always fully populated before the container's truststore-generation step runs on the first attempt. The pod recovers automatically on a subsequent restart within about a minute. If the kubectl wait command above times out, check kubectl get pods -n kafka -l strimzi.io/cluster=kafka-connect before troubleshooting further; the pods are often already Running by then.

Configure Monitoring

Prometheus scrapes the Kafka metrics that Strimzi's JMX exporters expose, and Grafana visualizes them, replacing CloudWatch monitoring.

1. Enable Strimzi metrics exporters in the Kafka cluster.

   console Copy

   $ nano kafka-cluster.yaml
   

   Explain Code

2. Add the metricsConfig section under spec.kafka:

   yaml Copy

   spec:
     kafka:
       # ... existing configuration ...
       metricsConfig:
         type: jmxPrometheusExporter
         valueFrom:
           configMapKeyRef:
             name: kafka-metrics
             key: kafka-metrics-config.yml
   

   Explain Code

   Save and close the file.

3. Create the metrics configuration ConfigMap.

   console Copy

   $ nano kafka-metrics-configmap.yaml
   

   Explain Code

4. Add the following configuration.

   yaml Copy

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: kafka-metrics
     namespace: kafka
   data:
     kafka-metrics-config.yml: |
       lowercaseOutputName: true
       rules:
       - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), topic=(.+), partition=(.*)><>Value
         name: kafka_server_$1_$2
         type: GAUGE
         labels:
           clientId: "$3"
           topic: "$4"
           partition: "$5"
       - pattern: kafka.server<type=(.+), name=(.+), clientId=(.+), brokerHost=(.+), brokerPort=(.+)><>Value
         name: kafka_server_$1_$2
         type: GAUGE
         labels:
           clientId: "$3"
           broker: "$4:$5"
       - pattern: kafka.server<type=(.+), cipher=(.+), protocol=(.+), listener=(.+), networkProcessor=(.+)><>connections
         name: kafka_server_$1_connections_tls_info
         type: GAUGE
         labels:
           cipher: "$2"
           protocol: "$3"
           listener: "$4"
           networkProcessor: "$5"
       - pattern: kafka.server<type=(.+), clientSoftwareName=(.+), clientSoftwareVersion=(.+), listener=(.+), networkProcessor=(.+)><>connections
         name: kafka_server_$1_connections_software
         type: GAUGE
         labels:
           clientSoftwareName: "$2"
           clientSoftwareVersion: "$3"
           listener: "$4"
           networkProcessor: "$5"
       - pattern: "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+-total):"
         name: kafka_server_$1_$4
         type: COUNTER
         labels:
           listener: "$2"
           networkProcessor: "$3"
       - pattern: "kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+):"
         name: kafka_server_$1_$4
         type: GAUGE
         labels:
           listener: "$2"
           networkProcessor: "$3"
       - pattern: kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+-total)
         name: kafka_server_$1_$4
         type: COUNTER
         labels:
           listener: "$2"
           networkProcessor: "$3"
       - pattern: kafka.server<type=(.+), listener=(.+), networkProcessor=(.+)><>(.+)
         name: kafka_server_$1_$4
         type: GAUGE
         labels:
           listener: "$2"
           networkProcessor: "$3"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>MeanRate
         name: kafka_$1_$2_$3_percent
         type: GAUGE
       - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*><>Value
         name: kafka_$1_$2_$3_percent
         type: GAUGE
       - pattern: kafka.(\w+)<type=(.+), name=(.+)Percent\w*, (.+)=(.+)><>Value
         name: kafka_$1_$2_$3_percent
         type: GAUGE
         labels:
           "$4": "$5"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+), (.+)=(.+)><>Count
         name: kafka_$1_$2_$3_total
         type: COUNTER
         labels:
           "$4": "$5"
           "$6": "$7"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*, (.+)=(.+)><>Count
         name: kafka_$1_$2_$3_total
         type: COUNTER
         labels:
           "$4": "$5"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)PerSec\w*><>Count
         name: kafka_$1_$2_$3_total
         type: COUNTER
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Value
         name: kafka_$1_$2_$3
         type: GAUGE
         labels:
           "$4": "$5"
           "$6": "$7"
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Value
         name: kafka_$1_$2_$3
         type: GAUGE
         labels:
           "$4": "$5"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Value
         name: kafka_$1_$2_$3
         type: GAUGE
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+), (.+)=(.+)><>Count
         name: kafka_$1_$2_$3_count
         type: COUNTER
         labels:
           "$4": "$5"
           "$6": "$7"
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*), (.+)=(.+)><>(\d+)thPercentile
         name: kafka_$1_$2_$3
         type: GAUGE
         labels:
           "$4": "$5"
           "$6": "$7"
           quantile: "0.$8"
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.+)><>Count
         name: kafka_$1_$2_$3_count
         type: COUNTER
         labels:
           "$4": "$5"
       - pattern: kafka.(\w+)<type=(.+), name=(.+), (.+)=(.*)><>(\d+)thPercentile
         name: kafka_$1_$2_$3
         type: GAUGE
         labels:
           "$4": "$5"
           quantile: "0.$6"
       - pattern: kafka.(\w+)<type=(.+), name=(.+)><>Count
         name: kafka_$1_$2_$3_count
         type: COUNTER
       - pattern: kafka.(\w+)<type=(.+), name=(.+)><>(\d+)thPercentile
         name: kafka_$1_$2_$3
         type: GAUGE
         labels:
           quantile: "0.$4"
       - pattern: "kafka.server<type=raft-metrics><>(.+-total|.+-max):"
         name: kafka_server_raftmetrics_$1
         type: COUNTER
       - pattern: "kafka.server<type=raft-metrics><>(current-state): (.+)"
         name: kafka_server_raftmetrics_$1
         value: 1
         type: UNTYPED
         labels:
           $1: "$2"
       - pattern: "kafka.server<type=raft-metrics><>(.+):"
         name: kafka_server_raftmetrics_$1
         type: GAUGE
       - pattern: "kafka.server<type=raft-channel-metrics><>(.+-total|.+-max):"
         name: kafka_server_raftchannelmetrics_$1
         type: COUNTER
       - pattern: "kafka.server<type=raft-channel-metrics><>(.+):"
         name: kafka_server_raftchannelmetrics_$1
         type: GAUGE
       - pattern: "kafka.server<type=broker-metadata-metrics><>(.+):"
         name: kafka_server_brokermetadatametrics_$1
         type: GAUGE
   

   Explain Code

   Save and close the file.

5. Apply the metrics ConfigMap.

   console Copy

   $ kubectl apply -f kafka-metrics-configmap.yaml
   

   Explain Code

6. Apply the updated Kafka cluster configuration.

   console Copy

   $ kubectl apply -f kafka-cluster.yaml
   

   Explain Code

   Strimzi now exposes Prometheus metrics on port 9404 of each broker and controller pod. The next steps deploy the Prometheus Operator and create a PodMonitor that scrapes those endpoints directly, since the pods expose this port but the Kafka Services do not. The PodMonitor resource depends on the monitoring.coreos.com/v1 Custom Resource Definition that the Prometheus Operator installs.

7. Add the Prometheus Community Helm repository.

   console Copy

   $ helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
   

   Explain Code

8. Update the Helm repository cache.

   console Copy

   $ helm repo update
   

   Explain Code

9. Install the kube-prometheus-stack.

   console Copy

   $ helm install prometheus prometheus-community/kube-prometheus-stack \
       --namespace monitoring \
       --create-namespace \
       --set grafana.adminPassword=admin \
       --set prometheus.prometheusSpec.podMonitorSelectorNilUsesHelmValues=false \
       --set grafana.resources.requests.memory=256Mi \
       --set grafana.resources.limits.memory=512Mi
   

   Explain Code

   The podMonitorSelectorNilUsesHelmValues=false setting allows Prometheus to discover PodMonitor resources outside the Helm release, including the Kafka PodMonitor you create in the next steps.

10. Wait for the Prometheus and Grafana pods to become ready.

    console Copy

    $ kubectl wait --for=condition=Ready pod \
        -l app.kubernetes.io/name=grafana -n monitoring --timeout=300s
    

    Explain Code
    console Copy

    $ kubectl wait --for=condition=Ready pod \
        -l app.kubernetes.io/name=prometheus -n monitoring --timeout=300s
    

    Explain Code

    If Grafana remains in Error or CrashLoopBackOff, inspect the pod with kubectl describe pod -l app.kubernetes.io/name=grafana -n monitoring and increase memory limits or add a node with more capacity.

11. Create the PodMonitor manifest that tells Prometheus to scrape Kafka metrics. A PodMonitor is required instead of a ServiceMonitor because Strimzi exposes the tcp-prometheus metrics port directly on the broker and controller pods; the headless *-kafka-brokers Service it generates does not include that port.

    console Copy

    $ nano kafka-podmonitor.yaml
    

    Explain Code

12. Add the following configuration.

    yaml Copy

    apiVersion: monitoring.coreos.com/v1
    kind: PodMonitor
    metadata:
      name: kafka-resources-metrics
      namespace: kafka
      labels:
        app: strimzi
    spec:
      selector:
        matchExpressions:
          - key: "strimzi.io/kind"
            operator: In
            values: ["Kafka", "KafkaConnect", "KafkaMirrorMaker2"]
      namespaceSelector:
        matchNames:
          - kafka
      podMetricsEndpoints:
        - path: /metrics
          port: tcp-prometheus
          relabelings:
            - separator: ;
              regex: __meta_kubernetes_pod_label_(strimzi_io_.+)
              replacement: $1
              action: labelmap
            - sourceLabels: [__meta_kubernetes_namespace]
              targetLabel: namespace
              action: replace
            - sourceLabels: [__meta_kubernetes_pod_name]
              targetLabel: kubernetes_pod_name
              action: replace
            - sourceLabels: [__meta_kubernetes_pod_node_name]
              targetLabel: node_name
              action: replace
            - sourceLabels: [__meta_kubernetes_pod_host_ip]
              targetLabel: node_ip
              action: replace
    

    Explain Code

    Save and close the file.

    The matchExpressions selector matches pods from any Kafka, KafkaConnect, or KafkaMirrorMaker2 resource in the namespace, so this single PodMonitor also picks up the KafkaConnect worker pods once you enable metricsConfig on the KafkaConnect resource the same way you did for the Kafka cluster.

13. Apply the PodMonitor.

    console Copy

    $ kubectl apply -f kafka-podmonitor.yaml
    

    Explain Code

14. Verify that Prometheus discovered the Kafka scrape targets.

    console Copy

    $ kubectl get podmonitor kafka-resources-metrics -n kafka
    

    Explain Code

    Port-forward the Prometheus UI and open Status, followed by Targets, to confirm kafka-resources-metrics endpoints appear as UP.

    console Copy

    $ kubectl port-forward svc/prometheus-kube-prometheus-prometheus -n monitoring 9090:9090
    

    Explain Code

    Open http://localhost:9090/targets in a web browser.

15. Access Grafana using port-forwarding.

    console Copy

    $ kubectl port-forward svc/prometheus-grafana -n monitoring 3000:80
    

    Explain Code

16. Open http://localhost:3000 in a web browser and log in with username admin and password admin.

17. Open Connections, then Data sources in the left-side menu, and confirm that a Prometheus data source is listed. The kube-prometheus-stack chart provisions it automatically through a Grafana sidecar, but this step can fail silently. If a Prometheus data source already appears, continue to the next step.

18. If no Prometheus data source is present, click Add new data source and select Prometheus. Set the Prometheus server URL to the in-cluster Prometheus service, then scroll to the bottom and click Save & test. Verify that Grafana reports the data source is working.

    http://prometheus-kube-prometheus-prometheus.monitoring.svc.cluster.local:9090

19. Download a Strimzi dashboard definition. This example downloads the Kafka broker dashboard. The grafana-dashboards folder also includes strimzi-kafka-connect.json for Kafka Connect metrics.

    console Copy

    $ wget https://raw.githubusercontent.com/strimzi/strimzi-kafka-operator/main/examples/metrics/grafana-dashboards/strimzi-kafka.json
    

    Explain Code

20. In Grafana, click the + icon in the top-right corner and select Import dashboard. Under Upload dashboard JSON file, drag in or browse to the downloaded strimzi-kafka.json, or paste its contents into the Import via dashboard JSON model field. Click Load.

21. In the import screen, set the Prometheus data source selector to the Prometheus data source you verified earlier, then click Import.

    The dashboard populates once Prometheus has scraped the kafka-resources-metrics targets.

Verify the Deployment

Validate the complete deployment by testing end-to-end data flow from a producer through Kafka to the S3 sink. The Kafka cluster now requires SASL/SCRAM authentication over TLS, so the test client uses the connect-user credentials and the cluster CA certificate.

1. Verify that the connector and all its tasks are running.

   console Copy

   $ kubectl get kafkaconnector s3-sink-events -n kafka
   

   Explain Code

   The READY column shows True. If any task reports FAILED, inspect the trace with kubectl exec -n kafka kafka-connect-connect-0 -- curl -s localhost:8083/connectors/s3-sink-events/status.

2. Create a client pod manifest that mounts the cluster CA certificate and exposes the SCRAM password as an environment variable.

   console Copy

   $ nano kafka-client.yaml
   

   Explain Code

3. Add the following configuration.

   yaml Copy

   apiVersion: v1
   kind: Pod
   metadata:
     name: kafka-client
     namespace: kafka
   spec:
     restartPolicy: Never
     containers:
       - name: client
         image: quay.io/strimzi/kafka:0.46.0-kafka-3.9.0
         command: ["sleep", "infinity"]
         env:
           - name: PASSWORD
             valueFrom:
               secretKeyRef:
                 name: connect-user
                 key: password
         volumeMounts:
           - name: cluster-ca
             mountPath: /etc/kafka-ca
             readOnly: true
     volumes:
       - name: cluster-ca
         secret:
           secretName: kafka-cluster-cluster-ca-cert
   

   Explain Code

   Save and close the file.

4. Apply the client pod manifest and wait for it to become ready.

   console Copy

   $ kubectl apply -f kafka-client.yaml
   

   Explain Code
   console Copy

   $ kubectl wait --for=condition=Ready pod/kafka-client -n kafka --timeout=120s
   

   Explain Code

5. Generate the client properties file inside the pod with the SASL credentials and the cluster truststore path. Use /bin/bash -lc so the pod's PASSWORD environment variable expands inside the container rather than in your local shell.

   console Copy

   $ kubectl exec -n kafka kafka-client -- /bin/bash -lc 'cat > /tmp/client.properties <<EOF
   security.protocol=SASL_SSL
   sasl.mechanism=SCRAM-SHA-512
   sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="connect-user" password="${PASSWORD}";
   ssl.truststore.type=PEM
   ssl.truststore.location=/etc/kafka-ca/ca.crt
   EOF'
   

   Explain Code

6. Produce a batch of 1500 test messages with the same key so they land on one topic partition and cross the connector's flush.size threshold. Files materialize in the bucket within seconds rather than waiting on rotate.interval.ms.

   console Copy

   $ kubectl exec -n kafka kafka-client -- /bin/bash -lc 'awk "BEGIN{for(i=1;i<=1500;i++) printf(\"verify:{\\\"event_id\\\":\\\"evt-%04d\\\",\\\"type\\\":\\\"test\\\",\\\"i\\\":%d}\n\",i,i)}" | bin/kafka-console-producer.sh --bootstrap-server kafka-cluster-kafka-bootstrap:9094 --topic events --property parse.key=true --property key.separator=: --producer.config /tmp/client.properties'
   

   Explain Code

7. Wait about 30 seconds for the connector to commit the batch.

8. List the objects in the bucket. Replace OBJECT-STORAGE-ACCESS-KEY, OBJECT-STORAGE-SECRET-KEY, OBJECT-STORAGE-REGION, OBJECT-STORAGE-ENDPOINT, and OBJECT-STORAGE-BUCKET with the values from your s3-credentials Secret and s3-config ConfigMap.

   console Copy

   $ kubectl run awscli --rm -it --restart=Never --image=amazon/aws-cli \
       --env AWS_ACCESS_KEY_ID=OBJECT-STORAGE-ACCESS-KEY \
       --env AWS_SECRET_ACCESS_KEY=OBJECT-STORAGE-SECRET-KEY \
       --env AWS_DEFAULT_REGION=OBJECT-STORAGE-REGION \
       -- --endpoint-url https://OBJECT-STORAGE-ENDPOINT s3 ls s3://OBJECT-STORAGE-BUCKET/kafka-events/ --recursive
   

   Explain Code

   This command runs the AWS CLI in a temporary pod, so no local installation is required. The output displays JSON files organized by the time-based partitioning structure, for example kafka-events/events/year=2026/month=06/day=25/hour=17/events+0+0000000000.json.

   Warning

   This command passes the Object Storage secret key on the command line, so it is recorded in the pod's container logs. The --rm flag deletes the pod immediately after the command exits, but rotate the key afterward if the cluster is not disposable, or browse the bucket in the Vultr Console instead to avoid exposing the credentials.

9. Verify the data format by downloading and inspecting one of the listed files. Replace the placeholders as before, and replace the object key with one from the previous listing. This command carries the same credential-exposure risk described in the previous step.

   console Copy

   $ kubectl run awscli --rm -it --restart=Never --image=amazon/aws-cli \
       --env AWS_ACCESS_KEY_ID=OBJECT-STORAGE-ACCESS-KEY \
       --env AWS_SECRET_ACCESS_KEY=OBJECT-STORAGE-SECRET-KEY \
       --env AWS_DEFAULT_REGION=OBJECT-STORAGE-REGION \
       -- --endpoint-url https://OBJECT-STORAGE-ENDPOINT s3 cp s3://OBJECT-STORAGE-BUCKET/kafka-events/events/year=2026/month=06/day=25/hour=17/events+0+0000000000.json -
   

   Explain Code

   Each line is a JSON record matching the messages produced earlier.

10. Verify topic delivery by consuming directly from Kafka. Pass --group verify-group so the consumer joins the group authorized in the connect-user ACLs.

    console Copy

    $ kubectl exec -it -n kafka kafka-client -- bin/kafka-console-consumer.sh \
        --bootstrap-server kafka-cluster-kafka-bootstrap:9094 \
        --topic events \
        --from-beginning \
        --max-messages 10 \
        --group verify-group \
        --consumer.config /tmp/client.properties
    

    Explain Code

    The output displays the messages produced earlier.

11. Delete the client pod once verification completes.

    console Copy

    $ kubectl delete pod kafka-client -n kafka
    

    Explain Code

Migrate from Amazon Data Firehose to Kafka Connect

Migrating from Amazon Data Firehose to Kafka Connect involves mapping Firehose streams to topic and connector pairs, updating producer applications, and converting Lambda-based transforms to Kafka Connect SMTs or Kafka Streams. The service was formerly known as Amazon Kinesis Data Firehose; API resources still use names such as DeliveryStreamName. The guidance below assumes you completed Deploy Kafka Cluster, Deploy Kafka Connect, and Set Up Authentication and Encryption earlier in this article.

Firehose Stream Migration

An Amazon Data Firehose stream bundles a source, a destination, and buffering rules into a single managed resource. Kafka Connect splits these responsibilities across two Strimzi resources: a KafkaTopic that holds the records and a KafkaConnector that delivers them to the destination.

• Firehose stream name: Choose a Kafka topic name and a matching KafkaConnector name for each Firehose stream. In the Firehose API, this is DeliveryStreamName.
• Source: Amazon Data Firehose supports Direct PUT, Amazon Kinesis Data Streams, and Amazon MSK as sources. A Kinesis data stream or MSK cluster backing a Firehose stream maps to a Kafka topic that producers write to directly after migration. Direct PUT streams map to a topic that producer applications write to using the Kafka client instead of the Firehose PutRecord API.
• Destination: Common Firehose destination types and their Kafka Connect equivalents:
  • Amazon S3: Confluent S3 Sink Connector
  • Amazon OpenSearch Service / OpenSearch Serverless: Elasticsearch Sink Connector (requires Elasticsearch 7.x compatibility; see the optional collapsible in Configure Additional Sink Connectors)
  • Amazon Redshift: S3 Sink Connector staging to Object Storage, then a separate load step. Firehose delivers to S3 first and runs a Redshift COPY command; it does not write directly to Redshift. Model this with an S3 sink connector and your existing warehouse load tooling rather than the JDBC sink alone.
  • Splunk: Splunk Sink Connector.
  • HTTP endpoints and third-party observability targets (Datadog, New Relic, and similar): HTTP Sink Connector or a custom sink connector
• Buffering: Firehose BufferingHints uses IntervalInSeconds (0–900, default 300) and SizeInMBs (1–128 MiB, default 5). If you set one, you must set the other. Map IntervalInSeconds to the connector's rotate.interval.ms (multiply seconds by 1000). SizeInMBs is a byte-size threshold with no direct Kafka Connect equivalent; approximate it by tuning flush.size based on your average record size. Use s3.part.size for multipart upload chunk sizing, not as a direct Firehose buffer equivalent.

Producer Migration

Producer changes depend on how data enters Firehose.

If producers use Firehose Direct PUT (the most common pattern), replace the Firehose API with Apache Kafka client libraries:

• PutRecord / PutRecordBatch: producer.send() (Amazon Data Firehose uses PutRecordBatch, not Amazon Kinesis Data Streams PutRecords)
• DeliveryStreamName: Kafka topic name
• Firehose records are opaque data blobs with no partition key; use a Kafka message key in your producer if you need partition affinity

If producers write to Amazon Kinesis Data Streams that Amazon Data Firehose reads as a source, replace the Kinesis Data Streams API and optionally the Kinesis Producer Library (KPL):

• PutRecord / PutRecords: producer.send()
• StreamName: Kafka topic name
• PartitionKey: Kafka message key, which Kafka hashes to assign records to partitions
• KPL aggregation and batching: Kafka producer batch.size, linger.ms, and compression.type

In both cases, update producers to use SASL/SCRAM credentials and the cluster TLS certificate configured in Set Up Authentication and Encryption. Refer to the Kafka producer configuration documentation for tuning options.

Transform Migration

Firehose invokes Lambda transforms synchronously on each buffered batch before delivery. Lambda buffering uses separate hints (BufferSizeInMBs up to 3 MB and IntervalInSeconds up to 900 in the processing configuration), distinct from destination BufferingHints.

Kafka Connect offers two equivalent mechanisms with different operational characteristics:

• Single Message Transforms (SMTs): Suitable for stateless, per-record operations such as field rename, field insertion, format conversion, and filtering with predicates. SMTs run inside the Kafka Connect worker process with no additional infrastructure. Refer to the Confluent SMT reference.
• Kafka Streams: Suitable for stateful operations, windowed aggregations, joins, and complex business logic that exceeds the SMT scope. Kafka Streams applications run as long-lived processes that consume from a source topic, transform records, and produce to a destination topic that the sink connector then reads. Refer to the Kafka Streams documentation.

Unlike Lambda transforms, which scale to zero between invocations, Kafka Streams applications run continuously and consume dedicated compute resources. Account for this cost-model difference when planning capacity.

Destination Configuration Migration

Firehose destination settings such as S3 prefix patterns, file formats, and compression map to Kafka Connect sink connector configuration keys.

• S3 prefix: Firehose Prefix expressions such as !{timestamp:yyyy/MM/dd/HH} map to the connector's topics.dir, partitioner.class, and path.format settings. Refer to the Firehose custom prefix documentation and the S3 Sink Connector partitioning documentation.
• Error output prefix: Firehose ErrorOutputPrefix (for example errors/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd}) maps to the connector's dead-letter queue topic configured with errors.deadletterqueue.topic.name. Route that topic to Object Storage with a second sink connector if you require failed records on disk, similar to Firehose writing failed records under ErrorOutputPrefix.
• Compression: Firehose CompressionFormat values (GZIP, ZIP, Snappy, HADOOP_SNAPPY, or UNCOMPRESSED) map to the connector's format class and s3.compression.type setting.
• File format: Firehose record format conversion from JSON to Parquet or ORC maps to the connector's format.class setting combined with a schema-aware converter such as Avro or Protobuf and an AWS Glue schema equivalent in your Kafka ecosystem.
• Buffering: Firehose BufferingHints (IntervalInSeconds, SizeInMBs) map to rotate.interval.ms, flush.size, and s3.part.size as described under Configure Buffering and Batching.

Data Storage Considerations

For destinations with existing data that needs to be preserved:

• Bucket reuse: Point the S3 sink connector at the same bucket Firehose was writing to, and reserve a new top-level prefix for Kafka Connect output to avoid clashes with existing objects.
• Historical backfill: Replay historical records by reading existing S3 objects with a custom producer script that publishes them back to the source topic, allowing the sink connector to write them under the new prefix structure.
• File format compatibility: Confirm downstream consumers such as data lakes, query engines, and ETL jobs can read the connector's output format. Kafka Connect writes JSON, Avro, or Parquet depending on the configured format.class.
• Partitioning alignment: Match the connector's path.format to the existing Firehose Prefix scheme so that downstream queries and data catalogs continue to discover new partitions automatically.

Migration Considerations

Review the following operational differences before cutting production traffic over to the new deployment.

• Auto-scaling: Firehose auto-scales transparently. Kafka Connect requires explicit scaling by adjusting replicas and tasksMax on the KafkaConnect resource, or by configuring a Horizontal Pod Autoscaler driven by consumer lag metrics.
• Lambda transforms versus Kafka Streams: Firehose Lambda transforms are serverless and billed per invocation. Kafka Streams applications run as always-on workloads, which changes the cost model and the failure surface.
• Monitoring transition: Replace Firehose CloudWatch metrics with the Prometheus metrics and Grafana dashboards configured under Configure Monitoring. Common mappings include IncomingRecords (ingestion volume), DeliveryToS3.Success (successful S3 deliveries), and BackupToS3.Records (records written to the S3 backup bucket when backup is enabled).
• Delivery semantics: Both Firehose and Kafka Connect provide at-least-once delivery by default. Firehose retries failed deliveries and may produce duplicates on producer retries. Implement idempotent writes or deduplication in downstream systems when exactly-once is required.
• Cost model change: Firehose charges per GB of ingested and delivered data. Kafka on Kubernetes costs are infrastructure-based, covering compute, storage, and network. Calculate total cost of ownership including operational overhead before cutover.
• Identity translation: Replace AWS IAM roles and policies with Kubernetes RBAC and KafkaUser ACLs. Map each Firehose IAM role to a KafkaUser with the minimum set of topic and group ACLs required for its workload.
• Throughput limits: Firehose enforces per-stream quotas such as RecordsPerSecondLimit and BytesPerSecondLimit. Kafka throughput is bounded by your cluster's broker count, partition count, and disk and network capacity.

Conclusion

You have deployed Apache Kafka with Strimzi on Kubernetes as a self-hosted alternative to Amazon Data Firehose. The deployment includes a Kafka cluster, Kafka Connect with S3-compatible sink connectors, stream transformations using SMTs, SASL/SCRAM authentication with TLS encryption, and Prometheus monitoring. For additional configuration options, connector plugins, and advanced features, visit the official Strimzi documentation and Kafka Connect documentation.