2.5. Apache Kafka 的源

流程

1. 在 Developer 视角中，进入到 +Add 页面并选择 Event Source。
2. 在 Event Sources 页面中，在 Type 部分选择 Kafka Source。

3. 配置 Kafka Source 设置：

   1. 添加用逗号分开的 Bootstrap 服务器列表。
   2. 添加以逗号分隔的标题列表。
   3. 添加一个 消费者组。
   4. 为您创建的服务帐户选择 Service Account Name。

   5. 在 Target 部分中，选择您的事件 sink。这可以是 Resource 或一个 URI ：

      1. 选择 Resource 来使用频道、代理或服务作为事件源的事件 sink。
      2. 选择 URI 来指定事件路由到的 Uniform Resource Identifier (URI)。
   6. 输入 Kafka 事件源的 名称。
4. 点 Create。

1. 在 Developer 视角中，导航到 Topology。

2. 查看 Kafka 事件源和接收器。

   

流程

1. 要验证 Kafka 事件源是否可以工作，请创建一个 Knative 服务，在服务日志中转储传入的事件：

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

2. 创建 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

   注意

   将此命令中的占位符值替换为源名称、引导服务器和主题的值。

   --servers、--topics 和 --consumergroup 选项指定到 Kafka 集群的连接参数。--consumergroup 选项是可选的。

3. 可选：查看您创建的 KafkaSource CR 的详情：

   $ kn source kafka describe <kafka_source_name>

   输出示例

   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

验证步骤

1. 触发 Kafka 实例将信息发送到主题：

   $ 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

   在提示符后输入信息。这个命令假设：

   • Kafka 集群安装在 kafka 命名空间中。
   • KafkaSource 对象已被配置为使用 my-topic 主题。

2. 通过查看日志来验证消息是否显示：

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

   输出示例

   ☁️  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!

流程

1. 创建 KafkaSource 对象作为 YAML 文件：

   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

   用户组是一组使用相同组群 ID 的用户，并消耗一个标题中的数据。

   2

   主题提供数据存储的目的地。每个主题都被分成一个或多个分区。

   3

   sink 指定事件从源发送到的位置。

   重要

   仅支持 OpenShift Serverless 上的 KafkaSource 对象的 v1beta1 API 版本。不要使用这个 API 的 v1alpha1 版本，因为这个版本现已弃用。

   KafkaSource 对象示例

   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. 应用 KafkaSource YAML 文件：

   $ oc apply -f <filename>

流程

1. 在所选命名空间中创建证书文件作为 secret：

   $ 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

   SASL 类型可以是 PLAIN、SCRAM-SHA-256 或 SCRAM-SHA-512。

2. 创建或修改 Kafka 源，使其包含以下 spec 配置：

   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

   如果您使用公共云 Kafka 服务，则不需要 caCert spec。