在 OpenShift 中部署和管理 AMQ Streams

流程

1. 编辑 AMQ Streams 安装文件，以使用 Cluster Operator 将安装到的命名空间。

   例如，在此过程中，Cluster Operator 被安装到命名空间 my-cluster-operator-namespace 中。

   在 Linux 中，使用：

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   在 MacOS 中，使用：

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. 部署 Cluster Operator：

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

3. 检查部署的状态：

   oc get deployments -n my-cluster-operator-namespace

   输出显示部署名称和就绪度

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 显示就绪/预期的副本数。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 编辑 AMQ Streams 安装文件，以使用 Cluster Operator 将安装到的命名空间。

   例如，在此过程中，Cluster Operator 被安装到命名空间 my-cluster-operator-namespace 中。

   在 Linux 中，使用：

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   在 MacOS 中，使用：

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. 编辑 install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 文件，以添加 Cluster Operator 将监视到 STRIMZI_NAMESPACE 环境变量的所有命名空间的列表。

   例如，在此过程中，Cluster Operator 会监视 watched-namespace-1,watched-namespace-2,watched-namespace-3。

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.1
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: watched-namespace-1,watched-namespace-2,watched-namespace-3

3. 对于列出的每个命名空间，请安装 RoleBindings。

   在这个示例中，将这些命令的 watched-namespace 替换为在前一步中列出的命名空间，为 watched-namespace-1, watched-namespace-2, watched-namespace-3 重复这个操作：

   oc create -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/023-RoleBinding-strimzi-cluster-operator.yaml -n <watched_namespace>
   oc create -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n <watched_namespace>

