Learn about the Kubernetes components required to run stateful Kafka Streams applications
Tags: tutorial, kubernetes, showdev, azure
Published: 2019-09-20 13:06:15

> Happy to get feedback via [@abhi_tweeter](https://twitter.com/abhi_tweeter) or just drop a comment!

One of the [previous blogs](https://dev.to/azure/learn-how-to-develop-a-kafka-streams-application-for-data-processing-and-deploy-it-to-kubernetes-25li) was about building a **stateless** stream processing application using the [Kafka Streams](https://kafka.apache.org/documentation/streams/) library and deploying it to Kubernetes as in the form of a `Deployment` object. 

In this part, we will continue exploring the powerful combination of Kafka Streams and Kubernetes. But this one is all about **stateful** applications and how to leverage specific Kubernetes primitives using [a Kubernetes cluster on Azure (AKS)](https://azure.microsoft.com/services/kubernetes-service/?WT.mc_id=devto-blog-abhishgu) to run it.

> I will admit right away that this is a slightly lengthy blog, but there are a lot of things to cover and learn!

As you go through this, you'll learn about the following:

- Kafka Streams
    - What is Kafka Streams?
    - Concepts of stateful Kafka Streams applications
- Behind-the-scenes
    - What's going on in the Java code for stream processing logic using Kafka Streams
    - Kubernetes components for running Stateful Kafka Streams apps such `StatefulSet`, Volume Claim templates and other configuration parameters like `Pod` anti-affinity
    - How is this all setup using [Azure Kubernetes Service](https://docs.microsoft.com/azure/aks/?WT.mc_id=devto-blog-abhishgu) for container orchestration and [Azure Disk](https://docs.microsoft.com/azure/virtual-machines/windows/managed-disks-overview?WT.mc_id=devto-blog-abhishgu) for persistence storage
- How to set up and configure a [Docker container registry](https://docs.microsoft.com/azure/container-registry/?WT.mc_id=devto-blog-abhishgu) and Azure Kubernetes cluster 
- How to build & deploy our app to Kubernetes and finally test it out using the Kafka CLI

> The source code is [on GitHub](https://github.com/abhirockzz/kafka-streams-stateful-kubernetes)

Let's get started!

![](https://media.giphy.com/media/9DGFdfnT6SJSE/giphy.gif)

### Pre-requisites: 

If you don't have it already, please [install the Azure CLI](https://docs.microsoft.com/cli/azure/install-azure-cli?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu) and [kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/). The stream processing app is written in Java and uses [Maven](https://maven.apache.org/install.html). You will also need [Docker](https://docs.docker.com/install/) to build the app container image.

> This tutorial assumes you have a Kafka cluster which is reachable from your Kubernetes cluster on Azure

### Kafka Streams

This section will provide a quick overview of Kafka Streams and what "state" means in the context of Kafka Streams based applications.

#### Overview of Kafka Streams

It is a simple and lightweight client library, which can be easily embedded in any Java app or microservice, where the input and output data are stored in [Kafka clusters](https://kafka.apache.org/). It has no external dependencies on systems other than Kafka itself and it’s partitioning model to horizontally scale processing while maintaining strong ordering guarantees. It has support for fault-tolerant local state, employs one-record-at-a-time processing to achieve millisecond processing latency and offers necessary stream processing primitives, along with a [high-level Streams DSL](https://kafka.apache.org/23/documentation/streams/developer-guide/dsl-api.html) and a [low-level Processor API](https://kafka.apache.org/23/documentation/streams/developer-guide/processor-api.html). The combination of "state stores" and [Interactive queries](https://kafka.apache.org/23/documentation/streams/developer-guide/interactive-queries.html) allow you to leverage the state of your application from outside your application.

#### Stateful Kafka Streams app

Most stream processing apps need contextual data i.e. state in order e.g. to maintain a running count of items in an inventory, you're going to need the last "count" in order to calculate the "current" count. 

You can deploy multiple Kafka Streams app instances to scale your processing. Since each instance churns data from one or more partitions (of a Kafka topic), the state associated with each instance is stored locally (unless you're the [`GlobalKTable` API](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/GlobalKTable.html) - deserves a dedicated blog post!). Kafka Streams supports "stateful" processing with the help of state stores. Typically, it is file-system based (Kafka Streams uses an embedded [RocksDB database](http://rocksdb.org/) internally) but you also have the option of using an in-memory hash-map, or use the pluggable nature of the [Kafka Streams Processor API](https://kafka.apache.org/23/documentation/streams/developer-guide/processor-api.html) to build a custom implementation a state store. 

In addition to storing the state, Kafka Streams has built-in mechanisms for fault-tolerance of these state stores. The contents of each state store are backed-up to a replicated, [log-compacted Kafka topic](https://kafka.apache.org/documentation/#compaction). If any of your Kafka Streams app instance fails, another one can come up, restore the current state from Kafka and continue processing. In addition to storing state, you can also "query" these state stores. That's a topic for another blog post altogether - stay tuned!

> Please note that it is possible to tune the "fault tolerance" behavior i.e. you can choose not to back-up your local state store to Kafka

Before you dive in, here is a high level overview of the solution

![](https://thepracticaldev.s3.amazonaws.com/i/l0gunt1pz1g2wnp52bsh.jpg)

### Behind the scenes

Let's look at what the stream processing code is upto and then dive into some of the nitty-gritty of the Kubernetes primitives and what value they offer when running "stateful" Kafka Streams apps.

#### Stream processing code

The processing pipeline executes something similar to the canonical "word count". It makes use of the [high-level Streams DSL](https://kafka.apache.org/23/documentation/streams/developer-guide/dsl-api.html) API:

- receives a stream of key-value pairs from an input/source Kafka topic e.g. `foo:bar`, `john:doe`, `foo:bazz` etc.
- keeps and stores a count of the keys (ignores the values) e.g. `foo=2`, `john=1` etc.
- forwards the count to an output Kafka topic (sink)

Please note that the latest Kafka Streams library version at the time of writing was `2.3.0` and that's what the app uses

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>2.3.0</version>
        </dependency>

We start off with an instance of `StreamsBuilder` and invoke it's `stream` method to hook on to the source topic. What we get is a [`KStream` object](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KStream.html) which is a representation of the continuous stream of records sent to the topic. 

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> inputStream = builder.stream(INPUT_TOPIC);
        
We use [`groupByKey`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KStream.html#groupByKey--) on the input `KStream` to group the records by their current key into a [`KGroupedStream`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KGroupedStream.html). In order to keep a count of the keys, we use the `count` method (not a surprise!). We also ensure that the word count i.e. `foo=5`, `bar=3` etc. is persisted to a state store - [`Materialized`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/Materialized.html) is used to describe how that state store should be persisted. In this case, a specific name is chosen and the exact location on disk is mentioned in the KafkaStreams configuration as such: `configurations.put(StreamsConfig.STATE_DIR_CONFIG, "/data/count-store");`

> the default behavior is to store the state on disk using RocksDB unless configured differently

        inputStream.groupByKey()
                   .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("count-store")
                        .withKeySerde(Serdes.String())
                        .withValueSerde(Serdes.Long()))

Finally, for ease of demonstration, we convert the [`KTable`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KTable.html) (created by `count`) back to a `KStream` using `toStream`, convert the `java.lang.Long` (that's the `count` data type) into a `String` using [`mapValues`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/kstream/KStream.html#mapValues-org.apache.kafka.streams.kstream.ValueMapper-) and pass on the result to an output topic. This just for easy consumption in the Kafka CLI so that you're able to actually see the final count of each of the words.

                .toStream()
                .mapValues(new ValueMapper<Long, String>() {
                    @Override
                    public String apply(Long v) {
                        return String.valueOf(v);
                    }
                })
                .to(OUTPUT_TOPIC);

That's all in terms of setting up the stream and defining the logic. We create a [`Topology`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/Topology.html) object using the `build` method in `StreamsBuilder` and use this object to create a [`KafkaStreams`](https://kafka.apache.org/23/javadoc/org/apache/kafka/streams/KafkaStreams.html) instance which is a representation of our application itself. We start the stream processing using the `start` method

        Topology topology = builder.build();
        KafkaStreams streamsApp = new KafkaStreams(topology, getKafkaStreamsConfig());
        streamsApp.start();

The `getKafkaStreamsConfig()` is just a helper method which creates a [`Properties`](https://docs.oracle.com/javase/8/docs/api/java/util/Properties.html?is-external=true) object which contains Kafka Streams specific configuration, including Kafka broker endpoint etc.

    static Properties getKafkaStreamsConfig() {
        String kafkaBroker = System.getenv().get(KAFKA_BROKER_ENV_VAR);
        Properties configurations = new Properties();
        configurations.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker + ":9092");
        configurations.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
        configurations.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        configurations.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        configurations.put(StreamsConfig.REQUEST_TIMEOUT_MS_CONFIG, "20000");
        configurations.put(StreamsConfig.RETRY_BACKOFF_MS_CONFIG, "500");
        configurations.put(StreamsConfig.STATE_DIR_CONFIG, STATE_STORE_DIR);

        return configurations;
    }

#### Kubernetes primitives

So far so good! We have the Kafka Streams app churning out word counts and storing them. We can simply run this as a Kubernetes `Deployment` (as demonstrated in the previous blog), but there are some benefits to be gained by using something called a [`StatefulSet`](https://kubernetes.io/docs/concepts/workloads/controllers/statefulset/).

> `StatefulSet` is a topic that deserves a blog (or more!) by itself. The goal is not to teach you everything about Kubernetes `StatefulSet`s in this blog, but provide enough background and demonstrate how its features can be leveraged for stateful Kafka Streams apps. 

![](https://media.giphy.com/media/26Ec4bWMsI2fFUqxG/giphy.gif)


**StatefulSet: the What**

Here is a gist what `StatefulSet`s offer for running stateful workloads in Kubernetes

- `Pod` uniqueness - Each `Pod` in a `StatefulSet` is unique and this maintained across restarts, re-scheduling etc. This also applies to networking and communication (inter-Pod or external)
- Persistent storage - using a Volume Claim template, you can request for storage allocation for each `Pod` in a `StatefulSet` such that there is one to one mapping b/w the `Pod` and the storage medium
- Managed lifecycle - You can be explicit about how to manage the lifecycle of the Pods across various stages including starts, updates, deletion. `StatefulSet` Pods can be configured to handle this is an ordered manner.

> All these are in stark contrast to general `Deployment`s which handle Pods as disposable entities with no identity, the concept of "stable" attached storage or ordered lifecycle management.

**StatefulSet: the Why**

Let's explore the motivation behind why we want to use `StatefulSet`s for this specific scenario i.e. a stateful Kafka Streams app.

As previously mentioned, you can run multiple Kafka Streams app instances. Each instance processes data from one or more partitions (of a topic) and stores the associated state locally. Fault tolerance and resiliency is also built into Kafka Streams app because the contents of each state store is backed-up to a replicated, log-compacted Kafka topic. If any of your Kafka Streams app instance fails, another one can come up, restore the current state from Kafka and continue processing.

Now here is the catch. Any serious application with a reasonably complex topology and processing pipeline will generate a lot of "state". In such as case, regular app operations like scale-out or anomalies such as crashes etc. will trigger the process of restore/refresh state from the Kafka back-up topic. This can be costly in terms of time, network bandwidth etc. Using `StatefulSet`, we can ensure that each `Pod` will always have a stable storage medium attached to it and this will be stable (not change) over the lifetime of the `StatefulSet`. This means that after restarts, upgrades etc. (most of) the state is already present locally on the disk and the app only needs to fetch the "delta" state from the Kafka topics (if needed). This implies that state recovery time will be much smaller or may not even be required in few cases.

> In this example, we will be making use of the first two features of `StatefulSet` i.e. Pod uniqueness and stable Persistent Storage.

**StatefulSet: the How**

It's time to see how it's done. Let's start by exploring the Kubernetes YAML manifest (in small chunks) for our application - we will later use this to deploy the app to AKS

We define the name of our `StatefulSet` (`kstreams-count`) and refer to a [`Headless` Service](https://kubernetes.io/docs/concepts/services-networking/service/#headless-services)(`kstreams-count-service`) which is responsible for the unique network identity - it is bundled along with `StatefulSet` itself.

    apiVersion: apps/v1
    kind: StatefulSet
    metadata:
        name: kstreams-count
    spec:
        serviceName: "kstreams-count-service"

> The `Headless` Service should be created before the `StatefulSet`

    apiVersion: v1
    kind: Service
    metadata:
    name: kstreams-count-service
    labels:
        app: kstreams-count
    spec:
    clusterIP: None
    selector:
        app: kstreams-count

The `Pod` specification (`spec.containers`) points to the Docker image and defines the environment variable `KAFKA_BROKER` which which will be injected within our app at runtime.

    spec:
      containers:
      - name: kstreams-count
        image: <to be filled>.azurecr.io/kstreams-count:latest
        env:
          - name: KAFKA_BROKER
            value: [to be filled]

In addition to the above, the container spec also defines the persistent storage. In this case, it means that the container will use a stable storage to store the contents in the specified path which in this case is `/data/count-store` (recall that this is the local state directory as configured in our Kafka Streams app)

    volumeMounts:
        - name: count-store
          mountPath: /data/count-store

How is this persistent storage going to come to life and made available to the `Pod`? The answer lies in a Volume Claim Template specified as a part of the [`StatefulSet` spec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.15/#statefulsetspec-v1-apps). One [`PersistentVolumeClaim`](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.15/#persistentvolumeclaim-v1-core) and  [`PersistentVolume`](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.15/#persistentvolume-v1-core) will be created for each Volume Claim Template. 

    volumeClaimTemplates:
    - metadata:
        name: count-store
        spec:
        accessModes: [ "ReadWriteOnce" ]
        resources:
            requests:
            storage: 1Gi

**So how does the storage medium get created?**

This is powered by **Dynamic Provisioning** which enables storage volumes to be created on-demand. Otherwise, cluster administrators have to manually provision cloud based storage and then create equivalent `PersistentVolume` objects in Kubernetes. Dynamic provisioning eliminates this by automatically provisioning storage when it is requested by users. 

Dynamic provisioning itself uses a `StorageClass` which provides a way to describe the type of storage using a set of parameters along with a volume plugin which actually takes care of the storage medium provisioning. Azure Kubernetes Service makes dynamic provisioning easy by including two pre-seeded storage classes:

- `default` storage class: provisions a standard [Azure Disk backed by a Standard HDD](https://docs.microsoft.com/azure/virtual-machines/windows/disks-types?WT.mc_id=devto-blog-abhishgu#standard-hdd)
- `managed-premium` storage class: provisions a premium [Azure Disk backed by Premium SSD](https://docs.microsoft.com/azure/virtual-machines/windows/disks-types?WT.mc_id=devto-blog-abhishgu#premium-ssd)

You can check the same by running `kubectl get storageclass` command

    NAME                PROVISIONER                AGE
    default (default)   kubernetes.io/azure-disk   6d10h
    managed-premium     kubernetes.io/azure-disk   6d10h

> Note that `kubernetes.io/azure-disk` is the volume plugin (provisioner implementation)

Since, we don't have an explicit `StorageClass` defined in the volume claim template, so the default `StorageClass` will be used. For each instance of your Kafka Streams app, an Azure Disk instance will be created and mounted into the `Pod` representing the app.

Finally, we use Pod anti-affinity (nothing to do with `StatefulSet`) - this is to ensure that no two instances of our app are located on the same node.  

      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
          - labelSelector:
              matchExpressions:
              - key: app
                operator: In
                values:
                - kstreams-count
            topologyKey: "kubernetes.io/hostname"


Let's move on to the infrastructure setup.

### AKS cluster setup

You need a single command to stand up a Kubernetes cluster on Azure. But, before that, we'll have to create a resource group

    export AZURE_SUBSCRIPTION_ID=[to be filled]
    export AZURE_RESOURCE_GROUP=[to be filled]
    export AZURE_REGION=[to be filled] (e.g. southeastasia)

Switch to your subscription and invoke `az group create`

    az account set -s $AZURE_SUBSCRIPTION_ID
    az group create -l $AZURE_REGION -n $AZURE_RESOURCE_GROUP

You can now invoke `az aks create` to create the new cluster

> To keep things simple, the below command creates a two node cluster. Feel free to change the specification as per your requirements

    export AKS_CLUSTER_NAME=[to be filled]

    az aks create --resource-group $AZURE_RESOURCE_GROUP --name $AKS_CLUSTER_NAME --node-count 2 --node-vm-size Standard_B2s --node-osdisk-size 30 --generate-ssh-keys

Get the AKS cluster credentials using `az aks get-credentials` - as a result, `kubectl` will now point to your new cluster. You can confirm the same

    az aks get-credentials --resource-group $AZURE_RESOURCE_GROUP --name $AKS_CLUSTER_NAME
    kubectl get nodes

> If you are interested in learning Kubernetes and Containers using [Azure](https://azure.microsoft.com/services/kubernetes-service/?WT.mc_id=devto-blog-abhishgu), simply [create a **free** account](https://azure.microsoft.com/en-us/free/?WT.mc_id=devto-blog-abhishgu) and get going! A good starting point is to use the [quickstarts, tutorials and code samples](https://docs.microsoft.com/azure/aks/?WT.mc_id=devto-blog-abhishgu) in the documentation to familiarize yourself with the service. I also highly recommend checking out the [50 days Kubernetes Learning Path](https://azure.microsoft.com/resources/kubernetes-learning-path/?WT.mc_id=devto-blog-abhishgu). Advanced users might want to refer to [Kubernetes best practices](https://docs.microsoft.com/azure/aks/best-practices?WT.mc_id=devto-blog-abhishgu) or the watch some of the [videos](https://azure.microsoft.com/resources/videos/index/?services=kubernetes-service&WT.mc_id=devto-blog-abhishgu) for demos, top features, and technical sessions.

### Setup Azure Container Registry

Simply put, [Azure Container Registry](https://azure.microsoft.com/services/container-registry/?WT.mc_id=devto-blog-abhishgu) (`ACR` in short) is a managed private Docker registry in the cloud which allows you to build, store, and manage images for all types of container deployments.

Start by creating an ACR instance

    export ACR_NAME=[to be filled]
    az acr create --resource-group $AZURE_RESOURCE_GROUP --name $ACR_NAME --sku Basic

> valid SKU values - `Basic`, `Classic`, `Premium`, `Standard`. See [command documentation](https://docs.microsoft.com/cli/azure/acr?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-acr-create)

### Configure ACR to work with AKS

To access images stored in ACR, you must grant the AKS service principal the correct rights to pull images from ACR.

Get the `appId` of the service principal which is associated with your AKS cluster

    AKS_SERVICE_PRINCIPAL_APPID=$(az aks show --name $AKS_CLUSTER_NAME --resource-group $AZURE_RESOURCE_GROUP --query servicePrincipalProfile.clientId -o tsv)

Find the ACR resource ID

    ACR_RESOURCE_ID=$(az acr show --resource-group $AZURE_RESOURCE_GROUP --name $ACR_NAME --query "id" --output tsv)

Grant `acrpull` permissions to AKS service principal

    az role assignment create --assignee $AKS_SERVICE_PRINCIPAL_APPID --scope $ACR_RESOURCE_ID --role acrpull

For some more details on this topic, check out one of my previous blog
 
{% link https://dev.to/azure/azure-tip-how-to-get-your-kubernetes-cluster-service-principal-and-use-it-to-access-other-azure-services-2735 %}

Alright, our AKS cluster along with ACR is ready to use!

### From your laptop to a Docker Registry in the cloud

Clone the GitHub repo, change to the correct directory and build the application JAR

    git clone https://github.com/abhirockzz/kafka-streams-stateful-kubernetes
    cd kafka-streams-stateful-kubernetes
    mvn clean install

> You should see `kstreams-count-statefulset-1.0.jar` in the `target` directory

Here is the `Dockerfile` for our stream processing app

    FROM openjdk:8-jre
    WORKDIR /
    COPY target/kstreams-count-statefulset-1.0.jar /
    CMD ["java", "-jar","kstreams-count-statefulset-1.0.jar"]

We will now build a Docker image ...

    export DOCKER_IMAGE=kstreams-count
    export ACR_SERVER=$ACR_NAME.azurecr.io
    docker build -t $DOCKER_IMAGE .

... and push it to [Azure Container Registry](https://azure.microsoft.com/services/container-registry/?WT.mc_id=devto-blog-abhishgu)

    az acr login --name $ACR_NAME
    docker tag $DOCKER_IMAGE $ACR_SERVER/$DOCKER_IMAGE
    docker push $ACR_SERVER/$DOCKER_IMAGE

Once this is done, you can confirm using `az acr repository list`

    az acr repository list --name $ACR_NAME --output table

### Deploy to Kubernetes

To deploy and confirm

    kubectl apply -f kstreams-count-statefulset.yaml
    kubectl get pods -l=app=kstreams-count

> The app will take some time to start up since this also involves storage (Azure Disk) creation and attachment. After some time, you should see two pods in the `Running` state

### The moment of truth!

It's time to test our end to end flow. Just to summarize:

- you will produce data to the input Kafka topic (`input-topic`) using the Kafka CLI locally
- the stream processing application in AKS will churn the data, store state and put it back to another Kafka topic
- your local Kafka CLI based consumer process will get that data from the output topic (`counts-topic`)

Let' create the Kafka topics first

    export KAFKA_HOME=[kafka installation directory]
    export INPUT_TOPIC=input-topic
    export OUTPUT_TOPIC=counts-topic

    $KAFKA_HOME/bin/kafka-topics.sh --create --topic $INPUT_TOPIC --partitions 4 --replication-factor 1 --bootstrap-server $KAFKA_BROKER
    $KAFKA_HOME/bin/kafka-topics.sh --create --topic $OUTPUT_TOPIC --partitions 4 --replication-factor 1 --bootstrap-server $KAFKA_BROKER

    $KAFKA_HOME/bin/kafka-topics.sh --list --bootstrap-server $KAFKA_BROKER


#### Start consumer process

    export KAFKA_HOME=[kafka installation directory]
    export KAFKA_BROKER=[kafka broker e.g. localhost:9092]
    export OUTPUT_TOPIC=counts-topic
    
    $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server 
    $KAFKA_BROKER --topic $OUTPUT_TOPIC --from-beginning --property "print.key=true"

#### Start producer process  (different terminal)

    export KAFKA_HOME=[kafka installation directory]
    export KAFKA_BROKER=[kafka broker e.g. localhost:9092]
    export INPUT_TOPIC=input-topic

    $KAFKA_HOME/bin/kafka-console-producer.sh --broker-list $KAFKA_BROKER --topic $INPUT_TOPIC

You will get a prompt and you can start entering values e.g.

    > foo:bar
    > hello:world
    > hello:universe
    > foo:baz
    > john:doe

In the consumer terminal, you should see the words and their respective counts e.g. `foo 2`, `hello 2`, `john 1` etc.

With the sanity testing out of the way...

#### ... let's look at the state of AKS cluster.

Check the `PersistentVolumeClaim`s (PVC) and `PersistentVolume`s (PV) - you will two separate set of PVC-PV pairs.

    kubectl get pv
    kubectl get pvc

The creation of `PersistentVolume`s means that Azure Disks were created as well. To check them, let's get AKS node resource group first

    AKS_NODE_RESOURCE_GROUP=$(az aks show --resource-group abhishgu-aks --name abhishgu-aks --query nodeResourceGroup -o tsv)

Assuming there this is a two node AKS cluster we will get back four disks - one each for the two nodes and one each of for two of our app instances

    az disk list -g $AKS_NODE_RESOURCE_GROUP

> You will notice that name of disk is the same as that of the PVC

Let's dig into the file system of the `Pod` which is running our application. `kstreams-count-0` is the name of one such instance (yes, the name is deterministic, thanks to `StatefulSet`). Recall that we specified `/data/count-store` as the state store directory in our app as well the as `volumeMounts` section of the app manifest - let's peek into that directory. 

    kubectl exec -it kstreams-count-0 -- ls -lrt /data/count-store/counts-app

You will notice that the state data split across multiple sub-directories whose number will be equal to the number of topic partitions which the app instance is handling e.g. if you have four partitions and two instances, each of them will handle data from two partitions each

    total 24
    drwxr-xr-x 3 root root 4096 Sep 16 11:58 0_0
    drwxr-xr-x 3 root root 4096 Sep 16 12:02 0_1

> you can repeat the same process for the second instance i.e. `kstreams-count-1`

If you list the number of topics using the Kafka CLI, you should also see a topic named `counts-app-counts-store-changelog`. This is the back-up, log-compacted changelog topic which we discussed earlier

> the name format is `<application_name>-<state_store_name>-changelog`

### Clean up

Start by deleting the `StatefulSet` and associated `Headless` Service

    kubectl delete -f kstreams-count-statefulset.yaml

The `PersistentVolume`s associated with the `PersistentVolumeClaim`s are not deleted automatically

    kubectl delete pvc

This will trigger the deletion of the `PersistentVolume`s and the corresponding Azure Disks. You can confirm the same

    kubectl get pv
    az disk list -g $AKS_NODE_RESOURCE_GROUP

Finally, to clean up your AKS cluster, ACR instance and related resources

    az group delete --name $AZURE_RESOURCE_GROUP --yes --no-wait



That's all for this blog! If you found it helpful, please like and follow :-)