Red Hat Summit 红帽全球峰会支持控制台（Console）开发人员开始试用联系人

8.6. 配置 Kafka Connect

更新 KafkaConnect 自定义资源的 spec 属性来配置 Kafka Connect 部署。

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

要深入了解 Kafka Connect 集群配置选项，请参阅 AMQ Streams 自定义资源 API 参考。

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 Connect，并存储在 Kafka 本身中。ConfigMap 和机密是用于存储配置和机密数据的标准 OpenShift 资源。您可以使用 ConfigMap 和 Secret 来配置连接器的特定元素。然后，您可以在 HTTP REST 命令中引用配置值，这样可保持配置独立且更安全。此方法特别适用于机密数据，如用户名、密码或证书。

处理大量信息

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

KafkaConnect 自定义资源配置示例

apiVersion: kafka.strimzi.io/v1beta2
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
  build:
8 

    output:
9 

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

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

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

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

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

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

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

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

  rack:
    topologyKey: topology.kubernetes.io/zone
18 

  template:
19 

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

      env:
        - name: OTEL_SERVICE_NAME
          value: my-otel-service
        - name: OTEL_EXPORTER_OTLP_ENDPOINT
          value: "http://otlp-host:4317"
  tracing:
    type: opentelemetry
21
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 集群的 bootstrap 服务器。
6
TLS 加密，使用密钥名称，其中 TLS 证书存储为集群的 X.509 格式。如果证书存储在同一 secret 中，则可以多次列出。
7
worker 的 Kafka 连接配置（而不是连接器）。标准 Apache Kafka 配置可能会提供，仅限于不直接由 AMQ Streams 管理的属性。
8
构建用于自动使用连接器插件构建容器镜像的配置属性。
9
（必需）推送新镜像的容器 registry 的配置。
10
（必需）添加到新容器镜像的连接器插件及其工件列表。每个插件必须配置至少一个 工件。
11
使用环境变量的连接器的外部配置，如此处或卷所示。您还可以使用配置供应商插件从外部来源加载配置值。
12
用于保留支持的资源、当前 cpu 和内存 以及限制的请求，以指定可消耗的最大资源。
13
指定 Kafka Connect 日志记录器和日志级别直接(内联)或通过 ConfigMap 间接(外部)。自定义 Log4j 配置必须放在 ConfigMap 中的 log4j.properties 或 log4j2.properties 键下。对于 Kafka Connect log4j.rootLogger 日志记录器，您可以将日志级别设置为 INFO, ERROR, WARN, TRACE, DEBUG,FATAL 或 OFF。
14
健康检查以了解何时重启容器（持续）以及容器可以接受流量（就绪状态）。
15
Prometheus 指标，通过引用包含在此示例中 Prometheus JMX 导出器配置的 ConfigMap 启用。您可以使用对 metricsConfig.valueFrom.configMapKeyRef.key 下包含空文件的 ConfigMap 的引用来启用指标。
16
JVM 配置选项，用于优化运行 Kafka Connect 的虚拟机(VM)的性能。
17
ADVANCED OPTION：容器镜像配置，这只在特殊情况下建议使用。
18
SPECIALIZED OPTION：部署的机架感知配置。这是用于在同一位置（而非跨地区）部署的专用选项。如果您希望连接器从最接近的副本而不是领导副本使用，则使用此选项。在某些情况下，使用来自最接近的副本的消耗可以提高网络利用率或降低成本。topologyKey 必须与包含机架 ID 的节点标签匹配。此配置中使用的示例使用标准 topology.kubernetes.io/zone 标签指定区。要从最接近的副本使用，请在 Kafka 代理配置中启用 RackAwareReplicaSelector。
19
模板自定义。此处的 pod 使用反关联性调度，因此 pod 不会调度到具有相同主机名的节点。
20
为分布式追踪设置环境变量。
21
使用 OpenTelemetry 启用分布式追踪。

8.6.1. 配置 Kafka Connect 用户授权
复制链接

这个步骤描述了如何授权用户对 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
1 

    offset.storage.topic: my-connect-cluster-offsets
2 

    config.storage.topic: my-connect-cluster-configs
3 

    status.storage.topic: my-connect-cluster-status
4 

    # ...
  # ...
Copy to Clipboard Toggle word wrap

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

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

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

注意

先决条件

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

流程

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

    在以下示例中，使用 字面 名称值为 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: "*"
    Copy to Clipboard Toggle word wrap

  2. 创建或更新资源。 

    oc apply -f KAFKA-USER-CONFIG-FILE
    Copy to Clipboard Toggle word wrap
返回顶部
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务，以及可以信赖的内容，帮助红帽用户创新并实现他们的目标。 了解我们当前的更新.

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情，请参阅红帽博客.

關於紅帽

我们提供强化的解决方案，使企业能够更轻松地跨平台和环境（从核心数据中心到网络边缘）工作。

Theme

© 2025 Red Hat