AMQ Streams on OpenShift 概述

Kafka 的底层数据流处理功能和组件架构可以提供：

Kafka 的功能使其适合：

AMQ Streams 为在 OpenShift 中运行 Kafka 提供容器镜像和操作器。AMQ Streams operator 是专门构建的，具有专家操作知识，以便在 OpenShift 上有效地管理 Kafka。

Operator 简化以下过程：

Kafka 代理集群处理消息的发送。

代理使用 Apache ZooKeeper 存储配置数据和集群协调。在运行 Apache Kafka 之前，Apache ZooKeeper 集群必须就绪。

其他 Kafka 组件与 Kafka 集群交互，以执行特定的角色。

要最大化使用 AMQ Streams 的经验，您需要了解 Kafka 如何作为消息代理运行。

Kafka 集群由多个代理组成。代理包含接收和存储数据的主题。主题按分区分割，分区中写入数据。分区在主题之间复制，以进行容错。

生产者和消费者通过代理发送和接收消息（发布和订阅）。消息包含一个可选的键和值，它包括了消息数据以及标头和相关的元数据。密钥用于识别消息的主题或消息的属性。消息在批处理中传输，并且记录包含标头和元数据，它们提供客户端过滤和路由相关的详细信息，如记录的时间戳和偏移位置。

Kafka Connect 使用连接器实例来与其他系统集成以流传输数据。

Kafka Connect 在启动时加载现有连接器实例，并在 worker pod 间分配数据流任务和连接器配置。worker 为连接器实例运行任务。每个 worker 作为单独的 pod 运行，使 Kafka Connect 集群更具容错性。如果任务数量超过 worker，则 worker 会被分配多个任务。如果 worker 失败，其任务会自动分配给 Kafka Connect 集群中的活跃 worker。

流数据中使用的主要的 Kafka Connect 组件如下：

Kafka Bridge 支持对 Kafka 集群的 HTTP 请求，使用以下方法：

该方法提供 JSON 响应和 HTTP 响应代码错误处理。消息可以使用 JSON 或二进制格式发送。

客户端可以在不需要使用原生 Kafka 协议的情况下生成和使用消息。

您可以使用 Kafka Bridge 将内部和外部 HTTP 客户端应用程序与 Kafka 集群集成。

AMQ Streams 使用 Cluster Operator 来部署和管理集群。默认情况下，当您部署 AMQ Streams 时，会部署单个 Cluster Operator 副本。您可以使用领导选举机制添加副本，以便在出现问题时有其他 Cluster Operator 处于待机状态。

Cluster Operator 管理以下 Kafka 组件的集群：

集群使用自定义资源部署。

例如，要部署 Kafka 集群：

Cluster Operator 还可通过配置 Kafka 资源来部署以下 AMQ Streams 操作器：

在部署时实体 Operator 中的主题 Operator 和用户 Operator 功能。

您可以使用 Cluster Operator 部署 AMQ Streams Drain Cleaner，以帮助 pod 驱除。通过部署 AMQ Streams Drain Cleaner，您可以使用 Cluster Operator 移动 Kafka pod 而不是 OpenShift。AMQ Streams Drain Cleaner 使用滚动更新注解来注解被驱除的 pod。该注解告知 Cluster Operator 执行滚动更新。

主题 Operator 提供了通过 KafkaTopic 资源管理 Kafka 集群中的主题的方法。

主题 Operator 通过监视 describe Kafka 主题的 KafkaTopic 资源来管理 Kafka 主题，并确保它们在 Kafka 集群中正确配置。

当创建、删除或更改 KafkaTopic 时，主题 Operator 对 Kafka 主题执行对应的操作。

您可以将 KafkaTopic 声明为应用程序部署的一部分，主题 Operator 会为您管理 Kafka 主题。

主题 Operator 以以下模式运行：

User Operator 提供了通过 KafkaUser 资源管理 Kafka 集群中的用户的方法。

User Operator 通过监视用于描述 Kafka 用户的 Kafka 资源来管理 Kafka 集群的 Kafka 用户，并确保它们在 Kafka 集群中正确配置。

