流程

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 显示 ready/expected 的副本数量。当 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/amq7/amq-streams-rhel8-operator:2.3.0
           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 显示 ready/expected 的副本数量。当 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 环境变量的值设置为 *。

   apiVersion: apps/v1
   kind: Deployment
   spec:
     # ...
     template:
       spec:
         # ...
         serviceAccountName: strimzi-cluster-operator
         containers:
         - name: strimzi-cluster-operator
           image: registry.redhat.io/amq7/amq-streams-rhel8-operator:2.3.0
           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 显示 ready/expected 的副本数量。当 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 集群的名称。

   使用默认部署，您要安装一个 Entity Operator 集群、3 Kafka pod 和 3 ZooKeeper pod。

   READY 显示 ready/expected 的副本数量。当 STATUS 显示为 Running 时，部署成功。

流程

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 显示 ready/expected 的副本数量。当 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 显示 ready/expected 的副本数量。当 STATUS 显示为 Running 时，部署成功。

流程

1. 将 Kafka 连接到您的 OpenShift 集群。使用 example /connect/kafka-connect.yaml 文件部署 Kafka Connect。

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

2. 检查部署的状态：

   oc get deployments -n <my_cluster_operator_namespace>

   输出显示了部署名称和就绪状态

   NAME                        READY  UP-TO-DATE  AVAILABLE
   my-connect-cluster-connect  1/1    1           1

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

   READY 显示 ready/expected 的副本数量。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

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

   对于 MirrorMaker:

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

   对于 MirrorMaker 2.0:

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

2. 检查部署的状态：

   oc get deployments -n <my_cluster_operator_namespace>

   输出显示了部署名称和就绪状态

   NAME                          READY  UP-TO-DATE  AVAILABLE
   my-mirror-maker-mirror-maker  1/1    1           1
   my-mm2-cluster-mirrormaker2   1/1    1           1

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

   READY 显示 ready/expected 的副本数量。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

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

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

2. 检查部署的状态：

   oc get deployments -n <my_cluster_operator_namespace>

   输出显示了部署名称和就绪状态

   NAME              READY  UP-TO-DATE  AVAILABLE
   my-bridge-bridge  1/1    1           1

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

   READY 显示 ready/expected 的副本数量。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

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

   oc get pods -o name
   
   pod/kafka-consumer
   # ...
   pod/my-bridge-bridge-7cbd55496b-nclrt

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

   oc port-forward pod/my-bridge-bridge-7cbd55496b-nclrt 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

   用于监视 KafkaTopic 资源的 topics Operator 的 OpenShift 命名空间。指定 Kafka 集群的命名空间。

   2

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

   3

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

   4

   连接到 ZooKeeper 集群的地址和端口对。这必须与 Kafka 集群使用的相同 ZooKeeper 集群。

   5

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

   6

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

   7

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

   8

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

   9

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

   10

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

   11

   （可选）为 Topic 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_PUBLI_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 显示 ready/expected 的副本数量。当 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_LOG_LEVEL 9
                 value: INFO
               - name: STRIMZI_GC_LOG_ENABLED 10
                 value: "true"
               - name: STRIMZI_CA_VALIDITY 11
                 value: "365"
               - name: STRIMZI_CA_RENEWAL 12
                 value: "30"
               - name: STRIMZI_JAVA_OPTS 13
                 value: "-Xmx=512M -Xms=256M"
               - name: STRIMZI_JAVA_SYSTEM_PROPERTIES 14
                 value: "-Djavax.net.debug=verbose -DpropertyName=value"
               - name: STRIMZI_SECRET_PREFIX 15
                 value: "kafka-"
               - name: STRIMZI_ACLS_ADMIN_API_SUPPORTED 16
                 value: "true"
               - name: STRIMZI_MAINTENANCE_TIME_WINDOWS 17
                 value: '* * 8-10 * * ?;* * 14-15 * * ?'
               - name: STRIMZI_KAFKA_ADMIN_CLIENT_CONFIGURATION 18
                 value: |
                   default.api.timeout.ms=120000
                   request.timeout.ms=60000

   1

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

   2

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

   3

   包含为 mTLS 身份验证签名新用户证书的证书颁发机构(ca.crt)值的 OpenShift Secret。

   4

   包含为 mTLS 身份验证签名新用户证书的证书颁发机构(ca.key)值的 OpenShift Secret。

   5

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

   6

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

   7

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

   8

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

   9

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

   10

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

   11

   证书颁发机构的有效性周期。默认值为 365 天。

   12

   证书颁发机构的续订期限。续订周期从当前证书的过期日期向后兼容。默认为 30 天，可在旧证书过期之前启动证书续订。

   13

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

   14

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

   15

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

   16

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

   17

   （可选）Semi-colon 分离的 Cron 表达式列表，用于定义过期用户证书的维护时间窗口。

   18

   （可选）用来以属性格式配置 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

   包含密钥存储(entity-operator.p12)的 OpenShift Secret，其私钥和证书用于对 Kafka 集群进行 mTLS 验证。Secret 还必须包含用于访问密钥存储的密码(entity-operator.password)。

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 显示 ready/expected 的副本数量。当 AVAILABLE 输出显示为 1 时，部署成功。

