// Module included in the following assembly:
//
// assembly-cluster-recovery-volume.adoc

[id="cluster-recovery-volume_{context}"]
= Recovering a deleted cluster from persistent volumes

This procedure describes how to recover a deleted cluster from persistent volumes (PVs).

In this situation, the Topic Operator identifies that topics exist in Kafka, but the `KafkaTopic` resources do not exist.

When you get to the step to recreate your cluster, you have two options:

. 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.

. 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.

NOTE: If the Topic Operator is not deployed, you only need to recover the `PersistentVolumeClaim` (PVC) resources.

.Before you begin

In this procedure, it is essential that PVs are mounted into the correct PVC to avoid data corruption.
A `volumeName` is specified for the PVC and this must match the name of the PV.

For more information, see:

* xref:ref-persistent-storage-{context}#pvc-naming[Persistent Volume Claim naming]
* xref:ref-jbod-storage-{context}#jbod-pvc[JBOD and Persistent Volume Claims]

NOTE: The procedure does not include recovery of `KafkaUser` resources, which must be recreated manually.
If passwords and certificates need to be retained, secrets must be recreated before creating the `KafkaUser` resources.

.Procedure

. Check information on the PVs in the cluster:
+
[source,shell,subs="+quotes,attributes"]
----
kubectl get pv
----
+
Information is presented for PVs with data.
+
Example output showing columns important to this procedure:
+
[source,shell,subs="+quotes,attributes"]
----
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
----
+
* _NAME_ shows the name of each PV.
* _RECLAIM POLICY_ shows that PVs are _retained_.
* _CLAIM_ shows the link to the original PVCs.

. Recreate the original namespace:
+
[source,shell,subs="+quotes,attributes"]
----
kubectl create namespace _myproject_
----

. Recreate the original PVC resource specifications, linking the PVCs to the appropriate PV:
+
For example:
+
[source,shell,subs="+quotes,attributes"]
----
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*
----

. Edit the PV specifications to delete the `claimRef` properties that bound the original PVC.
+
For example:
+
[source,shell,subs="+quotes,attributes"]
----
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
----
+
In the example, the following properties are deleted:
+
[source,shell,subs="+quotes,attributes"]
----
claimRef:
  apiVersion: v1
  kind: PersistentVolumeClaim
  name: data-0-my-cluster-kafka-2
  namespace: myproject
  resourceVersion: "39113"
  uid: 54be1c60-3319-11ea-97b0-0aef8816c7ea
----

. Deploy the Cluster Operator.
+
[source,shell,subs="+quotes,attributes"]
----
kubectl create -f install/cluster-operator -n _my-project_
----

. 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`:

. Recreate all `KafkaTopic` resources.
+
It is essential that you recreate the resources before deploying the cluster, or the Topic Operator will delete the topics.

. Deploy the Kafka cluster.
+
For example:
+
[source,shell,subs="+quotes,attributes"]
----
kubectl apply -f _kafka.yaml_
----
--
+
--
*_Option 2_*: If you do not have all the `KafkaTopic` resources that existed before you lost your cluster:

. 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.

. Delete the internal topic store topics from the Kafka cluster:
+
[source,shell,subs="+attributes"]
----
kubectl run kafka-admin -ti --image={DockerKafkaImageCurrent} --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
----
+
The command must correspond to the type of listener and authentication used to access the Kafka cluster.

. Enable the Topic Operator by redeploying the Kafka cluster with the `topicOperator` property to recreate the `KafkaTopic` resources.
+
For example:
+
[source,shell,subs="+quotes,attributes"]
----
apiVersion: {KafkaApiVersion}
kind: Kafka
metadata:
  name: my-cluster
spec:
  #...
  entityOperator:
    *topicOperator: {}* <1>
    #...
----
--
<1> Here we show the default configuration, which has no additional properties.
You specify the required configuration using the properties described in xref:type-EntityTopicOperatorSpec-reference[].

. Verify the recovery by listing the `KafkaTopic` resources:
+
[source,shell,subs="+quotes,attributes"]
----
kubectl get KafkaTopic
----
