Как взять Kafka Topic Backup и восстановить?

Question

Мне нужно сделать резервную копию всех тем в Kafka в файл, названный в соответствующих именах тем, и мне нужно восстановить тему в соответствии с требованиями пользователя. Примечание. Этот сценарий необходимо запустить в среде Kerberized.
kafkabackup.sh

monyear=`date | awk '{print $2$6}'`
dat=`date| awk '{print $2$3$6}'`
export BACKUPDIR=/root/backup/$monyear
mkdir -p $BACKUPDIR
mkdir -p $BACKUPDIR/$dat
cd $BACKUPSDIR
BKDIR=$BACKUPDIR/$dat
##Log into Kafka

##Get topics from Kafka Broker

kinit -kt /etc/security/keytabs/kafka.service.keytab kafka/[e-mail]
cd /usr/hdp/current/kafka-broker/bin/
export KAFKA_CLIENT_KERBEROS_PARAMS="-Djava.security.auth.login.config=/etc/kafka/conf/kafka_client_jaas.conf"
./kafka-topics.sh --zookeeper adminnode.localdomain:2181 --list > $BKDIR/listtopics.txt

##Remove if any mark of deletion topics exists
sed -i.bak '/deletion/d' $BKDIR/listtopics.txt

## Starting kill script in parallel 

bash checkandkill.sh& 

##Reading the file contents for topics
for line in $(cat $BKDIR/listtopics.txt)
do
    echo $line
    ./test.sh --bootstrap-server node1.localdomain:6668 --topic $line  --consumer.config /home/kafka/conf.properties --from-beginning --security-protocol SASL_SSL > $BKDIR/$line
done

##Delete empty files

/usr/bin/find . -size 0 -delete

## Killing checkandkill daemon

ps -ef |grep -i checkandkill.sh| grep -v grep | awk '{print $2}' | xargs kill

exit

Когда потребитель бежит, он постоянно ждет получения сообщений. Нам нужно убить процесс.
checkandkill.sh

sleep 0.5m
for line in $(cat /root/backup/listtopics.txt)
do
    echo $line
    sleep 1m
    ps -ef |grep -i $line| grep -v grep | awk '{print $2}' | xargs kill
done

Нужна ваша помощь, чтобы завершить восстановление сценария.