流程

1. 检查是否启用了对用户定义的项目的监控：

   oc get pods -n openshift-user-workload-monitoring

   如果启用，则返回监控组件的 Pod。例如：

   NAME                                   READY   STATUS    RESTARTS   AGE
   prometheus-operator-5cc59f9bc6-kgcq8   1/1     Running   0          25s
   prometheus-user-workload-0             5/5     Running   1          14s
   prometheus-user-workload-1             5/5     Running   1          14s
   thanos-ruler-user-workload-0           3/3     Running   0          14s
   thanos-ruler-user-workload-1           3/3     Running   0          14s

   如果没有返回 pod，则禁用对用户定义的项目的监控。请参阅 第 8.5 节 “在 OpenShift 中查看 Kafka 指标和仪表板” 中的先决条件。

2. 在 example /metrics/prometheus-install/strimzi-pod-monitor.yaml 中定义多个 PodMonitor 资源。

   对于每个 PodMonitor 资源，编辑 spec.namespaceSelector.matchNames 属性：

   apiVersion: monitoring.coreos.com/v1
   kind: PodMonitor
   metadata:
     name: cluster-operator-metrics
     labels:
       app: strimzi
   spec:
     selector:
       matchLabels:
         strimzi.io/kind: cluster-operator
     namespaceSelector:
       matchNames:
         - <project-name> 1
     podMetricsEndpoints:
     - path: /metrics
       port: http
   # ...

   1

   要从中提取指标的 Pod 的项目，如 Kafka。

3. 将 strimzi-pod-monitor.yaml 文件部署到运行 Kafka 集群的项目：

   oc apply -f strimzi-pod-monitor.yaml -n MY-PROJECT

4. 将示例 Prometheus 规则部署到同一个项目中：

   oc apply -f prometheus-rules.yaml -n MY-PROJECT

流程

1. 为 Grafana 创建 ServiceAccount。此处的资源名为 grafana-serviceaccount。

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: grafana-serviceaccount
     labels:
       app: strimzi

2. 将 ServiceAccount 部署到包含 Kafka 集群的项目：

   oc apply -f GRAFANA-SERVICEACCOUNT -n MY-PROJECT

3. 创建一个 ClusterRoleBinding 资源，它将 cluster-monitoring-view 角色分配给 Grafana ServiceAccount。此处的资源名为 grafana-cluster-monitoring-binding。

   apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRoleBinding
   metadata:
     name: grafana-cluster-monitoring-binding
     labels:
       app: strimzi
   subjects:
     - kind: ServiceAccount
       name: grafana-serviceaccount
       namespace: <my-project> 1
   roleRef:
     kind: ClusterRole
     name: cluster-monitoring-view
     apiGroup: rbac.authorization.k8s.io

   1

   项目的名称。

4. 将 ClusterRoleBinding 部署到包含 Kafka 集群的项目：

   oc apply -f <grafana-cluster-monitoring-binding> -n <my-project>

流程

1. 获取 Grafana ServiceAccount 的访问令牌：

   oc serviceaccounts get-token grafana-serviceaccount -n MY-PROJECT

   复制访问令牌以在下一步中使用。

