如何通过Knative创建CronJob向Kafka发送健康检查消息及解决配置问题

1. 前置准备：创建Kafka Topic #

先确保目标Kafka集群存在用于健康检查的Topic，以Strimzi为例：

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: health-check-topic
  namespace: kafka
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 3
  replicas: 3
  config:
    retention.ms: 7200000
    segment.bytes: 1073741824

执行kubectl apply -f <filename.yaml>完成创建。

2. 部署CronJob向Kafka发送消息 #

编写一个简单的消息发送脚本（比如Python/Go），打包成镜像后，配置CronJob每10分钟执行一次：

apiVersion: batch/v1
kind: CronJob
metadata:
  name: kafka-health-sender
spec:
  schedule: "*/10 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: sender
            image: your-registry/kafka-sender:v1
            env:
            - name: KAFKA_BOOTSTRAP_SERVERS
              value: "my-cluster-kafka-bootstrap.kafka:9093"
            - name: KAFKA_TOPIC
              value: "health-check-topic"
            - name: TLS_ENABLED
              value: "true"
            # 挂载TLS凭证
            volumeMounts:
            - name: kafka-certs
              mountPath: /etc/kafka/certs
              readOnly: true
          volumes:
          - name: kafka-certs
            secret:
              secretName: kafka-client-tls-secret # 包含ca.crt、user.crt、user.key的Secret
          restartPolicy: OnFailure

脚本中需要读取挂载的证书文件，建立TLS连接并发送健康检查消息（比如携带type: health.check.event的CloudEvent格式数据）。

3. 配置KafkaSource抓取消息到Knative Broker #

KafkaSource属于Knative Eventing的Source组件，apiVersion为sources.knative.dev/v1beta1是正常的（和KafkaBinding分属不同CRD组），带TLS的配置示例：

apiVersion: sources.knative.dev/v1beta1
kind: KafkaSource
metadata:
  name: kafka-health-source
spec:
  bootstrapServers:
  - my-cluster-kafka-bootstrap.kafka:9093
  topics:
  - health-check-topic
  consumerGroup: knative-health-check-group
  sink:
    ref:
      apiVersion: eventing.knative.dev/v1
      kind: Broker
      name: default
  tls:
    enable: true
    caCert:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: ca.crt
    cert:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: user.crt
    key:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: user.key

执行kubectl apply -f <filename.yaml>部署，该Source会订阅指定Topic，将消息转发到默认Broker。

4. 创建Trigger路由消息到接收服务 #

先部署接收健康检查消息的Knative Service（或普通K8s Service）：

apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: health-check-receiver
spec:
  template:
    spec:
      containers:
      - image: your-registry/health-receiver:v1

再创建Trigger，过滤并路由Broker中的消息到接收服务：

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: health-check-trigger
spec:
  broker: default
  filter:
    attributes:
      type: health.check.event # 和发送消息的type一致
  subscriber:
    ref:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: health-check-receiver

解决KafkaBinding的TLS配置问题 #

KafkaBinding（bindings.knative.dev/v1beta1）用于给工作负载注入Kafka连接信息，若官方模板缺少TLS配置，可手动扩展：

apiVersion: bindings.knative.dev/v1beta1
kind: KafkaBinding
metadata:
  name: kafka-health-binding
spec:
  subject:
    apiVersion: batch/v1
    kind: CronJob
    name: kafka-health-sender
  bootstrapServers:
  - my-cluster-kafka-bootstrap.kafka:9093
  tls:
    enable: true
    caCert:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: ca.crt
    clientCert:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: user.crt
    clientKey:
      secretKeyRef:
        name: kafka-client-tls-secret
        key: user.key

注意：需确保Knative版本支持KafkaBinding的TLS字段，若版本过旧建议升级。

替代方案（针对KafkaSource不稳定的情况） #

如果KafkaSource稳定性不符合预期，可尝试：

• Strimzi Kafka Connector：使用Strimzi官方的Knative Sink连接器，直接将Kafka消息转发到Knative Broker，配置更灵活、成熟。
• 自定义Kafka Consumer服务：编写常驻的Kafka Consumer程序，部署为Knative Service，订阅Topic后直接将消息转发到接收端点，跳过KafkaSource和Broker，适合快速测试流程。

稳定性排查建议 #

• 确认Knative Eventing与KafkaSource的版本兼容，参考官方兼容性矩阵。
• 查看KafkaSource日志定位问题：kubectl logs -l sources.knative.dev/name=kafka-health-source -c kafka-source
• 调整KafkaSource的消费者配置（如重试次数、offset重置策略）优化稳定性。

内容的提问来源于stack exchange，提问作者Prox