2.3. Kafka Connect 集群配置

使用 KafkaConnect 资源配置 Kafka Connect 部署。Kafka Connect 是一个在 Kafka 代理和其他使用连接器插件的系统间流传输数据的集成工具包。Kafka Connect 提供了一个框架，用于将 Kafka 与外部数据源或目标（如数据库）集成，如数据库，用于使用连接器导入或导出数据。连接器是提供所需的连接配置的插件。

第 6.2.61 节 “KafkaConnect 模式参考” 描述 KafkaConnect 资源的完整模式。

有关部署连接器插件的更多信息，请参阅使用连接器插件扩展 Kafka 连接。

2.3.1. 配置 Kafka 连接
复制链接

使用 Kafka Connect 为 Kafka 集群设置外部数据连接。使用 KafkaConnect 资源的属性来配置 Kafka Connect 部署。

KafkaConnector 配置

KafkaConnector 资源允许您以 OpenShift 原生的方式创建和管理 Kafka Connect 的连接器实例。

在 Kafka Connect 配置中，您可以通过添加 strimzi.io/use-connector-resources 注解来为 Kafka Connect 集群启用 KafkaConnectors。您还可以添加 构建配置，以便 AMQ Streams 使用数据连接所需的连接器插件自动构建容器镜像。Kafka Connect 连接器的外部配置通过 externalConfiguration 属性指定。

要管理连接器，您可以使用 KafkaConnector 自定义资源或 Kafka Connect REST API。KafkaConnector 资源必须部署到它们所链接的 Kafka Connect 集群相同的命名空间中。有关使用这些方法创建、重新配置或删除连接器的更多信息，请参阅添加连接器。

连接器配置作为 HTTP 请求的一部分传递并存储在 Kafka 本身中。ConfigMap 和机密是用于存储配置和机密数据的标准 OpenShift 资源。您可以使用 ConfigMap 和 Secret 来配置连接器的特定元素。然后，您可以在 HTTP REST 命令中引用配置值，并在需要时使配置保持独立且更安全。这个方法特别适用于机密数据，如用户名、密码或证书。

处理大量信息

您可以调整配置来处理大量信息。如需更多信息，请参阅处理大量信息。

先决条件

一个 OpenShift 集群
正在运行的 Cluster Operator

有关运行的步骤，请参阅在 OpenShift 中部署和升级 AMQ Streams 指南：

流程

编辑 KafkaConnect 资源的 spec 属性。

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

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect 
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true" 
spec:
  replicas: 3 
  authentication: 
    type: tls
    certificateAndKey:
      certificate: source.crt
      key: source.key
      secretName: my-user-source
  bootstrapServers: my-cluster-kafka-bootstrap:9092 
  tls: 
    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        certificate: ca.crt
      - secretName: my-cluster-cluster-cert
        certificate: ca2.crt
  config: 
    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
  build: 
    output: 
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 
      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
            sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
      - name: camel-telegram
        artifacts:
          - type: tgz
            url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
            sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
  externalConfiguration: 
    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: 
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging: 
    type: inline
    loggers:
      log4j.rootLogger: "INFO"
  readinessProbe: 
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metricsConfig: 
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-key
  jvmOptions: 
    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest 
  rack:
    topologyKey: topology.kubernetes.io/zone 
  template: 
    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer: 
      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect


metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"


