Chapter 27. Managing AMQ Streams

Procedure

1. Add or edit the maintenanceTimeWindows property in the Kafka resource. For example to allow maintenance between 0800 and 1059 and between 1400 and 1559 you would set the maintenanceTimeWindows as shown below:

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
   spec:
     kafka:
       # ...
     zookeeper:
       # ...
     maintenanceTimeWindows:
       - "* * 8-10 * * ?"
       - "* * 14-15 * * ?"

   Copy to Clipboard Toggle word wrap

2. Create or update the resource:

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

Procedure

1. Set maxUnavailable to 0 (zero) in the Kafka and ZooKeeper sections of the Kafka resource using template settings.

   Specifying a pod disruption budget

   apiVersion: kafka.strimzi.io/v1beta2
   kind: Kafka
   metadata:
     name: my-cluster
     namespace: myproject
   spec:
     kafka:
       template:
         podDisruptionBudget:
           maxUnavailable: 0
   
     # ...
     zookeeper:
       template:
         podDisruptionBudget:
           maxUnavailable: 0
     # ...

   Copy to Clipboard Toggle word wrap

   This setting prevents the automatic eviction of pods in case of planned disruptions, leaving the AMQ Streams Drain Cleaner and Cluster Operator to roll the pods on different worker nodes.

   Add the same configuration for ZooKeeper if you want to use AMQ Streams Drain Cleaner to drain ZooKeeper nodes.

2. Update the Kafka resource:

   oc apply -f <kafka_configuration_file>

   Copy to Clipboard Toggle word wrap

3. Deploy the AMQ Streams Drain Cleaner.

   • To run the Drain Cleaner on OpenShift, apply the resources in the /install/drain-cleaner/openshift directory.

     oc apply -f ./install/drain-cleaner/openshift

     Copy to Clipboard Toggle word wrap

Procedure

1. Drain a specified OpenShift node hosting the Kafka broker or ZooKeeper pods.

   oc get nodes
   oc drain <name-of-node> --delete-emptydir-data --ignore-daemonsets --timeout=6000s --force

   Copy to Clipboard Toggle word wrap

2. Check the eviction events in the AMQ Streams Drain Cleaner log to verify that the pods have been annotated for restart.

   AMQ Streams Drain Cleaner log show annotations of pods

   INFO ... Received eviction webhook for Pod my-cluster-zookeeper-2 in namespace my-project
   INFO ... Pod my-cluster-zookeeper-2 in namespace my-project will be annotated for restart
   INFO ... Pod my-cluster-zookeeper-2 in namespace my-project found and annotated for restart
   
   INFO ... Received eviction webhook for Pod my-cluster-kafka-0 in namespace my-project
   INFO ... Pod my-cluster-kafka-0 in namespace my-project will be annotated for restart
   INFO ... Pod my-cluster-kafka-0 in namespace my-project found and annotated for restart

   Copy to Clipboard Toggle word wrap

3. Check the reconciliation events in the Cluster Operator log to verify the rolling updates.

   Cluster Operator log shows rolling updates

   INFO  PodOperator:68 - Reconciliation #13(timer) Kafka(my-project/my-cluster): Rolling Pod my-cluster-zookeeper-2
   INFO  PodOperator:68 - Reconciliation #13(timer) Kafka(my-project/my-cluster): Rolling Pod my-cluster-kafka-0
   INFO  AbstractOperator:500 - Reconciliation #13(timer) Kafka(my-project/my-cluster): reconciled

   Copy to Clipboard Toggle word wrap

Procedure

1. Find the name of the resource that controls the Kafka or ZooKeeper pods you want to manually update.

   For example, if your Kafka cluster is named my-cluster, the corresponding names are my-cluster-kafka and my-cluster-zookeeper.

2. Use oc annotate to annotate the appropriate resource in OpenShift.

   Annotating a StrimziPodSet

   oc annotate strimzipodset <cluster_name>-kafka strimzi.io/manual-rolling-update=true
   
   oc annotate strimzipodset <cluster_name>-zookeeper strimzi.io/manual-rolling-update=true

   Copy to Clipboard Toggle word wrap

3. Wait for the next reconciliation to occur (every two minutes by default). A rolling update of all pods within the annotated resource is triggered, as long as the annotation was detected by the reconciliation process. When the rolling update of all the pods is complete, the annotation is removed from the resource.

Procedure