2. 创建包含 Grafana 的 Thanos Querier 配置的 datasource.yaml 文件。

   按照所示，将访问令牌粘贴到 httpHeaderValue1 属性中。

   apiVersion: 1
   
   datasources:
   - name: Prometheus
     type: prometheus
     url: https://thanos-querier.openshift-monitoring.svc.cluster.local:9091
     access: proxy
     basicAuth: false
     withCredentials: false
     isDefault: true
     jsonData:
       timeInterval: 5s
       tlsSkipVerify: true
       httpHeaderName1: "Authorization"
     secureJsonData:
       httpHeaderValue1: "Bearer ${GRAFANA-ACCESS-TOKEN}" 1
     editable: true

   1

   GRAFANA-ACCESS-TOKEN ：Grafana ServiceAccount 的访问令牌值。

3. 从 Datasource .yaml 文件中创建一个名为 grafana-config 的配置映射：

   oc create configmap grafana-config --from-file=datasource.yaml -n MY-PROJECT

4. 创建由 Deployment 和 Service 组成的 Grafana 应用程序。

   grafana-config 配置映射作为数据源配置的卷挂载。

   apiVersion: apps/v1
   kind: Deployment
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     replicas: 1
     selector:
       matchLabels:
         name: grafana
     template:
       metadata:
         labels:
           name: grafana
       spec:
         serviceAccountName: grafana-serviceaccount
         containers:
         - name: grafana
           image: grafana/grafana:9.3.1
           ports:
           - name: grafana
             containerPort: 3000
             protocol: TCP
           volumeMounts:
           - name: grafana-data
             mountPath: /var/lib/grafana
           - name: grafana-logs
             mountPath: /var/log/grafana
           - name: grafana-config
             mountPath: /etc/grafana/provisioning/datasources/datasource.yaml
             readOnly: true
             subPath: datasource.yaml
           readinessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 5
             periodSeconds: 10
           livenessProbe:
             httpGet:
               path: /api/health
               port: 3000
             initialDelaySeconds: 15
             periodSeconds: 20
         volumes:
         - name: grafana-data
           emptyDir: {}
         - name: grafana-logs
           emptyDir: {}
         - name: grafana-config
           configMap:
             name: grafana-config
   ---
   apiVersion: v1
   kind: Service
   metadata:
     name: grafana
     labels:
       app: strimzi
   spec:
     ports:
     - name: grafana
       port: 3000
       targetPort: 3000
       protocol: TCP
     selector:
       name: grafana
     type: ClusterIP

5. 将 Grafana 应用程序部署到包含 Kafka 集群的项目中：

   oc apply -f <grafana-application> -n <my-project>

流程

1. 获取到 Grafana 服务的路由详情。例如：

   oc get routes
   
   NAME               HOST/PORT                         PATH  SERVICES
   MY-GRAFANA-ROUTE   MY-GRAFANA-ROUTE-amq-streams.net        grafana

2. 在 Web 浏览器中，使用 Route 主机和端口的 URL 访问 Grafana 登录屏幕。

3. 输入您的用户名和密码，然后点 Log In。

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

4. 在 Configuration > Data Sources 中，检查是否已创建 Prometheus 数据源。数据源是在 第 8.5.5 节 “使用 Prometheus 数据源部署 Grafana” 中创建。
5. 单击 + 图标，然后单击 Import。
6. 在 example/metrics/grafana-dashboards 中，复制要导入的仪表板的 JSON。
7. 将 JSON 粘贴到文本框中，然后点 Load。
8. 为其他 Grafana 仪表板重复步骤 5-7。

