2.4. Source for Apache Kafka

Procedure

1. Navigate to the +Add page and select Event Source.
2. In the Event Sources page, select Kafka Source in the Type section.

3. Configure the Kafka Source settings:

   1. Add a comma-separated list of Bootstrap Servers.
   2. Add a comma-separated list of Topics.
   3. Add a Consumer Group.
   4. Select the Service Account Name for the service account that you created.

   5. In the Target section, select your event sink. This can be either a Resource or a URI:

      1. Select Resource to use a channel, broker, or service as an event sink for the event source.
      2. Select URI to specify a Uniform Resource Identifier (URI) where the events are routed to.
   6. Enter a Name for the Kafka event source.
4. Click Create.

1. Navigate to Topology.

2. View the Kafka event source and sink.

   

Procedure

1. To verify that the Kafka event source is working, create a Knative service that dumps incoming events into the service logs:

   $ kn service create event-display \
       --image quay.io/openshift-knative/showcase

2. Create a KafkaSource CR:

   $ kn source kafka create <kafka_source_name> \
       --servers <cluster_kafka_bootstrap>.kafka.svc:9092 \
       --topics <topic_name> --consumergroup my-consumer-group \
       --sink event-display

   注記

   Replace the placeholder values in this command with values for your source name, bootstrap servers, and topics.

   The --servers, --topics, and --consumergroup options specify the connection parameters to the Kafka cluster. The --consumergroup option is optional.

3. Optional: View details about the KafkaSource CR you created:

   $ kn source kafka describe <kafka_source_name>

   Example output

   Name:              example-kafka-source
   Namespace:         kafka
   Age:               1h
   BootstrapServers:  example-cluster-kafka-bootstrap.kafka.svc:9092
   Topics:            example-topic
   ConsumerGroup:     example-consumer-group
   
   Sink:
     Name:       event-display
     Namespace:  default
     Resource:   Service (serving.knative.dev/v1)
   
   Conditions:
     OK TYPE            AGE REASON
     ++ Ready            1h
     ++ Deployed         1h
     ++ SinkProvided     1h

Verification steps

1. Trigger the Kafka instance to send a message to the topic:

   $ oc -n kafka run kafka-producer \
       -ti --image=quay.io/strimzi/kafka:latest-kafka-2.7.0 --rm=true \
       --restart=Never -- bin/kafka-console-producer.sh \
       --broker-list <cluster_kafka_bootstrap>:9092 --topic my-topic

   Enter the message in the prompt. This command assumes that:

   • The Kafka cluster is installed in the kafka namespace.
   • The KafkaSource object has been configured to use the my-topic topic.

2. Verify that the message arrived by viewing the logs:

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   Example output

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.kafka.event
     source: /apis/v1/namespaces/default/kafkasources/example-kafka-source#example-topic
     subject: partition:46#0
     id: partition:46/offset:0
     time: 2021-03-10T11:21:49.4Z
   Extensions,
     traceparent: 00-161ff3815727d8755848ec01c866d1cd-7ff3916c44334678-00
   Data,
     Hello!

Procedure

1. Create a KafkaSource object as a YAML file:

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: <source_name>
   spec:
     consumerGroup: <group_name> 

   1

   
     bootstrapServers:
     - <list_of_bootstrap_servers>
     topics:
     - <list_of_topics> 

   2

   
     sink:
     - <list_of_sinks> 

   3

   1

   A consumer group is a group of consumers that use the same group ID, and consume data from a topic.

   2

   A topic provides a destination for the storage of data. Each topic is split into one or more partitions.

   3

   A sink specifies where events are sent to from a source.

   重要

   Only the v1beta1 version of the API for KafkaSource objects on OpenShift Serverless is supported. Do not use the v1alpha1 version of this API, as this version is now deprecated.

   Example KafkaSource object

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: kafka-source
   spec:
     consumerGroup: knative-group
     bootstrapServers:
     - my-cluster-kafka-bootstrap.kafka:9092
     topics:
     - knative-demo-topic
     sink:
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: event-display

2. Apply the KafkaSource YAML file:

   $ oc apply -f <filename>

Procedure

1. Create the certificate files as secrets in your chosen namespace:

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-literal=password="SecretPassword" \
     --from-literal=saslType="SCRAM-SHA-512" \ 

   1

   
     --from-literal=user="my-sasl-user"

   1

   The SASL type can be PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512.

2. Create or modify your Kafka source so that it contains the following spec configuration:

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: example-source
   spec:
   ...
     net:
       sasl:
         enable: true
         user:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: user
         password:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: password
         type:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: saslType
       tls:
         enable: true
         caCert: 

   1

   
           secretKeyRef:
             name: <kafka_auth_secret>
             key: ca.crt
   ...

   1

   The caCert spec is not required if you are using a public cloud Kafka service.

Procedure

1. In the KnativeKafka custom resource, enable KEDA scaling:

   Example YAML

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     name: knative-kafka
     namespace: knative-eventing
   spec:
     config:
       kafka-features:
         controller-autoscaler-keda: enabled

2. Apply the KnativeKafka YAML file:

   $ oc apply -f <filename>