创建、删除或更改 KafkaUser 时，User Operator 对 Kafka 用户执行对应的操作。

您可以将 KafkaUser 资源声明为应用程序部署的一部分，User Operator 会为您管理 Kafka 用户。您可以为用户指定身份验证和授权机制。您还可以配置控制使用 Kafka 资源的 用户配额 以确保用户不单调访问代理。

创建用户时，用户凭证会在 Secret 中创建。您的应用需要使用用户及其凭据来进行身份验证，才能生成或消耗消息。

除了管理用于身份验证的凭据外，User Operator 还通过在 KafkaUser 声明中包含用户访问权限的描述来管理授权规则。

您可以使用 功能门 启用和禁用 Operator 的一些功能。

功能门在 Operator 配置中设置，且有三个成熟度阶段： alpha、beta 或 General Availability (GA)。

如需更多信息，请参阅功能门。

在安装 CRD 中添加新的自定义资源类型后，您可以根据规格创建资源实例。

AMQ Streams 组件的自定义资源具有通用配置属性，它们在 spec 下定义。

在 Kafka 主题自定义资源中的这种片段中，apiVersion 和 kind 属性标识关联的 CRD。spec 属性显示定义主题的分区和副本数的配置。

Kafka 主题自定义资源

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaTopic
metadata:
  name: my-topic
  labels:
    strimzi.io/cluster: my-cluster
spec:
  partitions: 1
  replicas: 1
  # ...

有很多额外的配置选项可以合并到 YAML 定义中，一些常见和特定于特定组件的配置选项。

此处介绍了一些与资源通用的配置选项。在适当时，Security 和 metrics collection 可能也会采用。

显示常见配置的 YAML 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-cluster
spec:
  # ...
  bootstrapServers: my-cluster-kafka-bootstrap:9092
  resources:
    requests:
      cpu: 12
      memory: 64Gi
    limits:
      cpu: 12
      memory: 64Gi
  logging:
    type: inline
    loggers:
      connect.root.logger.level: INFO
  readinessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  livenessProbe:
    initialDelaySeconds: 15
    timeoutSeconds: 5
  jvmOptions:
    "-Xmx": "2g"
    "-Xms": "2g"
  template:
    pod:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
              - matchExpressions:
                  - key: node-type
                    operator: In
                    values:
                      - fast-network
  # ...

kafka 集群包含一个或多个代理。要使生产者和消费者能够访问代理中的主题，Kafka 配置必须定义如何存储数据，以及如何访问数据。您可以将 Kafka 集群配置为使用多个代理节点在机架之间运行。

如果您使用 OAuth 2.0 进行基于令牌的身份验证，您可以将监听程序配置为使用授权服务器。

显示 Kafka 配置的 YAML 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    # ...
    listeners:
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external1
        port: 9094
        type: route
        tls: true
        authentication:
          type: tls
    # ...
    storage:
      type: persistent-claim
      size: 10000Gi
    # ...
    rack:
      topologyKey: topology.kubernetes.io/zone
    config:
      replica.selector.class: org.apache.kafka.common.replica.RackAwareReplicaSelector
    # ...

节点池指的是 Kafka 集群中不同的 Kafka 节点组。通过使用节点池，节点可以在同一 Kafka 集群中有不同的配置。在节点池中没有指定的配置选项从 Kafka 配置继承。

节点池功能作为技术预览提供，可以使用 KafkaNodePool 功能门启用。您可以使用一个或多个节点池部署 Kafka 集群。节点池配置包括强制和可选设置。副本、角色和存储的配置是必需的。

如果您使用 KRaft 模式（也可以作为技术预览提供），您可以为节点池中的所有节点指定角色，以作为代理、控制器或两者运行。控制器和双角色特定于具体的 KRaft。如果您使用 ZooKeeper 用于集群管理的 Kafka 集群，则只能使用使用代理角色配置的节点池。

显示节点池配置的 YAML 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaNodePool
metadata:
  name: pool-a
  labels:
    strimzi.io/cluster: my-cluster