1. 在 spec.template 属性中配置 tracer 服务。

   • 使用 追踪环境变量 作为模板配置属性。
   • 对于 OpenTelemetry，将 spec.tracing.type 属性设置为 opentelemetry。
   • 对于 OpenTracing，将 spec.tracing.type 属性设置为 jaeger。

   使用 OpenTelemetry 的 Kafka Connect 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
     #...

   使用 OpenTelemetry 的 MirrorMaker 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     #...
     template:
       mirrorMakerContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   使用 OpenTelemetry 的 MirrorMaker 2.0 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   使用 OpenTelemetry 的 Kafka Bridge 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     #...
     template:
       bridgeContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-otel-service
           - name: OTEL_EXPORTER_JAEGER_ENDPOINT
             value: "http://jaeger-host:14250"
     tracing:
       type: opentelemetry
   #...

   使用 OpenTracing 进行 Kafka Connect 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaConnect
   metadata:
     name: my-connect-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
     #...

   使用 OpenTracing 的 MirrorMaker 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker
   metadata:
     name: my-mirror-maker
   spec:
     #...
     template:
       mirrorMakerContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

   使用 OpenTracing 的 MirrorMaker 2.0 的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

   使用 OpenTracing 的 Kafka 网桥的追踪配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaBridge
   metadata:
     name: my-bridge
   spec:
     #...
     template:
       bridgeContainer:
         env:
           - name: JAEGER_SERVICE_NAME
             value: my-jaeger-service
           - name: JAEGER_AGENT_HOST
             value: jaeger-agent-name
           - name: JAEGER_AGENT_PORT
             value: "6831"
     tracing:
       type: jaeger
   #...

2. 创建或更新资源：

   oc apply -f <resource_configuration_file>

1. 将 Maven 依赖项添加到客户端应用程序的 pom.xml 文件中：

   OpenTelemetry 的依赖项

   <dependency>
       <groupId>io.opentelemetry</groupId>
       <artifactId>opentelemetry-sdk-extension-autoconfigure</artifactId>
       <version>1.18.0-alpha</version>
   </dependency>
   <dependency>
     <groupId>io.opentelemetry.instrumentation</groupId>
     <artifactId>opentelemetry-kafka-clients-{OpenTelemetryKafkaClient}</artifactId>
     <version>1.18.0-alpha</version>
   </dependency>
   <dependency>
     <groupId>io.opentelemetry</groupId>
     <artifactId>opentelemetry-exporter-jaeger</artifactId>
     <version>1.18.0</version>
   </dependency>

   OpenTracing 的依赖项

   <dependency>
       <groupId>io.jaegertracing</groupId>
       <artifactId>jaeger-client</artifactId>
       <version>1.3.2</version>
   </dependency>
   <dependency>
     <groupId>io.opentracing.contrib</groupId>
     <artifactId>opentracing-kafka-client</artifactId>
     <version>0.1.15</version>
   </dependency>

2. 使用追踪环境变量定义 tracer 的配置。

3. 创建一个 tracer，它使用环境变量初始化：

   为 OpenTelemetry 创建 tracer

   OpenTelemetry ot = GlobalOpenTelemetry.get();

   为 OpenTracing 创建追踪器

   Tracer tracer = Configuration.fromEnv().getTracer();

4. 将 tracer 注册为全局 tracer：

   GlobalTracer.register(tracer);

5. 检测您的客户端：

   • 第 9.3.4 节 “提取制作者和消费者以进行追踪”
   • 第 9.3.5 节 “用于追踪的 Kafka Streams 应用程序”

流程

1. 将追踪工件添加到 AMQ Streams Kafka 镜像的 /opt/kafka/libs/ 目录中。

   您可以使用 红帽生态系统目录 上的 Kafka 容器镜像作为创建新自定义镜像的基础镜像。

   Zipkin 的 OpenTelemetry 构件

   io.opentelemetry:opentelemetry-exporter-zipkin

2. 为新的追踪实施设置追踪器和端点。

   Zikpin tracer 配置示例

   apiVersion: kafka.strimzi.io/v1beta2
   kind: KafkaMirrorMaker2
   metadata:
     name: my-mm2-cluster
   spec:
     #...
     template:
       connectContainer:
         env:
           - name: OTEL_SERVICE_NAME
             value: my-zipkin-service
           - name: OTEL_EXPORTER_ZIPKIN_ENDPOINT
             value: http://zipkin-exporter-host-name:9411/api/v2/spans 1
           - name: OTEL_TRACES_EXPORTER
             value: zipkin 2
     tracing:
       type: opentelemetry
   #...

   1

   指定要连接的 Zipkin 端点。

   2

   Zipkin exporter。

