Deploying Apache Kafka as a Self-Hosted Amazon Data Firehose Alternative

Apache Kafka + Kafka Connect, managed by the Strimzi Kubernetes operator, gives you a self-hosted alternative to Amazon Data Firehose , streaming ingestion with pluggable sink connectors instead of a fully-managed delivery pipeline. This guide deploys a KRaft-mode Kafka cluster via Strimzi, wires up Kafka Connect with an S3-compatible sink (plus optional Elasticsearch and JDBC sinks), applies Single Message Transforms, configures buffering, secures everything with SASL/SCRAM + TLS, adds Prometheus/Grafana monitoring, and covers migrating off Firehose.

Concept Mapping

Amazon Data Firehose	Kafka / Kafka Connect	Description
Kinesis Data Streams (source)	Kafka Topics	Persistent, partitioned streams
Firehose stream	Kafka Connect Sink Connector	Consumes from topics, writes to external systems
S3 destination	S3 Sink Connector	Writes JSON/Avro/Parquet to S3-compatible storage
OpenSearch destination	Elasticsearch Sink Connector	Indexes records for search/analytics
Lambda transform	Kafka Streams / SMTs	In-flight record transformation
Buffering (SizeInMBs/IntervalInSeconds)	Flush configuration	Controls batched writes by count/size/time
Record format conversion	Format converters	JSON/Avro/etc. via Connect converters
CloudWatch Metrics	Prometheus + Grafana	Via Strimzi's built-in exporters

Architecture: Kafka brokers (StatefulSets storing/replicating partitions) + KRaft mode (built-in consensus, no ZooKeeper) + Kafka Connect (source/sink connector framework) + the Strimzi Operator (manages everything via CRDs) + an optional Schema Registry for Avro/JSON Schema/Protobuf.

Prerequisites: a Kubernetes cluster (3+ worker nodes, 8GB RAM/node recommended), kubectl + Helm 3, an S3-compatible object storage bucket, and a container registry to host the custom Kafka Connect image.

---

Install the Strimzi Operator

$ helm repo add strimzi https://strimzi.io/charts/
$ helm repo update
$ kubectl create namespace kafka
$ helm install strimzi-kafka-operator strimzi/strimzi-kafka-operator \
    --namespace kafka \
    --version 0.46.0 \
    --set watchAnyNamespace=true

Pinned to 0.46.0 — its CRDs still serve kafka.strimzi.io/v1beta2, which this guide uses throughout. Strimzi 1.0.0+ only supports the v1 CRD API.

$ kubectl get pods -n kafka
$ kubectl get crds | grep strimzi

Should list kafkas.kafka.strimzi.io, kafkanodepools.kafka.strimzi.io, kafkatopics.kafka.strimzi.io, kafkaconnects.kafka.strimzi.io, kafkausers.kafka.strimzi.io.

---

Deploy the Kafka Cluster (KRaft Mode)

KRaft replaces ZooKeeper with Kafka's own consensus protocol. Strimzi models node roles via KafkaNodePool.

$ nano kafka-node-pools.yaml

---
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

3 controller nodes (KRaft quorum) + 3 broker nodes (data + client traffic) — separating roles is recommended for production. Increase broker volume size to match your retention/partition footprint.

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

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: {}

$ kubectl apply -f kafka-cluster.yaml
$ kubectl get kafka -n kafka -w

Wait for READY: True (can take several minutes).

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

3 controller + 3 broker + entity operator pods, all Running.

---

Create Topics

$ nano events-topic.yaml

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

12 partitions for parallelism, 3x replication, 7-day retention, 1GB segments.

$ kubectl apply -f events-topic.yaml
$ nano additional-topics.yaml

---
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

$ kubectl apply -f additional-topics.yaml
$ kubectl get kafkatopics -n kafka

---

Deploy Kafka Connect with an S3 Sink

1. Registry credentials for Strimzi to build and push a custom Connect image:

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

Set REGISTRY-HOSTNAME to just the host (e.g. registry.example.com), no scheme, no path.

2. KafkaConnect manifest — replace REGISTRY-HOSTNAME/REGISTRY-NAMESPACE:

$ nano kafka-connect.yaml

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

replicas: 2 for HA; build compiles a custom image with the S3 plugin baked in and pushes it to your registry.

$ kubectl apply -f kafka-connect.yaml
$ kubectl get kafkaconnect -n kafka -w

Initial build can take a few minutes.

$ kubectl get pods -n kafka -l strimzi.io/cluster=kafka-connect
$ kubectl exec -n kafka kafka-connect-connect-0 -- curl -s localhost:8083/connector-plugins | jq '.[].class'

Confirms io.confluent.connect.s3.S3SinkConnector is available.

---

Configure the S3-Compatible Sink

1. Credentials + config (kept separate so manifests stay environment-agnostic):

$ 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
$ 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

2. Expose them as env vars on the Connect workers — add under spec.template in kafka-connect.yaml:

