步骤

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 apply -f install/cluster-operator -n my-cluster-operator-namespace

3. 验证 Cluster Operator 是否已成功部署：

   oc get deployments

步骤

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/amq7/amq-streams-rhel7-operator:1.6.7
           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 apply -f install/cluster-operator/020-RoleBinding-strimzi-cluster-operator.yaml -n watched-namespace
   oc apply -f install/cluster-operator/031-RoleBinding-strimzi-cluster-operator-entity-operator-delegation.yaml -n watched-namespace
   oc apply -f install/cluster-operator/032-RoleBinding-strimzi-cluster-operator-topic-operator-delegation.yaml -n watched-namespace

4. 部署 Cluster Operator：

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

5. 验证 Cluster Operator 是否已成功部署：

   oc get deployments

步骤

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 环境变量的值设置为 *。

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

3. 创建 ClusterRoleBinding， 将所有命名空间的集群范围访问权限授予 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-entity-operator-delegation --clusterrole=strimzi-entity-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator
   oc create clusterrolebinding strimzi-cluster-operator-topic-operator-delegation --clusterrole=strimzi-topic-operator --serviceaccount my-cluster-operator-namespace:strimzi-cluster-operator

   将 my-cluster-operator-namespace 替换为您要将 Cluster Operator 安装到的命名空间。

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

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

5. 验证 Cluster Operator 是否已成功部署：

   oc get deployments

步骤

1. 在 OpenShift 4 Web 控制台中，点 Operators > OperatorHub。

2. 在流和 消息传递 类别中搜索或浏览 AMQ Streams Operator。

   
3. 单击 AMQ Streams 标题，然后在右侧的侧边栏中点 Install。

4. 在 Create Operator Subscription 屏幕上，从以下安装和更新选项中进行选择：

   • 安装模式 ：选择将 AMQ Streams Operator 安装到集群中的所有（项目）命名空间（默认选项）或特定（项目）命名空间。最好使用命名空间来分隔函数。建议您将特定命名空间专用于 Kafka 集群和其他 AMQ Streams 组件。
   • 批准策略 ：默认情况下，Operator Lifecycle Manager(OLM)自动将 AMQ Streams Operator 升级到最新的 AMQ Streams 版本。（可选）如果您想要手动批准以后的升级，请选择 Manual。如需更多信息，请参阅 OpenShift 文档中的 Operator 指南。

5. 点 Subscribe; AMQ Streams Operator 已安装到 OpenShift 集群。

   AMQ Streams Operator 将 Cluster Operator、CRD 和基于角色的访问控制(RBAC)资源部署到所选命名空间或所有命名空间中。

6. 在 Installed Operators 屏幕上，检查安装的进度。当 AMQ Streams Operator 的状态变为 InstallSucceeded 时，就可以使用它。

   

步骤

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

   对于开发或测试，您可能更喜欢使用临时集群。您可以随时使用持久性存储。

   • 创建和部署 临时 集群：

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

   • 创建和部署 持久 集群：

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

2. 验证 Kafka 集群是否已成功部署：

   oc get deployments

步骤

1. 编辑 Kafka 资源的 principal Operator 属性，使其包含 topicOperator:

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

2. 使用 EntityTopicOperatorSpec 模式参考 中所述的属性配置 Topic Operator spec。

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

3. 创建或更新资源：

   使用 oc apply:

   oc apply -f <your-file>

步骤

1. 编辑 Kafka 资源的 principal Operator 属性，使其包含 userOperator:

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

2. 使用 OpenShift 指南的使用 AMQ Streams 中的 EntityUserOperatorSpec schema 参考 中描述的属性配置 User Operator spec。

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

3. 创建或更新资源：

   oc apply -f <your-file>

步骤

