Kafka Cluster Management Procedures

11.1 Creating Topics for OCNADD

You can create topics (MAIN, SCP, SEPP, and NRF) using admin service, before starting data ingestion. For more details on topic and partitions see, "Kafka PVC Storage Requirements" section of Oracle Communications Network Analytics Data Director Benchmarking Guide.

To create a topic connect, to any worker node and send a POST curl request to the API Endpoint described below.

API Endpoint: <ClusterIP:Admin Port>/ocnadd-admin-svc/v1/topic

{
    "topicName":"<topicname>",
    "partitions":"3",
    "replicationFactor":"2",
    "retentionMs":"120000"
}

Note:

In case worker node access is not available then the adminservice Service-Type can be changed to LoadBalancer or NodePort in the admin service values.yaml (helm upgrade is required for any such changes).
For Loadbalancer service ensure that the admin port is not blocked in the cluster.

11.2 Adding Partitions to an Existing Topic

The topics in OCNADD must be created with the necessary partitions for the supported MPS during the deployment. In case the traffic load increases beyond the supported MPS, there may be the need to increase the number of partitions in the existing topics. This section describes the procedure to increase the number of partitions in the corresponding topic.

Caution:

The number of partitions cannot be decreased using this procedure. If partition reduction is required, the topic needs to be recreated with the desired number of partitions, in such case the complete data loss of the concerned topic is anticipated.

To add partitions to an existing topic:

Login to the bastion host.
Run the following command on any POD deployed in the OCNADD namespace:
```
kubectl exec -ti -n <namespace> <pod-name> -- bash
```
Example:
```
kubectl exec  -ti -n ocnadd-deploy ocnaddadminservice-5d6f8488d7-9hf6q -- bash
```
Note:
The procedure has been explained using the SCP topic name and https curls as an example. Modify the command with respect to the topic name you are updating. And use http in case intraTls is not enabled.
Describe the corresponding topic using the below command:
```
curl -k --location  "https://ocnaddadminservice:9181/ocnadd-admin-svc/v1/describe/topic/SCP"
```
The above command lists the topic details such as number of partitions, replication factor, retentionMs etc.
Add or increase partitions in the topic by running the following command:

Note:
Modify the "topicName" and "partitions" based on the requirements. For example, the number of partitions to be increased is 24 from the existing 12 of the SCP topic.
```
curl -v -k --location --request PUT "https://ocnaddadminservice:9181/ocnadd-admin-svc/v1/alter" --header 'Content-Type: application/json' \
--data-raw '{
    "topicName": "SCP",
    "partitions": "24"
}';  
```

Verify that the partitions have been added to the topic by running the following command:

curl -k --location  "https://ocnaddadminservice:9181/ocnadd-admin-svc/v1/describe/topic/SCP"

Example:

curl - k--location "https://ocnaddadminservice:9181/ocnadd-admin-svc/v1/describe/topic/SCP" {
    "topicName": "SCP",
    "numberOfPartitions": 24,
    "replicationFactor": 1,
    "partitions": [{
        "partitionId": 0,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 1,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 2,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }, {
        "partitionId": 3,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 4,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 5,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 6,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }, {
        "partitionId": 7,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 8,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 9,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 10,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }, {
        "partitionId": 11,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 12,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 13,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 14,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 15,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }, {
        "partitionId": 16,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 17,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 18,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 19,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }, {
        "partitionId": 20,
        "leaderBroker": "1003",
        "replicaList": [1003],
        "isr": [1003]
    }, {
        "partitionId": 21,
        "leaderBroker": "1004",
        "replicaList": [1004],
        "isr": [1004]
    }, {
        "partitionId": 22,
        "leaderBroker": "1001",
        "replicaList": [1001],
        "isr": [1001]
    }, {
        "partitionId": 23,
        "leaderBroker": "1002",
        "replicaList": [1002],
        "isr": [1002]
    }]
}

Exit from the POD(container).

11.3 Enabling Kafka Log Retention Policy

In Kafka, the log retention strategy determines how long the data is kept in the broker's logs before it is purged to free up storage space.