$ nano kafka-connect.yaml

    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

Connector configs reference these via ${strimzienv:VAR_NAME} — Strimzi auto-registers an EnvVarConfigProvider for this.

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

3. Create the connector:

$ nano s3-sink-connector.yaml

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

format.class can swap to Avro/Parquet; path.format gives time-partitioned directories; flush.size/rotate.interval.ms control when files actually get written.

$ kubectl apply -f s3-sink-connector.yaml
$ kubectl get kafkaconnector -n kafka
$ kubectl describe kafkaconnector s3-sink-events -n kafka

---

Optional: Additional Sink Connectors

Elasticsearch — deploy a single-node ES 7.x instance in-namespace, add the kafka-connect-elasticsearch plugin to the build.plugins list, configure an ES_CONNECTION_URL env var, then create a KafkaConnector with class io.confluent.connect.elasticsearch.ElasticsearchSinkConnector pointed at the logs topic. (OpenSearch isn't used here — the Confluent ES connector does a strict version check that rejects it.)

JDBC/PostgreSQL — deploy PostgreSQL in-namespace, add kafka-connect-jdbc plus the Postgres JDBC driver jar to build.plugins, store connection details in a Secret, then create a KafkaConnector with class io.confluent.connect.jdbc.JdbcSinkConnector targeting the metrics topic. This one needs value.converter.schemas.enable: "true" and pk.mode: kafka since the JDBC sink requires a schema-wrapped JSON envelope ({"schema": {...}, "payload": {...}}).

Both follow the same pattern as the S3 sink: build plugin → deploy backing service → Secret/ConfigMap for connection details → env vars on Connect → KafkaConnector resource.

---

Stream Transformations (SMTs)

Single Message Transforms give you Lambda-transform-equivalent behavior inline in Connect:

    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

Chain: insert a processed_at timestamp → insert a static source field → drop tombstone records → redact email/ssn. Add this block to an S3 sink connector config and apply it as its own KafkaConnector resource.

---

Buffering and Batching

For high-throughput scenarios, tune flush thresholds and add dead-letter handling:

    tasksMax: 6
    topics: events,logs,metrics
    flush.size: 10000
    rotate.interval.ms: 900000
    rotate.schedule.interval.ms: 3600000
    s3.part.size: 26214400
    s3.retry.backoff.ms: 200
    s3.retries: 3
    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

Create the DLQ topic (dlq-s3-sink, partitions: 3, replicas: 3, 30-day retention) so failed records land somewhere inspectable instead of silently dropping.

---

Authentication and Encryption

Enable Kafka's ACL authorizer and a SASL/SCRAM+TLS listener — the self-hosted analogue of Firehose's IAM security model.

1. Add to kafka-cluster.yaml — an authorization block under spec.kafka:

    authorization:
      type: simple

And a sasl listener under spec.kafka.listeners:

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

$ kubectl apply -f kafka-cluster.yaml
$ kubectl wait --for=condition=Ready kafka/kafka-cluster -n kafka --timeout=600s

2. Create a KafkaUser with scoped ACLs for the connect topics/groups (events, logs, metrics, connect-* prefix, dlq-s3-sink, connect-cluster group, verify-group):

$ nano kafka-user.yaml

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]

The connect- prefix ACLs let each sink connector create its own internal offset topic and consumer group.

$ kubectl apply -f kafka-user.yaml
$ kubectl get kafkauser connect-user -n kafka

For clients outside the cluster, grab the SCRAM password: kubectl get secret connect-user -n kafka -o jsonpath='{.data.password}' | base64 -d. In-cluster components (like Connect) read it straight from the Secret.

3. Point Kafka Connect at the SASL listener — change bootstrapServers to port 9094 and add TLS/auth blocks:

  bootstrapServers: kafka-cluster-kafka-bootstrap:9094
  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

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

The first worker pod restarting under the new config can briefly crash-loop with a NoSuchFileException on the truststore — it recovers within about a minute as the CA cert volume finishes mounting. Check kubectl get pods before troubleshooting further if wait times out.

---

Monitoring with Prometheus + Grafana

1. Enable JMX metrics export on the Kafka cluster via spec.kafka.metricsConfig pointing at a ConfigMap, then create that ConfigMap with the JMX-to-Prometheus mapping rules — Strimzi ships a reference rules file covering broker, controller, and RAFT metrics; copy its data.kafka-metrics-config.yml content into your ConfigMap named kafka-metrics.

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

Strimzi now exposes Prometheus metrics on port 9404 per broker/controller pod.

2. Install the Prometheus stack:

$ helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
$ helm repo update
$ 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
$ kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=grafana -n monitoring --timeout=300s
$ kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=prometheus -n monitoring --timeout=300s

podMonitorSelectorNilUsesHelmValues=false lets Prometheus pick up PodMonitors outside the Helm release.

3. Create a PodMonitor (needed instead of a ServiceMonitor since Strimzi exposes the metrics port directly on pods, not through the headless Service):