spec:
  replicas: 3
  roles:
    - broker
  storage:
    type: jbod
    volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false

Kafka MirrorMaker 2 在两个或多个活跃 Kafka 集群之间复制数据，并在数据中心之间复制数据。要设置 MirrorMaker 2，源和目标（目标） Kafka 集群必须正在运行。

将数据从源集群镜像 (mirror)到目标集群的过程是异步的。每个 MirrorMaker 2 实例将数据从一个源集群镜像到一个目标集群。您可以使用多个 MirrorMaker 2 实例在任意数量的集群间镜像数据。

图 7.1. 在两个集群间复制

MirrorMaker 2 使用源和目标集群配置，如下所示：

主题和消费者组复制以逗号分隔的列表或正则表达式模式指定。

显示 MirrorMaker 2 配置的 YAML 示例

  apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaMirrorMaker2
  metadata:
    name: my-mirror-maker2
    spec:
      version: 3.5.0
      connectCluster: "my-cluster-target"
      clusters:
      - alias: "my-cluster-source"
        bootstrapServers: my-cluster-source-kafka-bootstrap:9092
      - alias: "my-cluster-target"
        bootstrapServers: my-cluster-target-kafka-bootstrap:9092
      mirrors:
      - sourceCluster: "my-cluster-source"
        targetCluster: "my-cluster-target"
        sourceConnector: {}
      topicsPattern: ".*"
      groupsPattern: "group1|group2|group3"

Kafka MirrorMaker （也称为 MirrorMaker 1）使用制作者和消费者在集群间复制数据，如下所示：

消费者和制作者配置包括任何所需的身份验证和加密设置。include 属性定义要从源镜像到目标集群的主题。

关键的消费者配置

关键的生成者配置

显示 MirrorMaker 配置的 YAML 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaMirrorMaker
metadata:
  name: my-mirror-maker
spec:
  # ...
  consumer:
    bootstrapServers: my-source-cluster-kafka-bootstrap:9092
    groupId: "my-group"
    numStreams: 2
    offsetCommitInterval: 120000
    # ...
  producer:
    # ...
    abortOnSendFailure: false
    # ...
  include: "my-topic|other-topic"
  # ...

使用 AMQ Streams 的 KafkaConnect 资源快速轻松地创建新的 Kafka Connect 集群。

当使用 KafkaConnect 资源部署 Kafka Connect 时，您可以指定 bootstrap 服务器地址（在 spec.bootstrapServers中）以连接到 Kafka 集群。当服务器停机时，您可以指定多个地址。您还指定身份验证凭据和 TLS 客户端证书，以建立安全连接。

您还可以使用 KafkaConnect 资源来指定以下内容：

Cluster Operator 管理使用 KafkaConnector 资源部署的 Kafka Connect 集群，以及利用 KafkaConnector 资源创建的连接器。

插件配置

插件提供创建连接器实例的实施。当插件实例化时，会为连接特定类型的外部数据系统提供配置。插件提供一组或者多个 JAR 文件，该文件定义了连接器和任务实施，以连接到指定类型数据源。许多外部系统的插件可用于 Kafka 连接。您还可以创建自己的插件。

配置描述了要发送到 Kafka Connect 的源输入数据和目标输出数据。对于源连接器，外部源数据必须引用要存储消息的特定主题。插件也可以包含转换数据所需的库和文件。

Kafka Connect 部署可以有一个或多个插件，但每个插件只能有一个版本。

您可以创建自定义 Kafka Connect 镜像，其中包括您选择的插件。您可以通过两种方式创建镜像：

要自动创建容器镜像，您可以使用 KafkaConnect 资源的 build 属性指定要添加到 Kafka Connect 集群中的插件。AMQ Streams 会自动下载插件工件并将其添加到新容器镜像中。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  # ...
  build: 1
    output: 2
      type: docker
      image: my-registry.io/my-org/my-connect-cluster:latest
      pushSecret: my-registry-credentials
    plugins: 3
      - name: my-connector
        artifacts:
          - type: tgz
            url: https://<plugin_download_location>.tgz
            sha512sum: <checksum_to_verify_the_plugin>
      # ...
  # ...