4. 部署 Cluster Operator：

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. 检查部署的状态：

   oc get deployments -n my-cluster-operator-namespace

   输出显示部署名称和就绪度

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 显示就绪/预期的副本数。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 编辑 AMQ Streams 安装文件，以使用 Cluster Operator 将安装到的命名空间。

   例如，在此过程中，Cluster Operator 被安装到命名空间 my-cluster-operator-namespace 中。

   在 Linux 中，使用：

   sed -i 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

   在 MacOS 中，使用：

   sed -i '' 's/namespace: .*/namespace: my-cluster-operator-namespace/' install/cluster-operator/*RoleBinding*.yaml

2. 编辑 install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 文件，将 STRIMZI_NAMESPACE 环境变量的值设置为 sVirt。

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         # ...
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq-streams/strimzi-rhel8-operator:2.5.1
           imagePullPolicy: IfNotPresent
           env:
           - name: STRIMZI_NAMESPACE
             value: "*"
           # ...

3. 创建 ClusterRoleBindings，为 Cluster Operator 为所有命名空间赋予集群范围的访问权限。

   oc create clusterrolebinding strimzi-cluster-operator-namespaced --clusterrole=strimzi-cluster-operator-namespaced --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-watched --clusterrole=strimzi-cluster-operator-watched --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

4. 将 Cluster Operator 部署到 OpenShift 集群。

   oc create -f install/cluster-operator -n my-cluster-operator-namespace

5. 检查部署的状态：

   oc get deployments -n my-cluster-operator-namespace

   输出显示部署名称和就绪度

   NAME                      READY  UP-TO-DATE  AVAILABLE
   strimzi-cluster-operator  1/1    1           1

   READY 显示就绪/预期的副本数。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 创建并部署临时或持久集群。

   • 创建和部署临时集群：

     oc apply -f examples/kafka/kafka-ephemeral.yaml

   • 创建和部署持久集群：

     oc apply -f examples/kafka/kafka-persistent.yaml

2. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示 pod 名称和就绪状态

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-kafka-0          1/1     Running   0
   my-cluster-kafka-1          1/1     Running   0
   my-cluster-kafka-2          1/1     Running   0
   my-cluster-zookeeper-0      1/1     Running   0
   my-cluster-zookeeper-1      1/1     Running   0
   my-cluster-zookeeper-2      1/1     Running   0

   my-cluster 是 Kafka 集群的名称。

   从 0 开始的连续索引号标识每个 Kafka 和 ZooKeeper pod。

   使用默认部署，您可以创建一个 Entity Operator 集群、3 个 Kafka pod 和 3 ZooKeeper pod。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 从命令行启用 KafkaNodePools 功能门：

   oc set env deployment/strimzi-cluster-operator STRIMZI_FEATURE_GATES="+KafkaNodePools"

   或者，通过编辑 Cluster Operator Deployment 并更新 STRIMZI_FEATURE_GATES 环境变量：

   env
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

   这会更新 Cluster Operator。

   如果使用 KRaft 模式，还要启用 UseKRaft 功能门。

2. 创建节点池。

   • 使用三个代理的两个节点池部署 Kafka 集群和 ZooKeeper 集群：

     oc apply -f examples/kafka/nodepools/kafka.yaml

   • 使用使用 dual-role 节点的单一节点池，以 KRaft 模式部署 Kafka 集群：

     oc apply -f examples/kafka/nodepools/kafka-with-dual-role-kraft-nodes.yaml

   • 使用 KRaft 模式部署 Kafka 集群，对于代理和控制节点有独立的节点池，：

     oc apply -f examples/kafka/nodepools/kafka-with-kraft.yaml

3. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示节点池名称和就绪

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   my-cluster-pool-a-kafka-0   1/1     Running   0
   my-cluster-pool-a-kafka-1   1/1     Running   0
   my-cluster-pool-a-kafka-4   1/1     Running   0

   • my-cluster 是 Kafka 集群的名称。

   • pool-a 是节点池的名称。

     以 0 开始的连续索引号标识每个创建的 Kafka pod。如果使用 ZooKeeper，您还会看到 ZooKeeper pod。

     READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

     有关部署的信息也会显示在 KafkaNodePool 资源的状态中，包括池中节点的 ID 列表。

     注意

     节点 ID 按顺序分配自集群中所有节点池中的 0 （零）。这意味着节点 ID 可能无法在特定节点池中按顺序运行。如果集群中节点 ID 序列出现差距，则会为要添加的下一个节点分配一个填充空白的 ID。缩减时，池中具有最高节点 ID 的节点会被删除。

流程

1. 编辑 Kafka 资源的 entityOperator 属性，使其包含 topicOperator ：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 使用 EntityTopicOperatorSpec schema reference 中所述的属性配置 Topic Operator spec。

   如果您希望所有属性使用默认值，请使用空对象({})。

3. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

4. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示 pod 名称和就绪度

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster 是 Kafka 集群的名称。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 编辑 Kafka 资源的 entityOperator 属性，使其包含 userOperator ：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     #...
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 使用 EntityUserOperatorSpec schema reference 中所述的属性配置 User Operator spec。

   如果您希望所有属性使用默认值，请使用空对象({})。

3. 创建或更新资源：

   oc apply -f <kafka_configuration_file>

4. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示 pod 名称和就绪度

   NAME                        READY   STATUS    RESTARTS
   my-cluster-entity-operator  3/3     Running   0
   # ...

   my-cluster 是 Kafka 集群的名称。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 部署 Kafka 连接到 OpenShift 集群。使用 examples/connect/kafka-connect.yaml 文件来部署 Kafka Connect。

   oc apply -f examples/connect/kafka-connect.yaml

2. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示部署名称和就绪度

   NAME                                 READY  STATUS   RESTARTS
   my-connect-cluster-connect-<pod_id>  1/1    Running  0

   my-connect-cluster 是 Kafka Connect 集群的名称。

   pod ID 标识创建的每个 pod。

   使用默认部署，您可以创建一个 Kafka Connect pod。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 将 Kafka MirrorMaker 部署到 OpenShift 集群中：

   对于 MirrorMaker：

   oc apply -f examples/mirror-maker/kafka-mirror-maker.yaml

   对于 MirrorMaker 2：

   oc apply -f examples/mirror-maker/kafka-mirror-maker-2.yaml

2. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示部署名称和就绪度

   NAME                                    READY  STATUS   RESTARTS
   my-mirror-maker-mirror-maker-<pod_id>   1/1    Running  1
   my-mm2-cluster-mirrormaker2-<pod_id>    1/1    Running  1

   my-mirror-maker 是 Kafka MirrorMaker 集群的名称。my-mm2-cluster 是 Kafka MirrorMaker 2 集群的名称。

   pod ID 标识创建的每个 pod。

   使用默认部署，您要安装单个 MirrorMaker 或 MirrorMaker 2 pod。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 将 Kafka Bridge 部署到 OpenShift 集群：

   oc apply -f examples/bridge/kafka-bridge.yaml

2. 检查部署的状态：

   oc get pods -n <my_cluster_operator_namespace>

   输出显示部署名称和就绪度

   NAME                       READY  STATUS   RESTARTS
   my-bridge-bridge-<pod_id>  1/1    Running  0

   my-bridge 是 Kafka Bridge 集群的名称。

   pod ID 标识创建的每个 pod。

   使用默认部署，您会安装单个 Kafka Bridge pod。

   READY 显示就绪/预期的副本数。当 STATUS 显示为 Running 时，部署成功。

流程

1. 列出 OpenShift 集群中的 pod 的名称：

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-<pod_id>

2. 在端口 8080 上连接到 Kafka Bridge pod：

   oc port-forward pod/my-bridge-bridge-<pod_id> 8080:8080 &

   注意

   如果本地计算机上的端口 8080 已在使用中，请使用其他 HTTP 端口，如 8008。

流程

1. 编辑 install/topic-operator/05-Deployment-strimzi-topic-operator.yaml 独立部署文件中的 env 属性。

   独立主题 Operator 部署配置示例

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-topic-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-topic-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_RESOURCE_LABELS 3
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_ZOOKEEPER_CONNECT 4
                 value: my-cluster-zookeeper-client:2181
               - name: STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 5
                 value: "18000"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS 7
                 value: "6"
               - name: STRIMZI_LOG_LEVEL 8
                 value: INFO
               - name: STRIMZI_TLS_ENABLED 9
                 value: "false"
               - name: STRIMZI_JAVA_OPTS 10
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 11
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_PUBLIC_CA 12
                 value: "false"
               - name: STRIMZI_TLS_AUTH_ENABLED 13
                 value: "false"
               - name: STRIMZI_SASL_ENABLED 14
                 value: "false"
               - name: STRIMZI_SASL_USERNAME 15
                 value: "admin"
               - name: STRIMZI_SASL_PASSWORD 16
                 value: "password"
               - name: STRIMZI_SASL_MECHANISM 17
                 value: "scram-sha-512"
               - name: STRIMZI_SECURITY_PROTOCOL 18
                 value: "SSL"

   1

   主题 Operator 的 OpenShift 命名空间，以监视 KafkaTopic 资源。指定 Kafka 集群的命名空间。

   2

   bootstrap 代理地址的主机和端口对，以发现并连接到 Kafka 集群中的所有代理。在服务器停机时，使用以逗号分隔的列表指定两个或多个代理地址。

   3

   用于标识主题 Operator 管理的 KafkaTopic 资源的标签。这不一定是 Kafka 集群的名称。它可以是分配给 KafkaTopic 资源的标签。如果您部署多个主题 Operator，则标签必须是唯一的。也就是说，Operator 无法管理同一资源。

   4

   (ZooKeeper)用于连接到 ZooKeeper 集群的地址的主机和端口对。这必须是 Kafka 集群正在使用的同一 ZooKeeper 集群。

   5

   (ZooKeeper) ZooKeeper 会话超时，以毫秒为单位。默认值为 18000 (18 秒)。

   6

   定期协调之间的间隔，以毫秒为单位。默认值为 120000 (2 分钟)。

   7

   从 Kafka 获取主题元数据的尝试次数。每次尝试之间的时间定义为指数 backoff。由于分区或副本数，当主题创建需要更长的时间时，请考虑增加这个值。默认值为 6 个尝试。

   8

   打印日志记录消息的级别。您可以将级别设置为 ERROR、WARNING、INFO、DEBUG 或 TRACE。

   9

   启用 TLS 支持与 Kafka 代理的加密通信。

   10

   （可选）运行主题 Operator 的 JVM 使用的 Java 选项。

   11

   （可选）为主题 Operator 设置的调试(-D)选项。

   12

   （可选）如果通过 STRIMZI_TLS_ENABLED 启用 TLS，请跳过信任存储证书的生成。如果启用了此环境变量，代理必须使用公共可信证书颁发机构作为其 TLS 证书。默认值为 false。

   13

   （可选）为 mTLS 身份验证生成密钥存储证书。把它设置为 false 可禁用 mTLS 到 Kafka 代理的客户端身份验证。默认值是 true。

   14

   （可选）在连接到 Kafka 代理时启用对客户端身份验证的 SASL 支持。默认值为 false。

   15

   （可选）客户端身份验证的 SASL 用户名。只有通过 STRIMZI_SASL_ENABLED 启用 SASL 时才强制。

   16

   （可选）客户端身份验证的 SASL 密码。只有通过 STRIMZI_SASL_ENABLED 启用 SASL 时才强制。

   17

   （可选）客户端身份验证的 SASL 机制。只有通过 STRIMZI_SASL_ENABLED 启用 SASL 时才强制。您可以将值设为 plain,scram-sha-256, 或 scram-sha-512。

   18

   （可选）用于与 Kafka 代理通信的安全协议。默认值为 "PLAINTEXT"。您可以将值设为 PLAINTEXT、SSL、SASL_PLAINTEXT 或 SASL_SSL。

2. 如果要连接到使用来自公共证书颁发机构证书的 Kafka 代理，请将 STRIMZI_PUBLIC_CA 设置为 true。将此属性设置为 true，例如，如果您使用 Amazon AWS MSK 服务。

3. 如果您使用 STRIMZI_TLS_ENABLED 环境变量启用了 mTLS，请指定用于验证与 Kafka 集群的连接的密钥存储和信任存储。

   mTLS 配置示例

   # ....
   env:
     - name: STRIMZI_TRUSTSTORE_LOCATION 1
       value: "/path/to/truststore.p12"
     - name: STRIMZI_TRUSTSTORE_PASSWORD 2
       value: "TRUSTSTORE-PASSWORD"
     - name: STRIMZI_KEYSTORE_LOCATION 3
       value: "/path/to/keystore.p12"
     - name: STRIMZI_KEYSTORE_PASSWORD 4
       value: "KEYSTORE-PASSWORD"
   # ...

   1

   truststore 包含用于为 Kafka 和 ZooKeeper 服务器证书签名的证书颁发机构的公钥。

   2

   用于访问 truststore 的密码。

   3

   密钥存储包含 mTLS 身份验证的私钥。

   4

   用于访问密钥存储的密码。

4. 部署主题 Operator。

   oc create -f install/topic-operator

5. 检查部署的状态：

   oc get deployments

   输出显示部署名称和就绪度

   NAME                    READY  UP-TO-DATE  AVAILABLE
   strimzi-topic-operator  1/1    1           1

   READY 显示就绪/预期的副本数。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 编辑 install/user-operator/05-Deployment-strimzi-user-operator.yaml 独立部署文件中的以下 env 属性。

   独立用户 Operator 部署配置示例

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: strimzi-user-operator
     labels:
       app: strimzi
   spec:
     # ...
     template:
       # ...
       spec:
         # ...
         containers:
           - name: strimzi-user-operator
             # ...
             env:
               - name: STRIMZI_NAMESPACE 1
                 valueFrom:
                   fieldRef:
                     fieldPath: metadata.namespace
               - name: STRIMZI_KAFKA_BOOTSTRAP_SERVERS 2
                 value: my-kafka-bootstrap-address:9092
               - name: STRIMZI_CA_CERT_NAME 3
                 value: my-cluster-clients-ca-cert
               - name: STRIMZI_CA_KEY_NAME 4
                 value: my-cluster-clients-ca
               - name: STRIMZI_LABELS 5
                 value: "strimzi.io/cluster=my-cluster"
               - name: STRIMZI_FULL_RECONCILIATION_INTERVAL_MS 6
                 value: "120000"
               - name: STRIMZI_WORK_QUEUE_SIZE 7
                 value: 10000
               - name: STRIMZI_CONTROLLER_THREAD_POOL_SIZE 8
                 value: 10
               - name: STRIMZI_USER_OPERATIONS_THREAD_POOL_SIZE 9
                 value: 4
               - name: STRIMZI_LOG_LEVEL 10
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 11
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 12
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 13
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 14
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 15
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 16
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 17
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 18
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 19
                 value: |
                   default.api.timeout.ms=120000
                   request.timeout.ms=60000

   1

   User Operator 的 OpenShift 命名空间，用于监视 KafkaUser 资源。只能指定一个命名空间。

   2

   bootstrap 代理地址的主机和端口对，以发现并连接到 Kafka 集群中的所有代理。在服务器停机时，使用以逗号分隔的列表指定两个或多个代理地址。

   3

   包含为 mTLS 身份验证签名的 CA (certificate 授权)的公钥(ca.crt)值的 OpenShift Secret。

   4

   包含 CA 的私钥(ca.key)值的 OpenShift Secret，用于为 mTLS 身份验证为新用户证书签名。

   5

   用于标识由 User Operator 管理的 KafkaUser 资源的标签。这不一定是 Kafka 集群的名称。它可以是分配给 KafkaUser 资源的标签。如果部署多个 User Operator，则标签必须是唯一的。也就是说，Operator 无法管理同一资源。

   6

   定期协调之间的间隔，以毫秒为单位。默认值为 120000 (2 分钟)。

   7

   控制器事件队列的大小。队列的大小应至少与您希望 User Operator 操作的最大用户量一样。默认值为 1024。

   8

   用于协调用户的 worker 池大小。较大的池可能需要更多资源，但它也会处理更多 KafkaUser 资源，默认值为 50。

   9

   Kafka Admin API 和 OpenShift 操作的 worker 池大小。较大的池可能需要更多资源，但它也会处理更多的 KafkaUser 资源。默认为 4。

   10

   打印日志记录消息的级别。您可以将级别设置为 ERROR、WARNING、INFO、DEBUG 或 TRACE。

   11

   启用垃圾回收(GC)日志记录。默认值是 true。

   12

   CA 的有效性周期。默认值为 365 天。

   13

   CA 的续订周期。续订周期从当前证书的到期日期后衡量。默认值为 30 天，以在旧证书过期前启动证书续订。

   14

   （可选）运行用户 Operator 的 JVM 使用的 Java 选项

   15

   （可选）为 User Operator 设置的调试(-D)选项

   16

   （可选）由 User Operator 创建的 OpenShift secret 名称的前缀。

   17

   （可选）指示 Kafka 集群是否支持使用 Kafka Admin API 管理授权 ACL 规则。当设置为 false 时，User Operator 将拒绝具有 simple 授权 ACL 规则的所有资源。这有助于避免 Kafka 集群日志中不必要的异常。默认值是 true。

   18

   （可选） Semi-colon 分隔 Cron Expressions 列表，用于定义维护时间窗，其中过期用户证书。

   19

   （可选）配置 User Operator 以属性格式使用的 Kafka Admin 客户端的配置选项。

2. 如果您使用 mTLS 连接到 Kafka 集群，请指定用于验证连接的 secret。否则，进入下一步。

   mTLS 配置示例

   # ....
   env:
     - name: STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 1
       value: my-cluster-cluster-ca-cert
     - name: STRIMZI_EO_KEY_SECRET_NAME 2
       value: my-cluster-entity-operator-certs
   # ..."

   1

   包含为 Kafka 代理证书的 CA 的公钥(ca.crt)值的 OpenShift Secret。

   2

   包含用于针对 Kafka 集群的 mTLS 身份验证的证书公钥(entity-operator.crt)和私钥(entity-operator.key)的 OpenShift Secret。

3. 部署 User Operator。

   oc create -f install/user-operator

4. 检查部署的状态：

   oc get deployments

   输出显示部署名称和就绪度

   NAME                   READY  UP-TO-DATE  AVAILABLE
   strimzi-user-operator  1/1    1           1

   READY 显示就绪/预期的副本数。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 为 Kafka 资源的 config 添加插件属性。

   此示例配置中显示了插件属性。

   Kafka 静态配额插件配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
       config:
         client.quota.callback.class: io.strimzi.kafka.quotas.StaticQuotaCallback 1
         client.quota.callback.static.produce: 1000000 2
         client.quota.callback.static.fetch: 1000000 3
         client.quota.callback.static.storage.soft: 400000000000 4
         client.quota.callback.static.storage.hard: 500000000000 5
         client.quota.callback.static.storage.check-interval: 5 6

   1

   加载 Kafka Static Quota 插件。

   2

   设置制作者字节阈值。本示例中为 1 Mbps.

   3

   设置消费者字节阈值。本示例中为 1 Mbps.

   4

   为存储设置较低软限制。在本示例中为 400 GB。

   5

   为存储设置更高的硬限制。本例中为 500 GB。

   6

   设置存储检查间隔（以秒为单位）。本例中有 5 秒。您可以将其设置为 0 来禁用检查。

2. 更新资源。

   oc apply -f <kafka_configuration_file>

流程

1. 使用 ID 为节点池添加在扩展或缩减时要使用的 ID，如下例所示。

   用于向上扩展的 ID 会被分配给节点 池池 ：

   为扩展分配 ID

   oc annotate kafkanodepool pool-a strimzi.io/next-node-ids="[0,1,2,10-20,30]"

   将节点添加到 池 时使用此范围内的最低可用 ID。

   用于缩减的 ID 分配给节点池 pool-b ：

   为缩减分配 ID

   oc annotate kafkanodepool pool-b strimzi.io/remove-node-ids="[60-50,9,8,7]"

   缩减 pool-b 时会删除此范围内的最高可用 ID。

2. 现在，您可以扩展节点池。

   如需更多信息，请参阅以下：

   • 第 8.3.2 节 “（预览）将节点添加到节点池中”
   • 第 8.3.3 节 “（预览）从节点池中删除节点”
   • 第 8.3.4 节 “（预览）在节点池之间移动节点”

   在协调时，如果注解被错误格式化，则会提供一个警告。

流程

1. 在节点池中创建新节点。

   例如，节点池 pool-a 有三个副本。我们通过增加副本数量来添加节点：

   oc scale kafkanodepool pool-a --replicas=4

2. 检查部署的状态，并等待节点池中的 pod 创建，并且状态为 READY。

   oc get pods -n <my_cluster_operator_namespace>

   输出显示节点池中的四个 Kafka 节点

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-a-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-2  1/1    Running  0
   my-cluster-pool-a-kafka-3  1/1    Running  0

3. 在增加节点池中的节点数后重新分配分区。

   在扩展节点池后，您可以使用 Cruise Control add-brokers 模式将分区副本从现有代理移到新添加的代理中。

流程

1. 在减少节点池中的节点数前重新分配分区。

   在缩减节点池前，您可以使用 Cruise Control remove-brokers 模式将分区副本移出要删除的代理。

2. 在重新分配过程完成后，删除的节点没有实时分区，从而减少节点池中的 Kafka 节点数量。

   例如，节点池 pool-a 有四个副本。我们通过减少副本数来删除节点：

   oc scale kafkanodepool pool-a --replicas=3

   输出显示节点池中的三个 Kafka 节点

   NAME                        READY   STATUS    RESTARTS
   my-cluster-pool-b-kafka-0   1/1     Running   0
   my-cluster-pool-b-kafka-1   1/1     Running   0
   my-cluster-pool-b-kafka-2   1/1     Running   0

流程

1. 在目标节点池中创建新节点。

   例如，节点池 pool-a 有三个副本。我们通过增加副本数量来添加节点：

   oc scale kafkanodepool pool-a --replicas=4

2. 检查部署的状态，并等待节点池中的 pod 创建，并且状态为 READY。

   oc get pods -n <my_cluster_operator_namespace>

   输出显示目标节点池中的四个 Kafka 节点

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-a-kafka-0  1/1    Running  0
   my-cluster-pool-a-kafka-1  1/1    Running  0
   my-cluster-pool-a-kafka-4  1/1    Running  0
   my-cluster-pool-a-kafka-5  1/1    Running  0

   节点 ID 在创建时附加到节点的名称中。我们添加节点 my-cluster-pool-a-kafka-5，其节点 ID 为 5。

3. 将分区从旧节点重新分配给新节点。

   在缩减源节点池前，您可以使用 Cruise Control remove-brokers 模式将分区副本移出要删除的代理。

4. 重新分配过程完成后，减少源节点池中的 Kafka 节点数量。

   例如，节点池 pool-b 具有四个副本。我们通过减少副本数来删除节点：

   oc scale kafkanodepool pool-b --replicas=3

   池中具有最高 ID 的节点已被删除。

   输出显示源节点池中的三个 Kafka 节点

   NAME                       READY  STATUS   RESTARTS
   my-cluster-pool-b-kafka-2  1/1    Running  0
   my-cluster-pool-b-kafka-3  1/1    Running  0
   my-cluster-pool-b-kafka-6  1/1    Running  0

流程

1. 创建新的 KafkaNodePool 资源。

   1. 将资源命名为 kafka。
   2. 将 strimzi.io/cluster 标签指向现有的 Kafka 资源。
   3. 设置副本数和存储配置以匹配您当前的 Kafka 集群。
   4. 将角色设置为 代理。

   迁移 Kafka 集群的节点池配置示例

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

2. 应用 KafkaNodePool 资源：

   oc apply -f <node_pool_configuration_file>

   通过应用此资源，您可以将 Kafka 切换到使用节点池。

   没有更改或滚动更新和资源与之前的资源相同。

3. 更新 Cluster Operator 配置中的 STRIMZI_FEATURE_GATES 环境变量，使其包含 +KafkaNodePools。

   env:
     - name: STRIMZI_FEATURE_GATES
       value: +KafkaNodePools

4. 使用 strimzi.io/node-pools: enabled 注解，在 Kafka 资源中启用 KafkaNodePools 功能门。

   使用 ZooKeeper 在集群中节点池的配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     annotations:
       strimzi.io/node-pools: enabled
   spec:
     kafka:
       version: 3.5.0
       replicas: 3
     # ...
     storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false

5. 应用 Kafka 资源：

   oc apply -f <kafka_configuration_file>

流程

1. 要在 Cluster Operator 中添加代理环境变量，请更新其部署配置 (install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml)。

   Cluster Operator 的代理配置示例

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "HTTP_PROXY"
             value: "http://proxy.com" 1
           - name: "HTTPS_PROXY"
             value: "https://proxy.com" 2
           - name: "NO_PROXY"
             value: "internal.com, other.domain.com" 3
     # ...

   1

   代理服务器的地址。

   2

   代理服务器的安全地址。

   3

   作为代理服务器的例外访问的服务器的地址。URL 是用逗号分开的。

   或者，直接编辑 部署 ：

   oc edit deployment strimzi-cluster-operator

2. 如果您更新了 YAML 文件而不是直接编辑 Deployment，请应用更改：

   oc create -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

流程

1. 要在 Cluster Operator 中禁用 FIPS 模式，更新其部署配置install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml) 并增加 FIPS_MODE 环境变量。

   Cluster Operator 的 FIPS 配置示例

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         serviceAccountName: strimzi-cluster-operator
         containers:
           # ...
           env:
           # ...
           - name: "FIPS_MODE"
             value: "disabled" 1
     # ...

   1

   禁用 FIPS 模式。

   或者，直接编辑 部署 ：

   oc edit deployment strimzi-cluster-operator

2. 如果您更新了 YAML 文件而不是直接编辑 Deployment，请应用更改：

   oc apply -f install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml

流程

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

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

   属性	名称
   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: "*"

2. 创建或更新资源。

   oc apply -f KAFKA-USER-CONFIG-FILE

流程

1. 配置两个 Kafka 资源，一个用于保护源 Kafka 集群，另一个用于保护目标 Kafka 集群。

   您可以为身份验证和启用授权添加监听程序配置。

   在本例中，为带有 TLS 加密和 mTLS 身份验证的 Kafka 集群配置了内部监听程序。启用 Kafka 简单 授权。

   使用 TLS 加密和 mTLS 身份验证的源 Kafka 集群配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-source-cluster
   spec:
     kafka:
       version: 3.5.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.5"
       storage:
         type: jbod
         volumes:
         - id: 0
           type: persistent-claim
           size: 100Gi
           deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

   使用 TLS 加密和 mTLS 身份验证的目标 Kafka 集群配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-target-cluster
   spec:
     kafka:
       version: 3.5.0
       replicas: 1
       listeners:
         - name: tls
           port: 9093
           type: internal
           tls: true
           authentication:
             type: tls
       authorization:
         type: simple
       config:
         offsets.topic.replication.factor: 1
         transaction.state.log.replication.factor: 1
         transaction.state.log.min.isr: 1
         default.replication.factor: 1
         min.insync.replicas: 1
         inter.broker.protocol.version: "3.5"
       storage:
         type: jbod
         volumes:
           - id: 0
             type: persistent-claim
             size: 100Gi
             deleteClaim: false
     zookeeper:
       replicas: 1
       storage:
         type: persistent-claim
         size: 100Gi
         deleteClaim: false
     entityOperator:
       topicOperator: {}
       userOperator: {}

2. 在单独的命名空间中创建或更新 Kafka 资源。

   oc apply -f <kafka_configuration_file> -n <namespace>

   Cluster Operator 创建监听程序并设置集群和客户端证书颁发机构(CA)证书，以便在 Kafka 集群中启用身份验证。

   证书在 secret < cluster_name> -cluster-ca-cert 中创建。

3. 配置两个 KafkaUser 资源，一个用于源 Kafka 集群的用户，另一个用于目标 Kafka 集群的用户。

   1. 配置与对应的源和目标 Kafka 集群相同的身份验证和授权类型。例如，如果您在源 Kafka 集群的 Kafka 配置中使用了 tls 验证和 simple 授权类型，请在 KafkaUser 配置中使用相同的。

   2. 配置 MirrorMaker 2 所需的 ACL，以允许对源和目标 Kafka 集群执行操作。

      内部 MirrorMaker 连接器和底层 Kafka Connect 框架使用 ACL。

   mTLS 验证的源用户配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-source-user
     labels:
       strimzi.io/cluster: my-source-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # MirrorSourceConnector
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Create
             - DescribeConfigs
             - Read
             - Write
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - DescribeConfigs
             - Read
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource: # Needed for every group for which offsets are synced
             type: group
             name: "*"
           operations:
             - Describe
         - resource: # Not needed if offset-syncs.topic.location=target
             type: topic
             name: mm2-offset-syncs.my-target-cluster.internal
           operations:
             - Read

   mTLS 验证的目标用户配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaUser
   metadata:
     name: my-target-user
     labels:
       strimzi.io/cluster: my-target-cluster
   spec:
     authentication:
       type: tls
     authorization:
       type: simple
       acls:
         # Underlying Kafka Connect internal topics to store configuration, offsets, or status
         - resource:
             type: group
             name: mirrormaker2-cluster
           operations:
             - Read
         - resource:
             type: topic
             name: mirrormaker2-cluster-configs
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-status
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         - resource:
             type: topic
             name: mirrormaker2-cluster-offsets
           operations:
             - Create
             - Describe
             - DescribeConfigs
             - Read
             - Write
         # MirrorSourceConnector
         - resource: # Needed for every topic which is mirrored
             type: topic
             name: "*"
           operations:
             - Create
             - Alter
             - AlterConfigs
             - Write
         # MirrorCheckpointConnector
         - resource:
             type: cluster
           operations:
             - Describe
         - resource:
             type: topic
             name: my-source-cluster.checkpoints.internal
           operations:
             - Create
             - Describe
             - Read
             - Write
         - resource: # Needed for every group for which the offset is synced
             type: group
             name: "*"
           operations:
             - Read
             - Describe
         # MirrorHeartbeatConnector
         - resource:
             type: topic
             name: heartbeats
           operations:
             - Create
             - Describe
             - Write

   注意

   您可以通过将 type 设置为 tls-external 来使用 User Operator 外部发布的证书。如需更多信息，请参阅 KafkaUserSpec 模式参考。

4. 在您为源和目标 Kafka 集群创建的每个命名空间中创建或更新 KafkaUser 资源。

   oc apply -f <kafka_user_configuration_file> -n <namespace>

   User Operator 根据所选的验证类型创建代表客户端(MirrorMaker)的用户，以及用于客户端身份验证的安全凭证。

   User Operator 创建一个名称与 KafkaUser 资源相同的新 secret。secret 包含 mTLS 验证的私钥和公钥。公钥包含在用户证书中，该证书由客户端 CA 签名。

5. 使用身份验证详情配置 KafkaMirrorMaker2 资源，以连接到源和目标 Kafka 集群。

   带有 TLS 加密和 mTLS 身份验证的 MirrorMaker 2 配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mirror-maker-2
   spec:
     version: 3.5.0
     replicas: 1
     connectCluster: "my-target-cluster"
     clusters:
       - alias: "my-source-cluster"
         bootstrapServers: my-source-cluster-kafka-bootstrap:9093
         tls: 1
           trustedCertificates:
             - secretName: my-source-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 2
           type: tls
           certificateAndKey:
             secretName: my-source-user
             certificate: user.crt
             key: user.key
       - alias: "my-target-cluster"
         bootstrapServers: my-target-cluster-kafka-bootstrap:9093
         tls: 3
           trustedCertificates:
             - secretName: my-target-cluster-cluster-ca-cert
               certificate: ca.crt
         authentication: 4
           type: tls
           certificateAndKey:
             secretName: my-target-user
             certificate: user.crt
             key: user.key
         config:
           # -1 means it will use the default replication factor configured in the broker
           config.storage.replication.factor: -1
           offset.storage.replication.factor: -1
           status.storage.replication.factor: -1
     mirrors:
       - sourceCluster: "my-source-cluster"
         targetCluster: "my-target-cluster"
         sourceConnector:
           config:
             replication.factor: 1
             offset-syncs.topic.replication.factor: 1
             sync.topic.acls.enabled: "false"
         heartbeatConnector:
           config:
             heartbeats.topic.replication.factor: 1
         checkpointConnector:
           config:
             checkpoints.topic.replication.factor: 1
             sync.group.offsets.enabled: "true"
         topicsPattern: "topic1|topic2|topic3"
         groupsPattern: "group1|group2|group3"

   1

   源 Kafka 集群的 TLS 证书。如果它们位于单独的命名空间中，请将集群 secret 从 Kafka 集群的命名空间中复制。

   2

   使用 TLS 机制访问源 Kafka 集群的用户身份验证。

   3

   目标 Kafka 集群的 TLS 证书。

   4

   访问目标 Kafka 集群的用户身份验证。

6. 在与目标 Kafka 集群相同的命名空间中创建或更新 KafkaMirrorMaker2 资源。

   oc apply -f <mirrormaker2_configuration_file> -n <namespace_of_target_cluster>