Kafka

README.md

kafkacat -b bootstrap.kafka.host:9092 \
  -C -t topic \
  -s avro -r http://schema-registry.host:8081 \
  -f '
  Key (%K bytes): %k
  Value (%S bytes): %s
  Timestamp: %T
  Partition: %p
  Offset: %o
  Headers: %h
  '

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic quickstart-events --from-beginning \
  --property print.key=true \
  --property print.headers=true \
  --property print.timestamp=true

datagen.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka-datagen
  name: kafka-datagen
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-datagen
  template:
    metadata:
      name: databus-kafka
      labels:
        app: kafka-datagen
    spec:
      containers:
      - name: datagen
        image: cnfldemos/cp-server-connect-datagen:0.4.0-6.1.0
        imagePullPolicy: IfNotPresent
        env:
        - name: CONNECT_BOOTSTRAP_SERVERS
          value: ingestion-kafka-bootstrap.kafka.svc:9092
        - name: CONNECT_GROUP_ID
          value: datagen
        - name: CONNECT_REST_ADVERTISED_HOST_NAME
          valueFrom:
            fieldRef:
              fieldPath: status.podIP

        - name: CONNECT_CONFIG_STORAGE_TOPIC
          value: connect.configs
        - name: CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR
          value: "1"
        - name: CONNECT_OFFSET_STORAGE_TOPIC
          value: connect.offsets
        - name: CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR
          value: "1"
        - name: CONNECT_STATUS_STORAGE_TOPIC
          value: connect.status
        - name: CONNECT_STATUS_STORAGE_REPLICATION_FACTOR
          value: "1"

        - name: CONNECT_KEY_CONVERTER
          value: org.apache.kafka.connect.storage.StringConverter
        - name: CONNECT_VALUE_CONVERTER
          value: org.apache.kafka.connect.json.JsonConverter
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-datagen
  labels:
    app: kafka-datagen
spec:
  selector:
    app: kafka-datagen
  ports:
  - name: rest-api
    port: 8083
    protocol: TCP
    targetPort: 8083
---
# datagen-pageviews connector config
# {
#   "name": "datagen-pageviews",
#   "connector.class": "io.confluent.kafka.connect.datagen.DatagenConnector",
#   "quickstart": "pageviews",
#   "kafka.topic": "pageviews",
#   "max.interval": "100"
# }

kafkacat.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafkacat
  namespace: kafka
  labels:
    app.kubernetes.io/name: kafkacat
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: kafkacat
  template:
    metadata:
      creationTimestamp: null
      labels:
        app.kubernetes.io/name: kafkacat
    spec:
      containers:
      - name: kafkacat
        image: edenhill/kafkacat:1.6.0
        command:
        - /bin/sh
        tty: true

kafkactl.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafkactl
  namespace: kafka
  labels:
    app.kubernetes.io/name: kafkactl
spec:
  replicas: 1
  selector:
    matchLabels:
      app.kubernetes.io/name: kafkactl
  template:
    metadata:
      creationTimestamp: null
      labels:
        app.kubernetes.io/name: kafkactl
    spec:
      containers:
      - name: kafkactl
        image: deviceinsight/kafkactl:v1.17.1
        command:
        - /bin/bash
        tty: true
        volumeMounts:
        - name: config
          mountPath: /etc/kafkactl/
      volumes:
      - name: config
        configMap:
          name: kafkactl
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafkactl
  namespace: kafka
  labels:
    app.kubernetes.io/name: kafkactl
data:
  config.yml: |
    current-context: default
    contexts:
      default:
        brokers:
        - kafka-bootstrap.kafka.svc:9092
        avro:
          schemaRegistry: http://schema-registry.kafka.svc:8081

• Kafka Connect Deep Dive – Converters and Serialization Explained