1. Find the name of the Kafka or ZooKeeper Pod you want to manually update.

   For example, if your Kafka cluster is named my-cluster, the corresponding Pod names are my-cluster-kafka-index and my-cluster-zookeeper-index. The index starts at zero and ends at the total number of replicas minus one.

2. Annotate the Pod resource in OpenShift.

   Use oc annotate:

   oc annotate pod cluster-name-kafka-index strimzi.io/manual-rolling-update=true
   
   oc annotate pod cluster-name-zookeeper-index strimzi.io/manual-rolling-update=true

   Copy to Clipboard Toggle word wrap
3. Wait for the next reconciliation to occur (every two minutes by default). A rolling update of the annotated Pod is triggered, as long as the annotation was detected by the reconciliation process. When the rolling update of a pod is complete, the annotation is removed from the Pod.

1. Use Option 1 when you can recover all KafkaTopic resources.

   The KafkaTopic resources must therefore be recovered before the cluster is started so that the corresponding topics are not deleted by the Topic Operator.

2. Use Option 2 when you are unable to recover all KafkaTopic resources.

   In this case, you deploy your cluster without the Topic Operator, delete the Topic Operator topic store metadata, and then redeploy the Kafka cluster with the Topic Operator so it can recreate the KafkaTopic resources from the corresponding topics.

Procedure

1. Check information on the PVs in the cluster:

   oc get pv

   Copy to Clipboard Toggle word wrap

   Information is presented for PVs with data.

   Example output showing columns important to this procedure:

   NAME                                         RECLAIMPOLICY CLAIM
   pvc-5e9c5c7f-3317-11ea-a650-06e1eadd9a4c ... Retain ...    myproject/data-my-cluster-zookeeper-1
   pvc-5e9cc72d-3317-11ea-97b0-0aef8816c7ea ... Retain ...    myproject/data-my-cluster-zookeeper-0
   pvc-5ead43d1-3317-11ea-97b0-0aef8816c7ea ... Retain ...    myproject/data-my-cluster-zookeeper-2
   pvc-7e1f67f9-3317-11ea-a650-06e1eadd9a4c ... Retain ...    myproject/data-0-my-cluster-kafka-0
   pvc-7e21042e-3317-11ea-9786-02deaf9aa87e ... Retain ...    myproject/data-0-my-cluster-kafka-1
   pvc-7e226978-3317-11ea-97b0-0aef8816c7ea ... Retain ...    myproject/data-0-my-cluster-kafka-2

   Copy to Clipboard Toggle word wrap
   • NAME shows the name of each PV.
   • RECLAIM POLICY shows that PVs are retained.
   • CLAIM shows the link to the original PVCs.

2. Recreate the original namespace:

   oc create namespace myproject

   Copy to Clipboard Toggle word wrap

3. Recreate the original PVC resource specifications, linking the PVCs to the appropriate PV:

   For example:

   apiVersion: v1
   kind: PersistentVolumeClaim
   metadata:
     name: data-0-my-cluster-kafka-0
   spec:
     accessModes:
     - ReadWriteOnce
     resources:
       requests:
         storage: 100Gi
     storageClassName: gp2-retain
     volumeMode: Filesystem
     volumeName: pvc-7e1f67f9-3317-11ea-a650-06e1eadd9a4c

   Copy to Clipboard Toggle word wrap

4. Edit the PV specifications to delete the claimRef properties that bound the original PVC.

   For example:

   apiVersion: v1
   kind: PersistentVolume
   metadata:
     annotations:
       kubernetes.io/createdby: aws-ebs-dynamic-provisioner
       pv.kubernetes.io/bound-by-controller: "yes"
       pv.kubernetes.io/provisioned-by: kubernetes.io/aws-ebs
     creationTimestamp: "<date>"
     finalizers:
     - kubernetes.io/pv-protection
     labels:
       failure-domain.beta.kubernetes.io/region: eu-west-1
       failure-domain.beta.kubernetes.io/zone: eu-west-1c
     name: pvc-7e226978-3317-11ea-97b0-0aef8816c7ea
     resourceVersion: "39431"
     selfLink: /api/v1/persistentvolumes/pvc-7e226978-3317-11ea-97b0-0aef8816c7ea
     uid: 7efe6b0d-3317-11ea-a650-06e1eadd9a4c
   spec:
     accessModes:
     - ReadWriteOnce
     awsElasticBlockStore:
       fsType: xfs
       volumeID: aws://eu-west-1c/vol-09db3141656d1c258
     capacity:
       storage: 100Gi
     claimRef:
       apiVersion: v1
       kind: PersistentVolumeClaim
       name: data-0-my-cluster-kafka-2
       namespace: myproject
       resourceVersion: "39113"
       uid: 54be1c60-3319-11ea-97b0-0aef8816c7ea
     nodeAffinity:
       required:
         nodeSelectorTerms:
         - matchExpressions:
           - key: failure-domain.beta.kubernetes.io/zone
             operator: In
             values:
             - eu-west-1c
           - key: failure-domain.beta.kubernetes.io/region
             operator: In
             values:
             - eu-west-1
     persistentVolumeReclaimPolicy: Retain
     storageClassName: gp2-retain
     volumeMode: Filesystem

   Copy to Clipboard Toggle word wrap

   In the example, the following properties are deleted:

   claimRef:
     apiVersion: v1
     kind: PersistentVolumeClaim
     name: data-0-my-cluster-kafka-2
     namespace: myproject
     resourceVersion: "39113"
     uid: 54be1c60-3319-11ea-97b0-0aef8816c7ea

   Copy to Clipboard Toggle word wrap