1. 通过设置，编辑 install/topic-operator/05-Deployment-strimzi-topic-operator .yaml 文件中的 Deployment.spec.template.spec.containers[0]. env 属性：

   1. STRIMZI_KAFKA_BOOTSTRAP_SERVERS，用于列出 Kafka 集群中的 bootstrap 代理，以逗号分隔 的主机名 ：\{0\}端口 对列表形式指定。
   2. 用于列出 ZooKeeper 节点的 STRIMZI_ZOOKEEPER_CONNECT 节点，以逗号分隔 的主机名 ：\{0\}端口 对列表形式提供。这应该与 Kafka 集群使用的 ZooKeeper 集群相同。
   3. STRIMZI_NAMESPACE 到您希望操作员监视 KafkaTopic 资源的 OpenShift 命名空间。
   4. STRIMZI_RESOURCE_LABELS 到标签选择器，用于标识由操作器管理的 KafkaTopic 资源。
   5. STRIMZI_FULL_RECONCILIATION_INTERVAL_MS，以毫秒为单位指定定期协调之间的间隔。
   6. STRIMZI_TOPIC_METADATA_MAX_ATTEMPTS，以指定从 Kafka 获取主题元数据的尝试次数。每次尝试之间的时间都定义为指数回退。考虑在因为分区或副本数量而创建主题需要更多时间时增加这个值。默认 6.
   7. STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 到 ZooKeeper 会话超时，以毫秒为单位。例如，10000。Default 20000（ 20 秒）。
   8. STRIMZI_TOPICS_PATH 到 Topic Operator 存储其元数据的 Zookeeper 节点路径。默认 /strimzi/topics.
   9. STRIMZI_TLS_ENABLED 启用 TLS 支持以加密与 Kafka 代理的通信。默认为 true。
   10. STRIMZI_TRUSTSTORE_LOCATION，指向包含用于启用基于 TLS 的通信的证书的信任存储的路径。只有在通过 STRIMZI_TLS_ENABLED 启用 TLS 时才必需。
   11. STRIMZI_TRUSTSTORE_PASSWORD 到用于访问 STRIMZI_TRUSTSTORE_LOCATION 定义的信任存储的密码。只有在通过 STRIMZI_TLS_ENABLED 启用 TLS 时才必需。
   12. STRIMZI_KEYSTORE_LOCATION 到含有私钥的密钥存储的路径，以启用基于 TLS 的通信。只有在通过 STRIMZI_TLS_ENABLED 启用 TLS 时才必需。
   13. STRIMZI_KEYSTORE_PASSWORD 到用于访问 STRIMZI_KEYSTORE_LOCATION 定义的密钥存储的密码.只有在通过 STRIMZI_TLS_ENABLED 启用 TLS 时才必需。
   14. STRIMZI_LOG_LEVEL 到打印日志记录消息的级别。该值可设置为：ERROR、WARNING、INFO、DEBUG 和 TRACE.默认 INFO.
   15. STRIMZI_JAVA_OPTS （可选），用于运行主题 Operator 的 JVM。例如 -Xmx=512M -Xms=256M。
   16. STRIMZI_JAVA_SYSTEM_PROPERTIES （可选） 列出设置为 Topic Operator 的 -D 选项。An example is -Djavax.net.debug=verbose -DpropertyName=value.

2. 部署主题 Operator:

   oc apply -f install/topic-operator

3. 验证 Topic Operator 是否已成功部署：

   oc describe deployment strimzi-topic-operator

   当 Replicas: 条目显示 1 可用 时，Tpic Operator 会被部署。

   注意

   如果您与 OpenShift 集群的连接较慢并且之前没有下载镜像，则部署可能会遇到延迟。

步骤