如果使用 Dockerfile 构建镜像，您可以使用 AMQ Streams 的最新容器镜像作为基础镜像来添加插件配置文件。

FROM registry.redhat.io/amq-streams/kafka-35-rhel8:2.5.1
USER root:root
COPY ./my-plugins/ /opt/kafka/plugins/
USER 1001

worker 的 Kafka Connect 集群配置

您可以在 KafkaConnect 资源的 config 属性中指定 worker 的配置。

分布式 Kafka Connect 集群有一个组 ID 和一组内部配置主题。

Kafka Connect 集群默认使用这些属性的值配置。Kafka Connect 集群无法共享组 ID 或主题名称，因为它将创建错误。如果使用多个不同的 Kafka Connect 集群，则每个创建的 Kafka Connect 集群的 worker 必须是唯一的。

每个 Kafka Connect 集群使用的连接器名称也必须是唯一的。

在以下示例中，指定了 JSON 转换器。Kafka Connect 使用的内部 Kafka 主题设置了复制因素。对于生产环境，至少应有 3 个。在创建主题后更改复制因素将无效。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
# ...
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
    key.converter: org.apache.kafka.connect.json.JsonConverter 5
    value.converter: org.apache.kafka.connect.json.JsonConverter 6
    key.converter.schemas.enable: true 7
    value.converter.schemas.enable: true 8
    config.storage.replication.factor: 3 9
    offset.storage.replication.factor: 3 10
    status.storage.replication.factor: 3 11
  # ...

连接器的 KafkaConnector 管理

在将插件添加到用于部署中的 worker pod 的容器镜像后，您可以使用 AMQ Streams 的 KafkaConnector 自定义资源或 Kafka Connect API 来管理连接器实例。您还可以使用这些选项创建新的连接器实例。

KafkaConnector 资源提供了一种 OpenShift 原生的方法来管理 Cluster Operator 连接器。要使用 KafkaConnector 资源管理连接器，您必须在 KafkaConnect 自定义资源中指定注解。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
  # ...

将 use-connector-resources 设置为 true 可启用 KafkaConnectors 创建、删除和重新配置连接器。

如果在 KafkaConnect 配置中启用了 use-connector-resources，则必须使用 KafkaConnector 资源来定义和管理连接器。KafkaConnector 资源被配置为连接到外部系统。它们部署到与 Kafka Connect 集群和与外部数据系统交互的 Kafka 集群相同的 OpenShift 集群。

配置指定连接器实例如何连接到外部数据系统，包括任何身份验证。您还需要陈述要监视的数据。对于源连接器，您可能会在配置中提供数据库名称。您还可以通过指定目标主题名称指定数据在 Kafka 中的位置。

使用 tasksMax 指定最大任务数。例如，带有 tasksMax: 2 的源连接器可能会将源数据导入两个任务。

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: my-source-connector  1
  labels:
    strimzi.io/cluster: my-connect-cluster 2
spec:
  class: org.apache.kafka.connect.file.FileStreamSourceConnector 3
  tasksMax: 2 4
  config: 5
    file: "/opt/kafka/LICENSE" 6
    topic: my-topic 7
    # ...

Kafka Connect API

使用 Kafka Connect REST API 作为使用 KafkaConnector 资源管理连接器的替代选择。Kafka Connect REST API 作为一个运行在 <connect_cluster_name>-connect-api:8083 的服务其中 <connect_cluster_name> 是 Kafka Connect 集群的名称。

您可以将连接器配置添加为 JSON 对象。

curl -X POST \
  http://my-connect-cluster-connect-api:8083/connectors \
  -H 'Content-Type: application/json' \
  -d '{ "name": "my-source-connector",
    "config":
    {
      "connector.class":"org.apache.kafka.connect.file.FileStreamSourceConnector",
      "file": "/opt/kafka/LICENSE",
      "topic":"my-topic",
      "tasksMax": "4",
      "type": "source"
    }
}'