$ nano kafka-podmonitor.yaml

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

$ kubectl apply -f kafka-podmonitor.yaml
$ kubectl get podmonitor kafka-resources-metrics -n kafka
$ kubectl port-forward svc/prometheus-kube-prometheus-prometheus -n monitoring 9090:9090

Check http://localhost:9090/targets for kafka-resources-metrics showing UP.

4. Grafana:

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

Open http://localhost:3000 (admin/admin). Confirm the Prometheus data source exists (auto-provisioned by the chart), or add it manually pointing at http://prometheus-kube-prometheus-prometheus.monitoring.svc.cluster.local:9090.

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

+ → Import dashboard, upload the JSON, select your Prometheus source, Import. (A strimzi-kafka-connect.json dashboard is also available in the same repo folder.)

---

Verify End-to-End

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

Create a test client with the CA cert mounted and SCRAM password injected:

$ nano kafka-client.yaml

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

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

Generate client properties and produce test messages:

$ 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'

$ 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'

1500 same-keyed messages land on one partition and cross the flush.size threshold, so the file materializes quickly rather than waiting for rotate.interval.ms.

List and inspect the delivered file (~30s after producing):

$ 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

This passes the secret key on the command line, which lands in the pod's logs — --rm deletes the pod right after, but rotate the key afterward on a non-disposable cluster, or just browse the bucket through your storage provider's web console instead.

$ 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 -

Consume directly from Kafka to double-check:

$ 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

$ kubectl delete pod kafka-client -n kafka

---

Migrating from Amazon Data Firehose

Firehose bundles source + destination + buffering into one resource; Kafka Connect splits this into a KafkaTopic + KafkaConnector pair.

Sources: a Kinesis stream or MSK cluster feeding Firehose maps to a topic producers write to directly; Direct PUT streams map to a topic written via Kafka client libraries instead of the Firehose PutRecord API.

Destinations:

• S3 → Confluent S3 Sink Connector
• OpenSearch/OpenSearch Serverless → Elasticsearch Sink Connector (7.x compatibility caveat noted above)
• Redshift → S3 Sink Connector staging + your existing warehouse COPY tooling (Firehose doesn't write directly to Redshift either)
• Splunk → Splunk Sink Connector
• HTTP/observability targets (Datadog, New Relic) → HTTP Sink Connector or custom sink

Buffering: Firehose IntervalInSeconds (0–900) × 1000 → rotate.interval.ms; SizeInMBs has no exact equivalent — approximate via flush.size based on average record size; s3.part.size is for multipart chunking, not a buffer analog.

Producers: replace Firehose's PutRecord/PutRecordBatch (or Kinesis PutRecord/PutRecords if that's your actual source) with producer.send(); DeliveryStreamName/StreamName → Kafka topic name; PartitionKey → Kafka message key. Update producers to use the SASL/SCRAM credentials and TLS cert from the auth section above.

Transforms: Firehose Lambda transforms → SMTs (stateless, per-record) or Kafka Streams (stateful, windowed, joins). Unlike Lambda, Kafka Streams apps run continuously and consume dedicated resources — factor that into capacity planning.

Destination config: Firehose Prefix expressions → topics.dir/partitioner.class/path.format; ErrorOutputPrefix → a DLQ topic (optionally sinked to storage separately); CompressionFormat → the connector's s3.compression.type; Firehose's JSON→Parquet/ORC conversion → format.class + a schema-aware converter.

Data migration: reuse the existing bucket but with a new top-level prefix to avoid clashes; backfill historical S3 objects by replaying them through a custom producer back into the source topic; verify downstream consumers can read your connector's output format; match path.format to the old Prefix scheme so catalogs keep discovering partitions.

Watch for:

• Scaling: Firehose auto-scales; Kafka Connect needs explicit replicas/tasksMax tuning or an HPA on consumer lag
• Cost model: Lambda transforms bill per-invocation and scale to zero; Kafka Streams is always-on
• Monitoring: CloudWatch IncomingRecords/DeliveryToS3.Success/BackupToS3.Records → the Prometheus/Grafana setup above
• Delivery semantics: both are at-least-once by default — dedupe downstream if you need exactly-once
• Cost: Firehose is per-GB ingested/delivered; self-hosted is infra-based — compute your real TCO
• Identity: AWS IAM roles → Kubernetes RBAC + KafkaUser ACLs, one per workload
• Throughput: Firehose enforces per-stream quotas; Kafka's ceiling is your broker/partition/disk/network capacity

---

Next Steps

Kafka + Kafka Connect is running with KRaft, an S3 sink, SASL/TLS security, and full Prometheus/Grafana observability. From here:

• Add the Elasticsearch or JDBC sink connectors if you need multi-destination delivery like Firehose supports
• Wire an HPA on Kafka Connect driven by consumer lag for auto-scaling
• Explore Kafka Streams for stateful transforms beyond what SMTs can express

For the full guide, visit the original article on Vultr Docs.