1. 通过设置，编辑 install/user-operator/05-Deployment-strimzi-user-operator .yaml 文件中的以下 Deployment.spec.template.spec.containers[0]. env 属性：

   1. STRIMZI_KAFKA_BOOTSTRAP_SERVERS 来列出 Kafka 代理，以逗号分隔 的主机名 ：\{0\}端口 对列表形式指定。
   2. 用于列出 ZooKeeper 节点的 STRIMZI_ZOOKEEPER_CONNECT 节点，以逗号分隔 的主机名 ：\{0\}端口 对列表形式提供。这必须与您的 Kafka 集群使用的 ZooKeeper 集群相同。不支持使用 TLS 加密连接到 ZooKeeper 节点。
   3. STRIMZI_NAMESPACE 到 OpenShift 命名空间，您希望操作员在该命名空间中监视 KafkaUser 资源。
   4. STRIMZI_LABELS 到用于标识由操作器管理的 KafkaUser 资源标签选择器。
   5. STRIMZI_FULL_RECONCILIATION_INTERVAL_MS，以毫秒为单位指定定期协调之间的间隔。
   6. STRIMZI_ZOOKEEPER_SESSION_TIMEOUT_MS 到 ZooKeeper 会话超时，以毫秒为单位。例如，10000。Default 20000（ 20 秒）。
   7. STRIMZI_CA_CERT_NAME 以指向包含证书颁发机构公钥的 OpenShift Secret，用于为 TLS 客户端身份验证签名新用户证书。Secret 必须包含密钥 ca.crt 下的证书颁发机构的公钥。
   8. STRIMZI_CA_KEY_NAME 以指向包含证书颁发机构私钥的 OpenShift Secret，用于为 TLS 客户端身份验证签名新用户证书。Secret 必须包含密钥 ca.key 下的证书颁发机构的私钥。
   9. STRIMZI_CLUSTER_CA_CERT_SECRET_NAME 以指向包含用于签署 Kafka 代理证书以启用基于 TLS 的通信的 OpenShift Secret。Secret 必须包含密钥 ca.crt 下的证书颁发机构的公钥。这个环境变量是可选的，只有在与 Kafka 集群的通信基于 TLS 时才应设置。
   10. STRIMZI_EO_KEY_SECRET_NAME 以指向包含对 Kafka 集群进行 TLS 客户端身份验证的私钥和相关证书的 OpenShift Secret。Secret 必须包含密钥存储，其私钥和证书位于 key entity -operator.p12 下，以及 key entity -operator.password 下的相关密码。这个环境变量是可选的，只有在与 Kafka 集群的通信基于 TLS 时需要 TLS 客户端身份验证时才应设置。
   11. STRIMZI_CA_VALIDITY 证书颁发机构的有效期限.默认值为 365 天。
   12. STRIMZI_CA_RENEWAL 认证机构的续订期限.
   13. STRIMZI_LOG_LEVEL 到打印日志记录消息的级别。该值可设置为：ERROR、WARNING、INFO、DEBUG 和 TRACE.默认 INFO.
   14. STRIMZI_GC_LOG_ENABLED，以启用垃圾收集(GC)日志。默认为 true。默认为在旧证书过期前启动证书续订的 30 天。
   15. STRIMZI_JAVA_OPTS （可选） 用于运行用户的 JVM 的 Java 选项。例如 -Xmx=512M -Xms=256M。
   16. STRIMZI_JAVA_SYSTEM_PROPERTIES （可选） 列出设置为 User Operator 的 -D 选项。An example is -Djavax.net.debug=verbose -DpropertyName=value.

2. 部署 User Operator：

   oc apply -f install/user-operator

3. 验证 User Operator 是否已成功部署：

   oc describe deployment strimzi-user-operator

   当 Replicas: 条目显示 1 available 时，会部署 User Operator。

   注意

   如果您与 OpenShift 集群的连接较慢并且之前没有下载镜像，则部署可能会遇到延迟。

步骤

1. 使用 registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.7 作为基础镜像创建新的 Dockerfile ：

   FROM registry.redhat.io/amq7/amq-streams-kafka-26-rhel7:1.6.7
   USER root:root
   COPY ./my-plugins/ /opt/kafka/plugins/
   USER 1001

   插件文件示例

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-3.4.2.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-3.4.2.jar
   │   ├── mongodb-driver-core-3.4.2.jar
   │   └── README.md
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-0.13.0.jar
   │   ├── mysql-connector-java-5.1.40.jar
   │   ├── README.md
   │   └── wkb-1.0.2.jar
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-0.7.1.jar
       ├── debezium-core-0.7.1.jar
       ├── LICENSE.txt
       ├── postgresql-42.0.0.jar
       ├── protobuf-java-2.6.1.jar
       └── README.md