5. Deploy the Cluster Operator.

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

   Copy to Clipboard Toggle word wrap

6. Recreate your cluster.

   Follow the steps depending on whether or not you have all the KafkaTopic resources needed to recreate your cluster.

   Option 1: If you have all the KafkaTopic resources that existed before you lost your cluster, including internal topics such as committed offsets from __consumer_offsets:

   1. Recreate all KafkaTopic resources.

      It is essential that you recreate the resources before deploying the cluster, or the Topic Operator will delete the topics.

   2. Deploy the Kafka cluster.

      For example:

      oc apply -f kafka.yaml

      Copy to Clipboard Toggle word wrap

   Option 2: If you do not have all the KafkaTopic resources that existed before you lost your cluster:

   1. Deploy the Kafka cluster, as with the first option, but without the Topic Operator by removing the topicOperator property from the Kafka resource before deploying.

      If you include the Topic Operator in the deployment, the Topic Operator will delete all the topics.

   2. Delete the internal topic store topics from the Kafka cluster:

      oc run kafka-admin -ti --image=registry.redhat.io/amq-streams/kafka-35-rhel8:2.5.2 --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

      Copy to Clipboard Toggle word wrap

      The command must correspond to the type of listener and authentication used to access the Kafka cluster.

   3. Enable the Topic Operator by redeploying the Kafka cluster with the topicOperator property to recreate the KafkaTopic resources.

      For example:

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

      1

      
          #...

      Copy to Clipboard Toggle word wrap

   1

   Here we show the default configuration, which has no additional properties. You specify the required configuration using the properties described in the EntityTopicOperatorSpec schema reference.

7. Verify the recovery by listing the KafkaTopic resources:

   oc get KafkaTopic

   Copy to Clipboard Toggle word wrap

Procedure

1. Navigate in the OpenShift web console to Operators > Installed Operators.

2. For the installed AMQ Streams operator, select the options icon (three vertical dots) and click Uninstall Operator.

   The operator is removed from Installed Operators.

3. Navigate to Home > Projects and select the project where you installed AMQ Streams and the Kafka components.

4. Click the options under Inventory to delete related resources.

   Resources include the following:

   • Deployments
   • StatefulSets
   • Pods
   • Services
   • ConfigMaps
   • Secrets
   Tip

   Use the search to find related resources that begin with the name of the Kafka cluster. You can also find the resources under Workloads.

1. Delete the AMQ Streams subscription.

   oc delete subscription amq-streams -n openshift-operators

   Copy to Clipboard Toggle word wrap

2. Delete the cluster service version (CSV).

   oc delete csv amqstreams.<version>  -n openshift-operators

   Copy to Clipboard Toggle word wrap

3. Remove related CRDs.

   oc get crd -l app=strimzi -o name | xargs oc delete

   Copy to Clipboard Toggle word wrap

Procedure

1. Delete the Cluster Operator Deployment, related CustomResourceDefinitions, and RBAC resources.

   Specify the installation files used to deploy the Cluster Operator.

   oc delete -f install/cluster-operator

   Copy to Clipboard Toggle word wrap

2. Delete the resources you identified in the prerequisites.

   oc delete <resource_type> <resource_name> -n <namespace>

   Copy to Clipboard Toggle word wrap

   Replace <resource_type> with the type of resource you are deleting and <resource_name> with the name of the resource.

   Example to delete a secret

   oc delete secret my-cluster-clients-ca-cert -n my-project

   Copy to Clipboard Toggle word wrap