如果启用了 KafkaConnectors，Cluster Operator 将恢复使用 Kafka Connect REST API 进行的手动更改。

REST API 支持的操作在 Apache Kafka Connect API 文档中 描述。

Kafka Bridge 配置需要它连接到的 Kafka 集群的 bootstrap 服务器规格，以及所需的加密和身份验证选项。

Kafka Bridge 消费者和生成者配置是标准的，如 Apache Kafka configuration documentation for consumers 和 Apache Kafka configuration documentation for producers 所述。

与 HTTP 相关的配置选项设置服务器侦听的端口连接。

CORS

Kafka Bridge 支持使用 Cross-Origin Resource Sharing (CORS)。CORS 是一种 HTTP 机制，它允许浏览器从多个来源访问选定资源，例如，不同域中的资源。如果您选择使用 CORS，可以定义一个允许的资源来源列表，并通过 HTTP 方法通过 Kafka Bridge 与 Kafka 集群交互。列表在 Kafka Bridge 配置的 http 规格中定义。

CORS 允许在不同域中的源之间的 简单 和 preflighted 请求。

显示 Kafka Bridge 配置的 YAML 示例

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaBridge
metadata:
  name: my-bridge
spec:
  # ...
  bootstrapServers: my-cluster-kafka:9092
  http:
    port: 8080
    cors:
      allowedOrigins: "https://strimzi.io"
      allowedMethods: "GET,POST,PUT,DELETE,OPTIONS,PATCH"
  consumer:
    config:
      auto.offset.reset: earliest
  producer:
    config:
      delivery.timeout.ms: 300000
  # ...

AMQ Streams 支持传输层安全(TLS)，这是一个用于加密通信的协议。

在通信是始终进行加密：

您还可以在 Kafka 代理和客户端之间配置 TLS 加密。在为 Kafka 代理配置外部监听程序时，为外部客户端指定 TLS。

AMQ Streams 组件和 Kafka 客户端使用数字证书进行加密。Cluster Operator 设置证书以便在 Kafka 集群中启用加密。您可以提供自己的服务器证书，称为 Kafka 侦听器证书，用于 Kafka 客户端和 Kafka 代理之间的通信，以及集群间通信。

AMQ Streams 使用 Secret 以 PEM 和 PKCS #12 格式存储 mTLS 所需的证书和私钥。

TLS CA （证书颁发机构）发布证书来验证组件的身份。AMQ Streams 根据 CA 证书验证组件的证书。

Kafka 侦听程序使用身份验证来确保到 Kafka 集群的安全客户端连接。

支持的验证机制：

User Operator 为 mTLS 和 SCRAM 身份验证管理用户凭证，但不管理 OAuth 2.0。例如，通过 User Operator，您可以创建一个代表需要访问 Kafka 集群的客户端的用户，并指定 tls 作为身份验证类型。

使用基于 OAuth 2.0 令牌的身份验证时，应用程序客户端可以在不公开帐户凭证的情况下访问 Kafka 代理。授权服务器处理访问权限的授予和询问有关访问权限的查询。

自定义身份验证允许任何类型的 kafka 支持的验证。它可以提供更大的灵活性，但也增加了复杂性。

Kafka 集群使用授权来控制特定客户端或用户在 Kafka 代理上允许的操作。如果应用到 Kafka 集群，会为用于客户端连接的所有监听程序启用授权。

如果一个用户被添加到 Kafka 代理配置的 超级用户列表中，无论通过授权机制实施的任何授权限制，用户都可以无限地访问集群。

支持的授权机制：

简单授权使用 AclAuthorizer，这是默认的 Kafka 授权插件。AclAuthorizer 使用访问控制列表 (ACL) 来定义哪些用户可以访问哪些资源。对于自定义授权，您可以将自己的 Authorizer 插件配置为强制实施 ACL 规则。

OAuth 2.0 和 OPA 从授权服务器提供基于策略的控制。在授权服务器中定义用于授予对 Kafka 代理资源访问权限的安全策略和权限。