2. 构建容器镜像。
3. 将自定义镜像推送到容器 registry。

4. 指向新容器映像。

   您可以：

   • 编辑 KafkaConnect.spec.image 属性，即 KafkaConnect 自定义资源。

     如果设置，此属性会覆盖 Cluster Operator 中的 STRIMZI_KAFKA_CONNECT_IMAGES 变量。

     apiVersion: kafka.strimzi.io/v1beta1
     kind: KafkaConnect
     metadata:
       name: my-connect-cluster
     spec: 1
       #...
       image: my-new-container-image 2
       config: 3
         #...

     1

     Kafka Connect 集群的规格。

     2

     容器集的 docker 镜像。

     3

     配置 Kafka Connect worker （而非连接器）。

     或

   • 在 install/cluster-operator/060-Deployment-strimzi-cluster-operator.yaml 文件中编辑 STRIMZI_KAFKA_CONNECT_IMAGES 变量以指向新容器镜像，然后重新安装 Cluster Operator。

步骤

1. 在命令行中，使用 oc apply 命令创建和部署 Kafka Connect S2I 集群：

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

2. 使用 Kafka Connect 插件创建目录：

   $ tree ./my-plugins/
   ./my-plugins/
   ├── debezium-connector-mongodb
   │   ├── bson-3.4.2.jar
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mongodb-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mongodb-driver-3.4.2.jar
   │   ├── mongodb-driver-core-3.4.2.jar
   │   └── README.md
   ├── debezium-connector-mysql
   │   ├── CHANGELOG.md
   │   ├── CONTRIBUTE.md
   │   ├── COPYRIGHT.txt
   │   ├── debezium-connector-mysql-0.7.1.jar
   │   ├── debezium-core-0.7.1.jar
   │   ├── LICENSE.txt
   │   ├── mysql-binlog-connector-java-0.13.0.jar
   │   ├── mysql-connector-java-5.1.40.jar
   │   ├── README.md
   │   └── wkb-1.0.2.jar
   └── debezium-connector-postgres
       ├── CHANGELOG.md
       ├── CONTRIBUTE.md
       ├── COPYRIGHT.txt
       ├── debezium-connector-postgres-0.7.1.jar
       ├── debezium-core-0.7.1.jar
       ├── LICENSE.txt
       ├── postgresql-42.0.0.jar
       ├── protobuf-java-2.6.1.jar
       └── README.md

3. 使用 oc start-build 命令，使用准备的目录启动镜像的新构建：

   oc start-build my-connect-cluster-connect --from-dir ./my-plugins/

   注意

   构建的名称与部署的 Kafka Connect 集群的名称相同。

4. 构建完成后，Kafka Connect 部署会自动使用新镜像。

步骤

1. 根据要安装到的命名空间 Prometheus 修改 Prometheus 安装文件(prometheus.yaml)：

   在 Linux 中，使用：

   sed -i 's/namespace: .*/namespace: my-namespace/' prometheus.yaml

   在 MacOS 中，使用：

   sed -i '' 's/namespace: .*/namespace: my-namespace/' prometheus.yaml

2. 编辑 PodMonitor 资源 in strimzi-pod-monitor.yaml，以定义 Prometheus 作业，它将从 pod 中提取指标数据。

   将 namespaceSelector.matchNames 属性更新为 pod 要从中提取指标的命名空间。

   PodMonitor 用于直接从 pod 中提取数据，用于 Apache Kafka、ZooKeeper、Operator、Kafka Bridge 和 Cruise Control。

3. 编辑 prometheus.yaml 安装文件，使其包含用于直接从节点提取指标的额外配置。

   提供的 Grafana 仪表板显示 CPU、内存和磁盘卷使用情况的指标，它们直接来自节点上的 OpenShift cAdvisor 代理和 kubelet。

   1. 从配置文件（如 example /metrics/prometheus-additional-properties 目录中的 prometheus-additional.yaml ）创建 Secret 资源：

      oc apply -f prometheus-additional.yaml

   2. 编辑 prometheus.yaml 文件中的 additionalScrapeConfigs 属性，使其包含 Secret 的名称和 prometheus-additional.yaml 文件。

