fix: kafka backup using sasl

addons/kafka/dataprotection/backup.sh

@@ -1,13 +1,31 @@
 #!/bin/bash
 
+# if the script exits with a non-zero exit code, touch a file to indicate that the backup failed,
+# the sync progress container will check this file and exit if it exists
+function handle_exit() {
+  exit_code=$?
+  if [ $exit_code -ne 0 ]; then
+    echo "failed with exit code $exit_code"
+    touch "${DP_BACKUP_INFO_FILE}.exit"
+    exit $exit_code
+  fi
+}
+
+trap handle_exit EXIT
+
 # topics.txt format is like:
 # (topic name)             (partitions)   (replication factor)
 # topic1                   1              1
 # topic2                   1              1
 #
 # We also ignores the __consumer_offsets topic as offsets won't be backuped up.
 echo "getting topics..."
-kafkactl get topics | tail -n +2 | grep -v __consumer_offsets | datasafed push - topics.txt
+topic_list=$(kafkactl get topics | tail -n +2)
+if [[ -z $topic_list ]]; then
+  echo "nothing to backup"
+  exit 1
+fi
+echo $topic_list | grep -v __consumer_offsets | datasafed push - topics.txt
 readarray -t topics < <(kafkactl get topics -o compact | grep -v  __consumer_offsets)
 
 for topic in "${topics[@]}"; do

addons/kafka/dataprotection/common.sh

@@ -26,3 +26,15 @@ function DP_save_backup_status_info() {
 export BROKERS="$DP_DB_HOST:$DP_DB_PORT"
 export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
 export DATASAFED_BACKEND_BASE_PATH=${DP_BACKUP_BASE_PATH}
+
+if [[ $KB_KAFKA_SASL_ENABLE == "true" ]]; then
+  echo "using sasl auth.."
+  if [[ $KB_KAFKA_SASL_MECHANISMS != *"PLAIN"* ]]; then
+    echo "unsupported KB_KAFKA_SASL_MECHANISMS: $KB_KAFKA_SASL_MECHANISMS"
+    exit 1
+  fi
+  export SASL_ENABLED="true"
+  export SASL_MECHANISM="plaintext"
+  export SASL_USERNAME=$KAFKA_ADMIN_USER
+  export SASL_PASSWORD=$KAFKA_ADMIN_PASSWORD
+fi