16-Kafka Broker

关于KafkaBroker

• Knative Kafka Broker是Apache Kafka针对Knative Broker API的原生实现
• 它支持Kafka的任意版本，并且在与Broker/Trigger模型集成时有着更好的表现
• 几个关键特性
  • Control plane High Availability
  • Horizontally scalable data plane
  • Extensively configurable
  • Ordered delivery of events based on CloudEvents partitioning extension
  • Support any Kafka version

部署KafkaBroker

• 部署Kafka Broker的Controller，包括kafka-controller和kafka-webhook。

  ~$ kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-controller.yaml
  

• 部署Kafka Broker的数据平面，包括kafka-broker-dispatcher和kafka-broker-receiver.

  ~$ kubectl apply -f https://github.com/knative-extensions/eventing-kafka-broker/releases/download/knative-v1.12.1/eventing-kafka-broker.yaml
  

• 确认运行状态

  ~$ kubectl get pods -n knative-eventing -l 'app in (kafka-webhook-eventing,kafka-broker-dispatcher,kafka-broker-receiver,kafka-controller)' 
  NAME                                       READY   STATUS    RESTARTS   AGE
  kafka-broker-dispatcher-5f474646df-29qlm   1/1     Running   0          6m19s
  kafka-broker-receiver-56dd84b6b7-njngl     1/1     Running   0          6m19s
  kafka-controller-8b648f844-q2zc4           1/1     Running   0          6m38s
  kafka-webhook-eventing-6f4966d4c5-928fd    1/1     Running   0          6m38s
  

KafkaBroker配置

• Kafka Broker自动创建KafkaChannel时遵循的配置保存于knative-eventig名称空间下的ConfigMaps/kafka-broker-config中:

  • bootstrap.servers: 即Kafka集群的Bootstrap Server的访问入口,例如,对于前面基于Strimzi Operator在Kafka名称空间下部署Kafka集群mycluster来说,其访问入口就是service资源my-cluster-kafka-bootstrap:9092；
  • default.topic.partitions: 在Topic上默认使用的partition的数量,默认为10；
  • default.topic.replication.factor : 在Topic上默认使用的复制因子，其值不能大于Kafka上的broker数量,即可用节点数，默认值为“3”；这里只有一个节点，需要修改为1

  apiVersion: v1
  kind: ConfigMap
  metadata:
    name: kafka-broker-config
    namespace: knative-eventing
  data:
    bootstrap.servers: my-cluster-kafka-bootstrap.kafka:9092
    default.topic.partitions: "5"
    default.topic.replication.factor: "1"
  

随后，我们即可创建和使用Kafka Broker。需要特别说明的是，在Knative Eventing上，Kafka Broker在底层仅支持使用KafkaChannel，而且，它需要根据配置指定的Kafka Cluster，以及相关的Partitions和Replication Factor的定义自动完成KafkaChannel的创建。因此，在创建Kafka Broker之前，通常需要为Knative Eventing全局或在特定名称空间上显式定义其相关的配置。

前面我们曾在event-demo名称空间上为MT-Channel Based Broker设定了默认要使用KafkaChannel，Kafka Broker的设定方式类似。

• 如下面的配置示例：指定dev名称空间默认使用KafkaBroker,且相关配置保存于knative-eventing名称空间下的ConfigMaps/kafka-broker-config

  apiVersion: v1
  kind: ConfigMap
  metadata:
    name: config-br-defaults
    namespace: knative-eventing
  data:
    default-br-config: |
      clusterDefault:
        brokerClass: MTChannelBasedBroker
        apiVersion: v1
        kind: ConfigMap
        name: config-br-default-channel
        namespace: knative-eventing
        delivery:
          retry: 10
          backoffPolicy: exponential
          backoffDelay: PT0.2S
      namespaceDefaults:
        event-demo:
          brokerClass: MTChannelBasedBroker
          apiVersion: v1
          kind: ConfigMap
          name: kafka-channel
          namespace: knative-eventing
        dev:
          brokerClass: Kafka
          apiVersion: v1
          kind: ConfigMap
          name: kafka-broker-config
          namespace: knative-eventing
  

• 下面可在dev的名称空间上测试KafkaBroker的使用

  ~$ kn broker create kbroker01 -n dev
  

  ~$ kn broker list -n dev
  NAME        URL                                                                            AGE   CONDITIONS   READY   REASON
  kbroker01   http://kafka-broker-ingress.knative-eventing.svc.cluster.local/dev/kbroker01   3s    7 OK / 7     True    
  

• 具体broker/trigger使用示例，参照上面。