4. 部署 Prometheus 资源：

   oc apply -f strimzi-pod-monitor.yaml
   oc apply -f prometheus-rules.yaml
   oc apply -f prometheus.yaml

步骤

1. 从 Alertmanager 配置文件(alert-manager-config.yaml)创建 Secret 资源：

   oc create secret generic alertmanager-alertmanager --from-file=alertmanager.yaml=alert-manager-config.yaml

2. 更新 alert-manager-config.yaml 文件以替换：

   • slack _api_url 属性，以及与 Slack 工作区应用相关的 Slack API URL 的实际值
   • 具有 要在其上发送通知的实际 Slack 频道的频道属性

3. 部署 Alertmanager：

   oc apply -f alert-manager.yaml

步骤

1. 部署 Grafana：

   oc apply -f grafana.yaml

2. 启用 Grafana 仪表板。

步骤

1. 获取 Grafana 服务的详情：

   oc get service grafana

   例如：

   NAME	TYPE	CLUSTER-IP	端口.
   grafana	ClusterIP	172.30.123.40	3000/TCP

   请注意端口转发的端口号。

2. 使用 port-forward 将 Grafana 用户界面重定向到 localhost:3000:

   oc port-forward svc/grafana 3000:3000

3. 将 Web 浏览器指向 http://localhost:3000

   这时会出现 Grafana Log In 页面。

4. 输入您的用户名和密码，然后单击 Log In。

   默认的 Grafana 用户名和密码都是 admin。第一次登录后，您可以更改密码。