spec:
  replicas: 3


  authentication:


    type: tls
    certificateAndKey:
      certificate: source.crt
      key: source.key
      secretName: my-user-source
  bootstrapServers: my-cluster-kafka-bootstrap:9092


  tls:


    trustedCertificates:
      - secretName: my-cluster-cluster-cert
        certificate: ca.crt
      - secretName: my-cluster-cluster-cert
        certificate: ca2.crt
  config:


    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
  build:


    output:


      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins:


      - name: debezium-postgres-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/2.1.3.Final/debezium-connector-postgres-2.1.3.Final-plugin.tar.gz
            sha512sum: c4ddc97846de561755dc0b021a62aba656098829c70eb3ade3b817ce06d852ca12ae50c0281cc791a5a131cb7fc21fb15f4b8ee76c6cae5dd07f9c11cb7c6e79
      - name: camel-telegram
        artifacts:
          - type: tgz
            url: https://repo.maven.apache.org/maven2/org/apache/camel/kafkaconnector/camel-telegram-kafka-connector/0.11.5/camel-telegram-kafka-connector-0.11.5-package.tar.gz
            sha512sum: d6d9f45e0d1dbfcc9f6d1c7ca2046168c764389c78bc4b867dab32d24f710bb74ccf2a007d7d7a8af2dfca09d9a52ccbc2831fc715c195a3634cca055185bd91
  externalConfiguration:


    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:


    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 2Gi
  logging:


    type: inline
    loggers:
      log4j.rootLogger: "INFO"
  readinessProbe:


    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  metricsConfig:


    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: my-config-map
        key: my-key
  jvmOptions:


    "-Xmx": "1g"
    "-Xms": "1g"
  image: my-org/my-image:latest


  rack:
    topologyKey: topology.kubernetes.io/zone


  template:


    pod:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: application
                    operator: In
                    values:
                      - postgresql
                      - mongodb
              topologyKey: "kubernetes.io/hostname"
    connectContainer:


      env:
        - name: JAEGER_SERVICE_NAME
          value: my-jaeger-service
        - name: JAEGER_AGENT_HOST
          value: jaeger-agent-name
        - name: JAEGER_AGENT_PORT
          value: "6831"

Copy to Clipboard

Toggle word wrap

1: 使用 KafkaConnect。
2: 为 Kafka Connect 集群启用 KafkaConnectors。
3: 用于运行服务的 worker 的副本节点数量。
4: Kafka Connect 集群的身份验证，指定为 mTLS、基于令牌的 OAuth、基于 SASL 的 SCRAM-SHA-256/SCRAM-SHA-512 或 PLAIN。默认情况下，Kafka Connect 使用纯文本连接连接到 Kafka 代理。
5: 用于连接到 Kafka Connect 集群的 Bootstrap 服务器。
6: 使用密钥名称进行 TLS 加密，其中 TLS 证书以 X.509 格式存储。如果证书存储在同一 secret 中，它可以多次列出。
7: worker 的 Kafka 连接配置（而不是连接器）。可以提供标准 Apache Kafka 配置，仅限于不受 AMQ Streams 直接管理的属性。
8: 构建用于自动使用连接器插件构建容器镜像的配置属性。
9: （必需）在推送新镜像的容器 registry 的配置。
10: （必需）连接器插件及其工件列表，以添加到新容器镜像中。每个插件必须配置至少一个 工件。
11: 使用环境变量（如此处所示）或卷为 Kafka 连接器的外部配置。您还可以使用 配置供应商插件 来加载来自外部来源的配置值。
12: 对支持的资源、当前 cpu 和内存的 保留请求，以及指定可消耗的最大资源的限制。
13: 指定 Kafka 连接日志记录器和日志级别直接(内联)或通过 ConfigMap 间接(外部)添加。自定义 ConfigMap 必须放在 log4j.properties 或 log4j2.properties 键下。对于 Kafka Connect log4j.rootLogger 日志记录器，您可以将日志级别设置为 INFO, ERROR, WARN, TRACE, DEBUG, FATAL 或 OFF。
14: 使用 HealthCheck 可以知道何时重启一个容器（存活度）以及何时一个容器可以开始接受流量（就绪度）。
15: Prometheus metrics，通过引用本例中的 Prometheus JMX exporter 配置的 ConfigMap 来启用。您可以使用对在 metricsConfig.valueFrom.configMapKeyRef.key 下包含空文件的 ConfigMap 的引用来启用指标。
16: JVM 配置选项优化运行 Kafka Connect 的虚拟机(VM)的性能。
17: ADVANCED OPTION：容器镜像配置，这只在特殊情况下建议使用。
18: SPECIALIZED OPTION： Rack awareness 配置用于部署。这是用于在同一位置（而不是跨地区）部署的专用选项。如果您希望连接器从最接近的副本而不是领导副本使用，则使用这个选项。在某些情况下，从最接近的副本消耗可以提高网络利用率或降低成本。topologyKey 必须与包含机架 ID 的节点标签匹配。此配置中使用的示例使用标准 topology.kubernetes.io/zone 标签指定区。要从最接近的副本使用，请在 Kafka 代理配置中启用 RackAwareReplicaSelector。
19: 模板自定义.这里的 pod 使用反关联性调度，因此 pod 不会调度到具有相同主机名的节点。
20: 为分布式追踪设置环境变量。