URL 用于连接到授权服务器，并验证客户端或用户请求的操作是允许或拒绝的。用户和客户端与授权服务器中创建的策略匹配，允许访问 Kafka 代理上的特定操作。

联邦信息处理标准(FIPS)是由美国政府建立的一组安全标准，以确保信息系统处理或传输敏感数据和信息的保密性、完整性和可用性。在启用了 FIPS 的 OpenShift 集群运行时，AMQ Streams 容器镜像中使用的 OpenJDK 会自动启用 FIPS 模式。

Prometheus 可以从 Kafka 组件和 AMQ Streams Operator 中提取指标数据。

要使用 Prometheus 获取指标数据并提供警报，必须部署 Prometheus 和 Prometheus Alertmanager 插件。还必须使用指标配置部署或重新部署 Kafka 资源，以公开指标数据。

Prometheus 提取公开的指标数据用于监控。当条件根据预定义的警报规则来指示潜在的问题时，Alertmanager 会发出警报。

AMQ Streams 提供了指标和警报规则配置文件示例。AMQ Streams 提供的示例警报机制被配置为将通知发送到 Slack 频道。

Grafana 使用 Prometheus 公开的指标数据呈现用于监控的仪表板视觉化。

需要部署 Grafana，Prometheus 会添加为数据源。由 AMQ Streams 作为 JSON 文件提供的仪表板示例通过 Grafana 界面导入，以显示监控数据。

Kafka Exporter 是一个开源项目，用于增强对 Apache Kafka 代理和客户端的监控。Kafka Exporter 通过 Kafka 集群部署，从与偏移、消费者组、消费者群和主题相关的 Kafka 代理提取额外的 Prometheus 指标数据。您可以使用提供的 Grafana 仪表板来视觉化 Prometheus 从 Kafka Exporter 收集的数据。

AMQ Streams 提供了示例配置文件、警报规则和 Kafka Exporter 的 Grafana 仪表板。

分布式追踪跟踪分布式系统中应用程序之间的事务进度。在微服务架构中，追踪跟踪服务间的事务进度。跟踪数据可用于监控应用程序性能和调查目标系统和最终用户应用的问题。

在 AMQ Streams 中，追踪有助于对消息的端到端跟踪：从源系统到 Kafka，然后从 Kafka 到目标系统和应用程序。分布式追踪补充了 Grafana 仪表板中的指标的监控，以及组件日志记录器。

以下 Kafka 组件内置了对追踪的支持：

Kafka 代理不支持追踪。

您可以通过其自定义资源为这些组件启用和配置追踪。您可以使用 spec.template 属性添加追踪配置。

您可以使用 spec.tracing.type 属性指定追踪类型来启用追踪：

Kafka 客户端的追踪

还可以设置 Kafka producer 和使用者等客户端应用程序，以便监控事务。客户端配置了追踪配置集，并且初始化 tracer 供客户端应用使用。

Cruise Control 是一个开源系统，支持以下 Kafka 操作：

此操作可帮助运行更有效地使用代理 pod 的 Kafka 集群。

典型的集群可能会随着时间不均匀地加载。处理大量消息流量的分区可能无法在可用的代理中平均分布。要重新平衡集群，管理员必须监控代理上的负载，并将忙碌的分区手动分配给具有备用容量的代理。

Cruise Control 自动执行集群重新平衡过程。它基于 CPU、磁盘和网络负载（您可以批准或拒绝）为集群构建资源利用率 的工作负载模型。一组可配置的优化目标用于计算这些提议。

您可以在特定模式中生成优化方案。默认 full 模式在所有代理间重新平衡分区。您还可以使用 add-brokers 和 remove-brokers 模式来适应扩展集群或缩减时的更改。

当您批准一个优化建议时，Cruise Control 会将它应用到您的 Kafka 集群。您可以使用 KafkaRebalance 资源配置和生成优化建议。您可以使用注解配置资源，以便自动或手动批准优化提议。