5. 添加 Prometheus 作为 数据源。

   • 指定名称
   • 添加 Prometheus 作为类型

   • 指定 Prometheus 服务器 URL(http://prometheus-operated:9090)

     添加详情后，保存并测试连接。

     
6. 从 Dashboards → Import，上传示例仪表板或直接粘贴 JSON。

7. 在顶部标头中，单击仪表板下拉菜单，然后选择您要查看的仪表板。

   当 Prometheus 服务器收集 AMQ Streams 集群的指标一段时间时，会填充仪表板。

步骤

1. 记录对现有 Cluster Operator 资源所做的任何配置更改（在 /install/cluster-operator 目录中）。任何更改都将被 Cluster Operator 的新版本 覆盖。

2. 更新 Cluster Operator。

   1. 根据 Cluster Operator 运行的命名空间修改新版本的安装文件。

      在 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 Deployment 中修改了一个或多个环境变量，请编辑 install/cluster-operator/060-Deployment-cluster-operator.yaml 文件来使用这些环境变量。

3. 当您有更新的配置时，请将其与其它安装资源一起部署：

   oc apply -f install/cluster-operator

   等待滚动更新完成。

4. 获取 Kafka pod 的镜像以确保升级成功：

   oc get po my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   镜像标签显示新的 AMQ Streams 版本，后跟 Kafka 版本。例如，<New AMQ Streams version>-kafka-<Current Kafka version>。

5. 更新现有资源以处理已弃用的自定义资源属性。

   • AMQ Streams 资源升级

步骤

1. 更新 Kafka 集群配置：

   oc edit kafka my-cluster

2. 如果配置了，请确保 Kafka.spec.kafka.config 的 log.message.format .version 和 inter.broker.protocol.version 设置为 当前 Kafka 版本的默认值。

   例如，如果从 Kafka 版本 2.5.0 升级到 2.6.0：

   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.5.0
       config:
         log.message.format.version: "2.5"
         inter.broker.protocol.version: "2.5"
         # ...

   如果没有配置 log.message.format .version 和 inter.broker.protocol.version，AMQ Streams 会在下一步中将这些版本更新至 Kafka 版本后自动将这些版本更新至当前的默认值。

   注意

   log.message.format.version 和 inter.broker.protocol.version 的值必须是字符串，以防止它们被解释为浮动点数。

3. 更改 Kafka.spec.kafka.version 以指定新的 Kafka 版本；保留 log.message.format.version 和 inter.broker.protocol.version，默认为 当前 Kafka 版本。

   注意

   更改 kafka.version 可确保集群中的所有代理都将升级为使用新的代理二进制文件。在此过程中，一些代理使用旧二进制文件，另一些则已升级至新二进制文件。不要更改 inter.broker.protocol.version，确保代理可以在升级过程中继续相互通信。

   例如，如果从 Kafka 2.5.0 升级到 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0 1
       config:
         log.message.format.version: "2.5" 2
         inter.broker.protocol.version: "2.5" 3
         # ...

   1

   Kafka 版本已改为新版本。

   2

   消息格式版本保持不变。

   3

   代理协议版本没有改变。

   警告

   如果新 Kafka 版本的 inter.broker.protocol.version 发生变化，则无法降级 Kafka。broker 协议版本决定了代理存储的持久元数据的模式，包括写入到 __consumer_offset 的消息。降级集群将不会理解消息。

4. 如果在 Kafka 自定义资源（在 Kafka .spec.kafka.image 中）中定义了 Kafka 集群 的镜像，请将镜像 更新为指向具有新 Kafka 版本的容器镜像。

   请参阅 Kafka 版本和镜像映射

5. 保存并退出编辑器，然后等待滚动更新完成。

   通过观察 pod 状态转换来检查滚动更新的进度：

   oc get pods my-cluster-kafka-0 -o jsonpath='{.spec.containers[0].image}'

   滚动更新可确保每个 pod 在新版本的 Kafka 中使用代理二进制文件。

6. 根据您选择的 升级客户端策略，升级所有客户端应用以使用新版本的客户端二进制文件。

   如果需要，将 Kafka Connect 和 MirrorMaker 的 version 属性设置为 Kafka 的新版本：

   1. 对于 Kafka Connect，更新 KafkaConnect.spec.version。
   2. 对于 MirrorMaker，更新 KafkaMirrorMaker.spec.version。
   3. 对于 MirrorMaker 2.0，更新 KafkaMirrorMaker2.spec.version。

7. 如果配置，更新 Kafka 资源，以使用新的 inter.broker.protocol.version 版本。否则，请转至第 9 步。

   例如，如果升级到 Kafka 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0
       config:
         log.message.format.version: "2.5"
         inter.broker.protocol.version: "2.6"
         # ...

8. 等待 Cluster Operator 更新集群。

9. 如果配置，更新 Kafka 资源，以使用新的 log.message.format.version 版本。否则，请转至第 10 步。

   例如，如果升级到 Kafka 2.6.0：

   apiVersion: kafka.strimzi.io/v1beta1
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 2.6.0
       config:
         log.message.format.version: "2.6"
         inter.broker.protocol.version: "2.6"
         # ...

10. 等待 Cluster Operator 更新集群。

    • Kafka 集群和客户端现在使用新的 Kafka 版本。
    • 代理配置为使用 Kafka 的新版本的 Inter-broker 协议版本和消息格式发送消息。

1. 将 Kafka client .jar 文件替换为新版本。
2. 在使用者配置中，将 合作粘滞 附加到 partition.assignment.strategy。例如，如果设定了 范围 策略，请将配置更改为 范围，协作粘性。
3. 依次重新启动组中的每个使用者，等待使用者在每次重启后重新加入组。
4. 通过从使用者配置中删除先前的 partition.assignment.strategy 来 重新配置组中的每个消费者，仅保留 协作式 策略。
5. 依次重新启动组中的每个使用者，等待使用者在每次重启后重新加入组。

1. 将 Kafka Streams .jar 文件替换为新版本。
2. 在 Kafka Streams 配置中，将 upgrade.from 配置参数设置为您要从中升级的 Kafka 版本（例如 2.3）。
3. 依次重新启动每个流处理器（节点）。
4. 从 Kafka Streams 配置中删除 upgrade.from 配置参数。
5. 依次重新启动组中的每个消费者。