Overview

While working on the Data Import test using AWS MSK, Kafka messages we getting blocked in Kafka topic and we had to clear out messages from all DI topics. Messages can get stuck in a topic for many reasons such as consumers not having enough resources to consume all topics, the consumer goes down or AWS MSK Broker becomes unstable. Depending on the number of tenants per FOLIO instance, there can be thousands of topics per FOLIO instance. Instead of manually clearing out all messages from each of these topics, we decided to automate it to make this process faster.

To clear out messages from Kafka topic, there are 3 steps:

1. Get the list of topics to clear out messages

• • Download the latest Kafka release and extract it - https://kafka.apache.org/quickstart
  • Navigate to the bin directory kafka_2.12-2.7.0/bin/ - we will be using kafka-topics.sh and kafka-configs.sh scripts out of the box
  • Get list of topics and write to file topics.txt. For example, we want all topics for tenant tenant001

    
    

    Code Block
    language bash theme Midnight
    ./kafka-topics.sh --describe --zookeeper <zookeeper-plaintext-connection-url> --topic "imtc.Default.tenant001.*" | grep Configs | awk '{printf "%s\n", $2}' > topics.txt

    
    

       2. For all topics in the topics.txt file from step 1, alter the topic by adding custom config to set retention time to 1 second or less

#!/bin/bash -e

awk '{ system("./kafka-configs.sh --alter --bootstrap-server <bootstrap-plaintext-connection-url> --entity-type topics --entity-name="$1" --add-config retention.ms=" 1000) }' topics.txt

Verify config is updated

./kafka-topics.sh --describe --zookeeper <zookeeper-plaintext-connection-url> --topic "imtc.Default.tenant001.DI_COMPLETED"
Topic: imtc.Default.tenant001.DI_COMPLETED PartitionCount: 1 ReplicationFactor: 2 Configs: retention.ms=1000
Topic: imtc.Default.tenant001.DI_COMPLETED Partition: 0 Leader: 1 Replicas: 1,2 Isr: 2,1

       3. Revert all topics by deleting the custom config from step 2

#!/bin/bash -e

awk '{ system("./kafka-configs.sh --bootstrap-server <bootstrap-plaintext-connection-url> --entity-type topics --entity-name="$1" --alter --delete-config retention.ms") }' topics.txt

Verify config is deleted

./kafka-topics.sh --describe --zookeeper <zookeeper-plaintext-connection-url> --topic "imtc.Default.tenant001.DI_COMPLETED"
Topic: imtc.Default.tenant001.DI_COMPLETED PartitionCount: 1 ReplicationFactor: 2 Configs:
Topic: imtc.Default.tenant001.DI_COMPLETED Partition: 0 Leader: 1 Replicas: 1,2 Isr: 2,1