创建或更新资源：
```
oc apply -f KAFKA-CONNECT-CONFIG-FILE
```
```
oc apply -f KAFKA-CONNECT-CONFIG-FILE
```
Copy to Clipboard Toggle word wrap
如果为 Kafka Connect 启用授权，请将 Kafka Connect 用户配置为启用对 Kafka Connect 消费者组和主题的访问。

2.3.2. 为多个实例配置 Kafka 连接
复制链接

如果您正在运行多个 Kafka Connect 实例，您必须更改以下配置属性 的默认配置 ：

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    group.id: connect-cluster 
    offset.storage.topic: connect-cluster-offsets 
    config.storage.topic: connect-cluster-configs 
    status.storage.topic: connect-cluster-status  
    # ...
# ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    group.id: connect-cluster


    offset.storage.topic: connect-cluster-offsets


    config.storage.topic: connect-cluster-configs


    status.storage.topic: connect-cluster-status


    # ...
# ...

Copy to Clipboard

Toggle word wrap

1: Kafka 中的 Kafka Connect 集群 ID。
2: 存储连接器偏移的 Kafka 主题。
3: 存储连接器和任务状态配置的 Kafka 主题。
4: 存储连接器和任务状态更新的 Kafka 主题。

注意

对于具有相同 group.id 的所有 Kafka Connect 实例，这三个主题的值必须相同。

除非更改默认设置，否则每个 Kafka Connect 实例都使用相同的值部署到同一 Kafka 集群。实际上，所有实例都与集群中运行的实例合并，使用相同的主题。

如果多个 Kafka Connect 集群试图使用相同的主题，Kafka Connect 将无法正常工作，并生成错误。

如果要运行多个 Kafka Connect 实例，请更改每个实例的这些属性的值。

2.3.3. 配置 Kafka 连接用户授权
复制链接

这个步骤描述了如何授权用户对 Kafka Connect 的访问权限。

当在 Kafka 中使用任何类型的授权时，Kafka Connect 用户需要读/写权限到消费者组和 Kafka Connect 的内部主题。

消费者组和内部主题的属性由 AMQ Streams 自动配置，也可以在 KafkaConnect 资源的 spec 中明确指定它们。

KafkaConnect 资源中的配置属性示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    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 
    # ...
  # ...

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect
spec:
  # ...
  config:
    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


    # ...
  # ...

Copy to Clipboard

Toggle word wrap

1: Kafka 中的 Kafka Connect 集群 ID。
2: 存储连接器偏移的 Kafka 主题。
3: 存储连接器和任务状态配置的 Kafka 主题。
4: 存储连接器和任务状态更新的 Kafka 主题。

此流程演示了如何在使用 简单 授权时提供访问权限。

简单授权使用 Kafka AclAuthorizer 插件处理的 ACL 规则来提供正确访问级别。有关将 KafkaUser 资源配置为使用简单授权的更多信息，请参阅 AclRule schema reference。

注意

在运行多个实例时，消费者组和主题的默认值将有所不同。

先决条件

一个 OpenShift 集群
正在运行的 Cluster Operator

流程

编辑 KafkaUser 资源中的 授权 属性，为用户提供访问权限。

在以下示例中，使用 字面 名称值为 Kafka Connect 主题和消费者组配置访问权限：

Expand

属性	名称
`offset.storage.topic`	`connect-cluster-offsets`
`status.storage.topic`	`connect-cluster-status`
`config.storage.topic`	`connect-cluster-configs`
`group`	`connect-cluster`

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  authorization:
    type: simple
    acls:
      # access to offset.storage.topic
      - resource:
          type: topic
          name: connect-cluster-offsets
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to status.storage.topic
      - resource:
          type: topic
          name: connect-cluster-status
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to config.storage.topic
      - resource:
          type: topic
          name: connect-cluster-configs
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # consumer group
      - resource:
          type: group
          name: connect-cluster
          patternType: literal
        operations:
          - Read
        host: "*"

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaUser
metadata:
  name: my-user
  labels:
    strimzi.io/cluster: my-cluster