1. 按照标准序列 将 AMQ Streams 升级到 1.7。
2. 使用 AMQ Streams 1.8 提供的 Red Hat AMQ Streams API Conversion Tool 将 AMQ Streams 自定义资源转换为 v1beta2。

3. 执行以下操作之一：

   • 升级到 AMQ Streams 1.8 （其中 ControlPlaneListener 功能被默认禁用）。
   • 升级到 AMQ Streams 2.0 或 2.2 （其中 ControlPlaneListener 功能门被默认启用），并禁用 ControlPlaneListener 功能门。
4. 启用 ControlPlaneListener 功能门。
5. 按照标准序列 升级到 AMQ Streams 2.3。

1. 升级到 AMQ Streams 1.7。
2. 下载由 AMQ Streams 1.8 提供的 Red Hat AMQ Streams API Conversion Tool。

3. 将自定义资源和 CRD 转换为 v1beta2。

   如需更多信息，请参阅 AMQ Streams 1.7 升级文档。

4. 在 OperatorHub 中，删除 Red Hat Integration - AMQ Streams Operator 版本 1.7。

5. 如果存在，请删除 Red Hat Integration - AMQ Streams Operator 的版本 2.3。

   如果不存在，请转到下一步。

   如果 AMQ Streams Operator 的批准策略被设置为 Automatic，则集群中可能已存在 operator 版本 2.3。如果您没有将 自定义资源和 CRD 转换为 v1beta2 API 版本，Operator 管理的自定义资源和 CRD 将使用旧的 API 版本。因此，2.3 Operator 处于 Pending 状态。在这种情况下，您需要删除 Red Hat Integration - AMQ Streams Operator 和 1.7 版本 2.3 版本。

   如果删除了这两个 Operator，协调会暂停，直到安装了新的 operator 版本。立即按照以下步骤操作，以便对自定义资源的任何更改都不会延迟。

6. 在 OperatorHub 中，执行以下操作之一：

   • 升级到 Red Hat Integration 版本 1.8 - AMQ Streams Operator （其中 ControlPlaneListener 功能门被默认禁用）。
   • 升级至 Red Hat Integration 的版本 2.0 或 2.2 - AMQ Streams Operator （默认为启用 ControlPlaneListener 功能门），且禁用了 ControlPlaneListener 功能门。

7. 马上升级到 Red Hat Integration 的 2.3 版本 - AMQ Streams Operator。

   安装的 2.3 operator 开始监视集群并执行滚动更新。您可能会注意到这个过程中集群性能的临时降低。

流程

1. 记录对现有 Cluster Operator 资源（在 /install/cluster-operator 目录中）所做的任何配置更改。Cluster Operator 的新版本 将覆盖 任何更改。
2. 更新您的自定义资源，以反映 AMQ Streams 版本 2.3 可用的受支持配置选项。

3. 更新 Cluster Operator。

   1. 根据 Cluster Operator 在其中运行的命名空间，为新的 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-strimzi-cluster-operator.yaml 文件来使用这些环境变量。

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

   oc replace -f install/cluster-operator

   等待滚动更新完成。

5. 如果新 Operator 版本不再支持从中升级的 Kafka 版本，Cluster Operator 会返回一条错误消息，表示不支持该版本。否则，不会返回任何错误消息。

   • 如果返回错误消息，升级到新 Cluster Operator 版本支持的 Kafka 版本：

     1. 编辑 Kafka 自定义资源。
     2. 将 spec.kafka.version 属性改为受支持的 Kafka 版本。
   • 如果未 返回错误消息，请转到下一步。稍后您将升级 Kafka 版本。

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

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

   镜像标签显示新的 Operator 版本。例如：

   registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0

代理级消费者第一个策略

1. 升级所有消耗的应用程序。
2. 将 broker-level log.message.format.version 更改为新版本。
3. 升级所有生产应用程序。

1. 升级所有消耗的应用程序。
2. 将 topic-level message.format.version 更改为新版本。
3. 升级所有生产应用程序。

1. 将 topic-level message.format.version 改为旧版本（或依赖主题 defaulting to the broker-level log.message.format.version）。
2. 升级所有消耗和生产应用程序。
3. 验证升级的应用程序是否正常工作。
4. 将 topic-level message.format.version 更改为新版本。

