Apache Kafka Source

    1. Install the Kafka controller by entering the following command:

    2. Install the Kafka Source data plane by entering the following command:

      1. kubectl apply -f https://github.com/knative-sandbox/eventing-kafka-broker/releases/download/knative-v1.7.6/eventing-kafka-source.yaml

    3. Verify that kafka-controller and kafka-source-dispatcher are running, by entering the following command:

      1. kubectl get deployments.apps -n knative-eventing

      Example output:

      1. NAME                           READY   UP-TO-DATE   AVAILABLE   AGE
      2. kafka-controller               1/1     1            1           3s
      3. kafka-source-dispatcher        1/1     1            1           4s

    Create a Kafka topic

    Note

    The create a Kafka topic section assumes you’re using Strimzi to operate Apache Kafka, however equivalent operations can be replicated using the Apache Kafka CLI or any other tool.

    If you are using Strimzi:

    1. Create a KafkaTopic YAML file:

      1. apiVersion: kafka.strimzi.io/v1beta2
      2. kind: KafkaTopic
      3. metadata:
      4.   name: knative-demo-topic
      5.   namespace: kafka
      6.   labels:
      7.     strimzi.io/cluster: my-cluster
      8. spec:
      9.   partitions: 3
      10.   replicas: 1
      11.   config:
      12.     retention.ms: 7200000
      13.     segment.bytes: 1073741824

    2. Deploy the KafkaTopic YAML file by running the command:

      1. kubectl apply -f <filename>.yaml

      Where <filename> is the name of your KafkaTopic YAML file.

      Example output:

      1. kafkatopic.kafka.strimzi.io/knative-demo-topic created

    3. Ensure that the KafkaTopic is running by running the command:

      1. kubectl -n kafka get kafkatopics.kafka.strimzi.io

      Example output:

      1. NAME                 CLUSTER      PARTITIONS   REPLICATION FACTOR
      2. knative-demo-topic   my-cluster   3            1

    1. Create the event-display Service as a YAML file:

      1. apiVersion: serving.knative.dev/v1
      2. kind: Service
      3. metadata:
      4.   name: event-display
      5.   namespace: default
      6. spec:
      7.   template:
      8.     spec:
      9.       containers:
      10.         - # This corresponds to
      11.           # https://github.com/knative/eventing/tree/main/cmd/event_display/main.go
      12.           image: gcr.io/knative-releases/knative.dev/eventing/cmd/event_display
      1. kubectl apply -f <filename>.yaml

      Where <filename> is the name of the file you created in the previous step.

      Example output:

    3. Ensure that the Service Pod is running, by running the command:

      1. kubectl get pods

      The Pod name is prefixed with event-display:

      1. NAME                                            READY     STATUS    RESTARTS   AGE
      2. event-display-00001-deployment-5d5df6c7-gv2j4   2/2       Running   0          72s

    1. Modify source/event-source.yaml accordingly with bootstrap servers, topics, and so on:

      1. apiVersion: sources.knative.dev/v1beta1
      2. metadata:
      3.   name: kafka-source
      4. spec:
      5.   consumerGroup: knative-group
      7.   - my-cluster-kafka-bootstrap.kafka:9092 # note the kafka namespace
      8.   topics:
      9.   - knative-demo-topic
      10.   sink:
      11.     ref:
      12.       apiVersion: serving.knative.dev/v1
      13.       kind: Service
      14.       name: event-display

    2. Deploy the event source:

      1. kubectl apply -f event-source.yaml

      Example output:

      1. kafkasource.sources.knative.dev/kafka-source created

    3. Verify that the KafkaSource is ready:

      1. kubectl get kafkasource kafka-source

      Example output:

      1. NAME           TOPICS                   BOOTSTRAPSERVERS                            READY   REASON   AGE
      2. kafka-source   ["knative-demo-topic"]   ["my-cluster-kafka-bootstrap.kafka:9092"]   True             26h

    Verify

    1. Produce a message ({"msg": "This is a test!"}) to the Apache Kafka topic as in the following example:

      1. kubectl -n kafka run kafka-producer -ti --image=strimzi/kafka:0.14.0-kafka-2.3.0 --rm=true --restart=Never -- bin/kafka-console-producer.sh --broker-list my-cluster-kafka-bootstrap:9092 --topic knative-demo-topic

      Tip

      If you don’t see a command prompt, try pressing Enter.

    2. Verify that the Service received the message from the event source:

      1. kubectl logs --selector='serving.knative.dev/service=event-display' -c user-container

      Example output:

      1. ☁️ cloudevents.Event
      2. Validation: valid
      3. Context Attributes,
      4.   specversion: 1.0
      5.   type: dev.knative.kafka.event
      6.   source: /apis/v1/namespaces/default/kafkasources/kafka-source#my-topic
      7.   subject: partition:0#564
      8.   id: partition:0/offset:564
      9.   time: 2020-02-10T18:10:23.861866615Z
      10.   datacontenttype: application/json
      11. Extensions,
      12.   key:
      13. Data,
      14.     {
      15.       "msg": "This is a test!"
      16.     }

    Clean up steps

    1. Example output:

      1. "kafka-source" deleted

    2. Delete the event-display Service:

      1. kubectl delete -f source/event-display.yaml service.serving.knative.dev

      Example output:

      1. "event-display" deleted

    3. Optional: Remove the Apache Kafka Topic

      1. kubectl delete -f kafka-topic.yaml

      Example output:

      1. kafkatopic.kafka.strimzi.io "knative-demo-topic" deleted

    When KafkaSource receives a message from Kafka, it dumps the key in the Event extension called Key and dumps Kafka message headers in the extensions starting with kafkaheader.

    You can specify the key deserializer among four types:

    • string (default) for UTF-8 encoded strings
    • int for 32-bit & 64-bit signed integers
    • float for 32-bit & 64-bit floating points
    • byte-array for a Base64 encoded byte array

    To specify the key deserializer, add the label kafkasources.sources.knative.dev/key-type to the KafkaSource definition, as shown in the following example:

    1. apiVersion: sources.knative.dev/v1beta1
    2. kind: KafkaSource
    3. metadata:
    4. name: kafka-source
    5. labels:
    6.   kafkasources.sources.knative.dev/key-type: int
    7. spec:
    8. bootstrapServers:
    9. - my-cluster-kafka-bootstrap.kafka:9092 # note the kafka namespace
    10. topics:
    12. sink:
    13.   ref:
    14.     apiVersion: serving.knative.dev/v1
    15.     kind: Service
    16.     name: event-display

    Optional: Specify the initial offset

    By default the KafkaSource starts consuming from the latest offset in each partition. If you want to consume from the earliest offset, set the initialOffset field to earliest, for example:

    1. apiVersion: sources.knative.dev/v1beta1
    2. kind: KafkaSource
    3. metadata:
    4.   name: kafka-source
    5. spec:
    6. consumerGroup: knative-group
    7. initialOffset: earliest
    8. bootstrapServers:
    9. - my-cluster-kafka-bootstrap.kafka:9092 # note the kafka namespace
    10. topics:
    11. - knative-demo-topic
    12. sink:
    13.   ref:
    14.     apiVersion: serving.knative.dev/v1
    15.     kind: Service
    16.     name: event-display

    Note

    The valid values for initialOffset are earliest and latest. Any other value results in a validation error. This field is honored only if there are no committed offsets for that consumer group.

    The KafkaSource supports TLS and SASL authentication methods. To enable TLS authentication, you must have the following files:

    • CA Certificate
    • Client Certificate and Key

    KafkaSource expects these files to be in PEM format. If they are in another format, such as JKS, convert them to PEM.

    1. Create the certificate files as secrets in the namespace where KafkaSource is going to be set up, by running the commands:

      1. kubectl create secret generic cacert --from-file=caroot.pem
      1. kubectl create secret tls kafka-secret --cert=certificate.pem --key=key.pem
      1. apiVersion: sources.knative.dev/v1beta1
      2. kind: KafkaSource
      3. metadata:
      4.  name: kafka-source-with-tls
      5. spec:
      6.  net:
      7.    tls:
      8.      enable: true
      9.      cert:
      10.        secretKeyRef:
      11.          key: tls.crt
      12.          name: kafka-secret
      13.      key:
      14.        secretKeyRef:
      15.          key: tls.key
      16.          name: kafka-secret
      17.      caCert:
      18.        secretKeyRef:
      19.          key: caroot.pem
      20.          name: cacert
      21.  consumerGroup: knative-group
      22.  bootstrapServers:
      23.  - my-secure-kafka-bootstrap.kafka:443
      24.  topics:
      25.  - knative-demo-topic
      26.  sink:
      27.    ref:
      28.      apiVersion: serving.knative.dev/v1
      29.      name: event-display