2.2. Kafka Connect/S2I 集群配置

步骤

1. 编辑 KafkaConnect 或 KafkaConnect S2I 资源的 spec 属性。

   您可以配置的属性显示在此示例配置中：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: KafkaConnect 1
   metadata:
     name: my-connect-cluster
     annotations:
       strimzi.io/use-connector-resources: "true" 2
   spec:
     replicas: 3 3
     authentication: 4
       type: tls
       certificateAndKey:
         certificate: source.crt
         key: source.key
         secretName: my-user-source
     bootstrapServers: my-cluster-kafka-bootstrap:9092 5
     tls: 6
       trustedCertificates:
         - secretName: my-cluster-cluster-cert
           certificate: ca.crt
         - secretName: my-cluster-cluster-cert
           certificate: ca2.crt
     config: 7
       group.id: my-connect-cluster
       offset.storage.topic: my-connect-cluster-offsets
       config.storage.topic: my-connect-cluster-configs
       status.storage.topic: my-connect-cluster-status
       key.converter: org.apache.kafka.connect.json.JsonConverter
       value.converter: org.apache.kafka.connect.json.JsonConverter
       key.converter.schemas.enable: true
       value.converter.schemas.enable: true
       config.storage.replication.factor: 3
       offset.storage.replication.factor: 3
       status.storage.replication.factor: 3
     externalConfiguration: 8
       env:
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsAccessKey
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
               name: aws-creds
               key: awsSecretAccessKey
     resources: 9
       requests:
         cpu: "1"
         memory: 2Gi
       limits:
         cpu: "2"
         memory: 2Gi
     logging: 10
       type: inline
       loggers:
         log4j.rootLogger: "INFO"
     readinessProbe: 11
       initialDelaySeconds: 15
       timeoutSeconds: 5
     livenessProbe:
       initialDelaySeconds: 15
       timeoutSeconds: 5
     metrics: 12
       lowercaseOutputName: true
       lowercaseOutputLabelNames: true
       rules:
         - pattern: kafka.connect<type=connect-worker-metrics><>([a-z-]+)
           name: kafka_connect_worker_$1
           help: "Kafka Connect JMX metric worker"
           type: GAUGE
         - pattern: kafka.connect<type=connect-worker-rebalance-metrics><>([a-z-]+)
           name: kafka_connect_worker_rebalance_$1
           help: "Kafka Connect JMX metric rebalance information"
           type: GAUGE
     jvmOptions: 13
       "-Xmx": "1g"
       "-Xms": "1g"
     image: my-org/my-image:latest 14
     template: 15
       pod:
         affinity:
           podAntiAffinity:
             requiredDuringSchedulingIgnoredDuringExecution:
               - labelSelector:
                   matchExpressions:
                     - key: application
                       operator: In
                       values:
                         - postgresql
                         - mongodb
                 topologyKey: "kubernetes.io/hostname"
       connectContainer: 16
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"

   1

   根据需要，使用 KafkaConnect 或 KafkaConnectS2I。

   2

   为 Kafka Connect 集群启用 KafkaConnectors。

   3

   副本节点的数量。

   4

   Kafka Connect 集群的身份验证，使用 TLS 机制 （如下所示）、使用 OAuth bearer 令牌 或基于 SASL 的 SCRAM-SHA-512 或 PLAIN 机制。默认情况下，Kafka Connect 使用纯文本连接连接到 Kafka 代理。

   5

   用于连接到 Kafka Connect 集群的 bootstrap 服务器。

   6

   使用密钥名称进行 TLS 加密，在其下，TLS 证书以 X.509 格式存储到集群的 X.509 格式。如果证书存储在同一个 secret 中，则可以多次列出证书。

   7

   Kafka Connect worker （ 而不是连接器）配置。标准 Apache Kafka 配置可能会提供，仅限于不是由 AMQ Streams 直接管理的属性。

   8

   使用环境变量（如下所示）或卷 为 Kafka 连接器的外部配置。

   9

   请求保留 支持的资源、当前 cpu 和 memory ，以及限制，以指定可消耗的最大资源。

   10

   指定 Kafka 连接日志记录器和日志级别 直接（内联）或通过ConfigMap 间接（外部）添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。对于 Kafka Connect log4j.rootLogger 日志记录器，您可以将日志级别设置为 INFO、ERROR、WARN、TRACE、DEBUG、FATAL 或 OFF。

   11

   健康检查以了解 何时重新启动容器（存活度）以及容器何时可以接受流量（就绪度）。

   12

   Prometheus metrics，本例中通过 Prometheus JMX 导出器配置启用的 Prometheus 指标。您可以使用 metrics: {} 来在没有进一步配置的情况下启用指标。

   13

   运行 Kafka Connect 的虚拟机(VM)性能优化 JVM 配置选项。

   14

   高级选项：容器镜像配置，建议仅在特殊情况下使用。

   15

   模板自定义.在这里，pod 被调度为反关联性，因此 pod 不会调度到具有相同主机名的节点。

   16

   还 使用 Jaeger 为分布式追踪设置 环境变量。

2. 创建或更新资源：

   oc apply -f KAFKA-CONNECT-CONFIG-FILE

3. 如果为 Kafka Connect 启用了授权，请配置 Kafka Connect 用户以启用对 Kafka Connect consumer 组和主题的访问。