spec:
  # ...
  authorization:
    type: simple
    acls:
      # access to offset.storage.topic
      - resource:
          type: topic
          name: connect-cluster-offsets
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to status.storage.topic
      - resource:
          type: topic
          name: connect-cluster-status
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # access to config.storage.topic
      - resource:
          type: topic
          name: connect-cluster-configs
          patternType: literal
        operations:
          - Create
          - Describe
          - Read
          - Write
        host: "*"
      # consumer group
      - resource:
          type: group
          name: connect-cluster
          patternType: literal
        operations:
          - Read
        host: "*"

Copy to Clipboard

Toggle word wrap

创建或更新资源。
```
oc apply -f KAFKA-USER-CONFIG-FILE
```
```
oc apply -f KAFKA-USER-CONFIG-FILE
```
Copy to Clipboard Toggle word wrap

2.3.4. Kafka Connect 集群资源列表
复制链接

以下资源由 OpenShift 集群中的 Cluster Operator 创建：

connect-cluster-name-connect

提供给以下 Kafka Connect 资源的名称：

创建 Kafka Connect worker 节点 pod 的部署（当禁用 StableConnectIdentities 功能门时）。
创建 Kafka Connect worker 节点 pod 的 StrimziPodSet （当启用了 StableConnectIdentities 功能门时）。
为连接 pod 提供稳定的 DNS 名称的无头服务（当启用了 StableConnectIdentities 功能门时）。
为 Kafka Connect worker 节点配置的 Pod Disruption Budget。

connect-cluster-name-connect-idx

Kafka Connect StrimziPodSet 创建的 Pod （当启用了 StableConnectIdentities 功能门时）。

connect-cluster-name-connect-api

公开用于管理 Kafka Connect 集群的 REST 接口的服务。

connect-cluster-name-config

包含 Kafka Connect 辅助配置的 ConfigMap，并由 Kafka 代理 pod 作为卷挂载。

2.3.5. 与红帽构建的 Debezium 集成以更改数据捕获
复制链接

红帽构建的 Debezium 是一个分布式更改数据捕获平台。它捕获数据库中的行级更改，创建更改事件记录，并将记录流传输到 Kafka 主题。Debezium 基于 Apache Kafka 构建。您可以将红帽构建的 Debezium 与 AMQ Streams 一起部署并集成。部署 AMQ Streams 后，您可以通过 Kafka Connect 将 Debezium 部署为连接器配置。Debezium 将更改事件记录传递到 OpenShift 上的 AMQ Streams。应用程序可以读取 这些更改事件流，并按发生更改事件的顺序访问更改事件。

Debezium 具有多个用途，包括：

数据复制
更新缓存和搜索索引
简化单体式应用程序
数据集成
启用流查询

要捕获数据库更改，请使用 Debezium 数据库连接器部署 Kafka 连接。您可以配置 KafkaConnector 资源来定义连接器实例。

有关将红帽构建的 Debezium 与 AMQ Streams 一起部署的更多信息，请参阅产品文档。文档包括 Debezium 入门指南，指导您完成设置数据库更新事件记录所需的服务和连接器。

返回顶部

2.3. Kafka Connect 集群配置

2.3.1. 配置 Kafka 连接
复制链接

2.3.2. 为多个实例配置 Kafka 连接
复制链接

2.3.3. 配置 Kafka 连接用户授权
复制链接

2.3.4. Kafka Connect 集群资源列表
复制链接

2.3.5. 与红帽构建的 Debezium 集成以更改数据捕获
复制链接

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

2.3. Kafka Connect 集群配置

2.3.1. 配置 Kafka 连接复制链接链接已复制到粘贴板!

2.3.2. 为多个实例配置 Kafka 连接复制链接链接已复制到粘贴板!

2.3.3. 配置 Kafka 连接用户授权复制链接链接已复制到粘贴板!

2.3.4. Kafka Connect 集群资源列表复制链接链接已复制到粘贴板!

2.3.5. 与红帽构建的 Debezium 集成以更改数据捕获复制链接链接已复制到粘贴板!

学习

尝试、购买和销售

社区

关于红帽文档

让开源更具包容性

關於紅帽

Theme

Red Hat legal and privacy links

Red Hat legal and privacy links

2.3.1. 配置 Kafka 连接
复制链接

2.3.2. 为多个实例配置 Kafka 连接
复制链接

2.3.3. 配置 Kafka 连接用户授权
复制链接

2.3.4. Kafka Connect 集群资源列表
复制链接

2.3.5. 与红帽构建的 Debezium 集成以更改数据捕获
复制链接