By default, the Log Retention parameters are not available in the OCNADD helm chart. The parameters are customizable and can be added in the 'ocnadd/charts/ocnaddkafka/templates/scripts-config.yaml' file of the OCNADD helm charts under the "Configuring Kafka broker". The Kafka Broker restart is expected.

There are two main approaches for log retention in Kafka:

Time-based retention:

Once the logs reach the specified age, they are considered eligible for deletion, and the broker will start a background task to remove the log segments that are older than the retention time. The time-based retention policy applies to all logs in a topic, including both the active logs that are being written and the inactive logs that have already been compacted.

The retention time is usually set using the "log.retention.hours" or "log.retention.minutes" configuration.

Parameters used:

log.retention.minutes=5 Default value for "log.retention.minutes" is set to 5 min.

Size-based retention:

Once the logs reach the specified size threshold, the broker will start a background task to remove the oldest log segments to ensure that the total log size remains below the specified limit. The size-based retention policy applies to all logs in a topic, including both the active logs that are being written and the inactive logs that have already been compacted.

Parameters for size-based retention:

#The maximum size of a single log file
log.segment.bytes=1073741824
 
#The maximum size of the log before deleting it
log.retention.bytes=32212254720
 
#Enable the log cleanup process to run on the server
log.cleaner.enable=true
 
#This is default cleanup policy. This will discard old segments when their retention time or size limit has been reached.
log.cleanup.policy=delete
 
#The interval at which log segments are checked to see if they can be deleted
log.retention.check.interval.ms=1000
 
#The amount of time to sleep when there are no logs to clean
log.cleaner.backoff.ms=1000
 
#The number of background threads to use for log cleaning
log.cleaner.threads=5

Calculate "log.retention.bytes":

The log retention size can be calculated as follows:

Example: For 80% threshold of the PVC claim size, "log.retention.bytes" are calculated as: (pvc(in bytes) / TotalPartition) * threshold/100)

where, TotalPartition: sum of partitions of all the topics.

If any topic has replication factor 2, then number of partitions will be twice the number of partitions in that topic. For more information about number of partitions per topic, see "Kafka PVC Storage Requirements" section of Oracle Communications Network Analytics Data Director Benchmarking Guide.

Note:

It is important to choose an appropriate time-based retention and size-based retention policy that balances the need for retaining data for downstream processing against the need to free up disk space. If the retention size or retention time is set too high, it may result in large amounts of disk space being consumed. If it is set to a very low value, then the important data may be lost.
Kafka allows for a combination of both time-based and size-based retention by setting both "log.retention.hours" or "log.retention.minutes" and "log.retention.bytes" configurations. The broker retains logs for the smaller value.
The "log.segment.bytes" is used to control the size of log segments in a topic's partition and is usually set to a relatively high value (for example, 1 GB) to reduce the number of segment files and minimize the overhead of file management. It is recommended to set this value lower based on smaller PVC size.

11.4 Expanding Kafka Storage

Given the rise in throughput demands within the DD, it becomes imperative to augment the storage allocation in Kafka correspondingly. In cases where the user had initially allocated storage for lower throughput, it is necessary to expand the storage to align with the updated throughput prerequisites.

To increase the size of existing PVC:

Check the StorageClass to which your PVC is attached.

Note:

PVC storage size can only be increased. It cannot be decreased.
It is mandatory to keep the same PVC storage size for all the Kafka brokers.

Run the following command to get the list of storage classes:

kubectl get sc

Sample output:

NAME                 PROVISIONER                  RECLAIMPOLICY   VOLUMEBINDINGMODE   ALLOWVOLUMEEXPANSION   AGE
occne-esdata-sc      rook-ceph.rbd.csi.ceph.com   Delete          Immediate           true                   125d
occne-esmaster-sc    rook-ceph.rbd.csi.ceph.com   Delete          Immediate           true                   125d
occne-metrics-sc     rook-ceph.rbd.csi.ceph.com   Delete          Immediate           true                   125d
standard (default)   rook-ceph.rbd.csi.ceph.com   Delete          Immediate           true                   125d

Run the following command to describe a storage class:

kubectl describe sc <storage_class_name>

For example:

kubectl describe sc standard

Sample output:

Name:                 standard
IsDefaultClass:       yes
Provisioner:          rook-ceph.rbd.csi.ceph.com
Parameter:            clusterID=rook-ceph,csi.storage.k8s.io/
AllowVolumeExpansion: True
MountOptions:
  discard
ReclaimPolicy:        Delete
VolumeBindingMode:    Immediate
Events:               <none>

The PVC should belong to the StorageClass which has "AllowVolumeExpansion" set as True:

AllowVolumeExpansion: True

Run the following command to list all the available PVCs:

kubectl get pvc -n <namespace>

For example:

kubectl get pvc -n ocnadd-deploy

Sample output:

NAME STATUS VOLUME CAPACITY ACCESS MODES STORAGECLASS AGE
backup-mysql-pvc Bound pvc-6e0c8366-fdaa-488e-a09d-f976e74f025b 20Gi RWO standard 8d
kafka-broker-security-zookeeper-0 Bound pvc-4f3c310c-1173-4d3d-824a-ddd2640ab028 5Gi RWO standard 8d
kafka-broker-security-zookeeper-1 Bound pvc-70fffae2-4f2a-4ecc-8f74-abfa003a8c1e 5Gi RWO standard 8d
kafka-broker-security-zookeeper-2 Bound pvc-e1b0b96d-e000-4e5c-a01c-d41cc8f3cfa4 5Gi RWO standard 8d
kafka-volume-kafka-broker-0 Bound pvc-48845cf6-1708-400a-808e-5b9b3cda7242 20Gi RWO standard 8d
kafka-volume-kafka-broker-1 Bound pvc-8acbde89-b223-4015-a170-ad6417b08be7 20Gi RWO standard 8d
kafka-volume-kafka-broker-2 Bound pvc-7b444cba-d1ab-496b-8b60-7c7599ef8754 20Gi RWO standard 8d
kafka-volume-kafka-broker-3 Bound pvc-e1584794-8e62-4f6a-8eff-c0c86e0864a1 20Gi RWO standard 8d
ocnadd-cache-volume-ocnaddcache-0 Bound pvc-5bf8ce9c-7ea6-4cf3-b51e-6ca94104dc5b 1Gi RWO standard 8d
ocnadd-cache-volume-ocnaddcache-1 Bound pvc-ddb0b53c-ec69-4e07-ac44-3ab98bee1d4c 1Gi RWO standard 8d

Run the following command to edit the required PVC and update the storage size:
```
kubectl edit pvc <pvc_name> -n <namespace>
```
For example:
```
kubectl edit pvc kafka-volume-kafka-broker-0 -n ocnadd-deploy
```
Sample output:
```
spec:
  accessModes:
  - ReadWriteOnce
  resources:
    requests:
      storage: <pvc_size>Gi  
```
Repeat steps 2 and 3 for each available Kafka PVC. Increase the size of the next Kafka broker PVC only after confirming that the current Kafka broker PVC's augmented size is reflected in the output of step 3. To increase the <pvc_size> for all the Kafka broker pods created during deployment, follow these steps.

To understand the storage requirement for your Kafka broker pods based on supported throughput, see "Kafka PVC-Storage Requirements" section in Oracle Communications Network Analytics Data Director Benchmarking Guide.

Run the following command to delete the stateful set:

kubectl delete statefulset --cascade=orphan <statefulset_name> -n <namespace>

In the ocnadd-custom-values-23.3.0.yaml file, update the PVC size (.Values.ocnaddkafka.ocnadd.kafkaBroker.pvcClaimSize) to match the value configured in step 4.
```
ocnaddkafka:
   ocnadd:
       ################
       # kafka-broker #
       ################
       kafkaBroker:
           name: kafka-broker
           replicas: 4
           pvcClaimSize: <pvc_size>Gi
```
Perform the Helm upgrade to recreate the stateful set of Kafka broker:
```
helm upgrade <release_name> <chart_name> -n <namespace> -f ocnadd-custom-values.yaml
```

11 Kafka Cluster Management Procedures

11.1 Creating Topics for OCNADD

11.2 Adding Partitions to an Existing Topic

11.3 Enabling Kafka Log Retention Policy

11.4 Expanding Kafka Storage