流程

1. 更新 Kafka 集群配置：

   oc edit kafka <my_cluster>

2. 如果配置，请检查 inter.broker.protocol.version 和 log.message.format.version 属性是否已设置为 当前版本。

   例如，如果从 Kafka 版本 3.2.3 升级到 3.3.1，则当前版本为 3.2:

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

   如果没有配置 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 的 inter.broker.protocol.version 可确保代理可以在升级期间继续相互通信。

   例如，如果从 Kafka 3.2.3 升级到 3.3.1：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1 1
       config:
         log.message.format.version: "3.2" 2
         inter.broker.protocol.version: "3.2" 3
         # ...

   1

   Kafka 版本被改为新版本。

   2

   消息格式版本保持不变。

   3

   Inter-broker 协议版本保持不变。

   警告

   如果新 Kafka 版本有 inter.broker.protocol.version 的变化，则无法降级 Kafka。inter-broker 协议版本决定了用于代理存储的持久性元数据的 schema，包括写入 __consumer_offsets 的消息。降级的集群将无法了解信息。

4. 如果在 Kafka 自定义资源中定义了 Kafka 集群的镜像，在 Kafka.spec.kafka.image 中，更新 image 以指向新的 Kafka 版本的容器镜像。

   请参阅 Kafka 版本和镜像映射

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

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

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

   滚动更新可确保每个 pod 都对 Kafka 的新版本使用 broker 二进制文件。

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

   如果需要，将 Kafka Connect 和 MirrorMaker 的版本 属性设置为 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 3.3.1：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.2"
         inter.broker.protocol.version: "3.3"
         # ...

8. 等待 Cluster Operator 更新集群。

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

   例如，如果升级到 Kafka 3.3.1：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.3.1
       config:
         log.message.format.version: "3.3"
         inter.broker.protocol.version: "3.3"
         # ...

   重要

   从 Kafka 3.0.0，当 inter.broker.protocol.version 设置为 3.0 或更高版本时，logging.message.format.version 选项会被忽略，不需要设置。

10. 等待 Cluster Operator 更新集群。

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

流程

1. 更新 Kafka 集群配置。

   oc edit kafka KAFKA-CONFIGURATION-FILE

2. 更改 Kafka.spec.kafka.version 以指定上一个版本。

   例如，如果从 Kafka 3.3.1 降级到 3.2.3：

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   spec:
     # ...
     kafka:
       version: 3.2.3 1
       config:
         log.message.format.version: "3.2" 2
         inter.broker.protocol.version: "3.2" 3
         # ...

   1

   Kafka 版本被改为以前的版本。

   2

   消息格式版本保持不变。

   3

   Inter-broker 协议版本保持不变。

   注意

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

3. 如果 Kafka 版本的镜像与 Cluster Operator 的 STRIMZI_KAFKA_IMAGES 中定义的镜像不同，请更新 Kafka.spec.kafka.image。

   请查看 第 10.5.3 节 “Kafka 版本和镜像映射”

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

   检查日志中的更新，或者查看 pod 状态转换：

   oc logs -f CLUSTER-OPERATOR-POD-NAME | grep -E "Kafka version downgrade from [0-9.]+ to [0-9.]+, phase ([0-9]+) of \1 completed"

   oc get pod -w

   检查 Cluster Operator 日志中的 INFO 级别信息：

   Reconciliation #NUM(watch) Kafka(NAMESPACE/NAME): Kafka version downgrade from FROM-VERSION to TO-VERSION, phase 1 of 1 completed

5. 降级所有客户端应用程序（使用者），以使用之前版本的客户端二进制文件。

   Kafka 集群和客户端现在使用之前的 Kafka 版本。

6. 如果您恢复到 1.7 之前的 AMQ Streams 版本，该版本使用 ZooKeeper 作为存储主题元数据，请删除 Kafka 集群的内部主题存储主题。

   oc run kafka-admin -ti --image=registry.redhat.io/amq7/amq-streams-kafka-33-rhel8:2.3.0 --rm=true --restart=Never -- ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi-topic-operator-kstreams-topic-store-changelog --delete && ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --topic __strimzi_store_topic --delete