diff --git a/addons/kafka/configs/kafka-server.prop.tpl b/addons/kafka/configs/kafka-server.prop.tpl
index ae0fa2a6e..c04008258 100644
--- a/addons/kafka/configs/kafka-server.prop.tpl
+++ b/addons/kafka/configs/kafka-server.prop.tpl
@@ -225,7 +225,7 @@ password.encoder.key.length=128
 password.encoder.iterations=4096
 
 # SSL Keystore of an Existing Listener
-ssl.keystore.type=JKS
+# ssl.keystore.type=JKS
 # ssl.keystore.location=
 # ssl.keystore.password=
 # ssl.key.password=
@@ -233,7 +233,7 @@ ssl.keystore.type=JKS
 # ssl.keystore.certificate.chain=
 
 # SSL Truststore of an Existing Listener
-ssl.truststore.type=JKS
+# ssl.truststore.type=JKS
 # ssl.truststore.location=
 # ssl.truststore.password=
 # ssl.truststore.certificates=
diff --git a/addons/kafka/dataprotection/backup.sh b/addons/kafka/dataprotection/backup.sh
index 3a8e68ddd..9199f6fd9 100644
--- a/addons/kafka/dataprotection/backup.sh
+++ b/addons/kafka/dataprotection/backup.sh
@@ -23,7 +23,8 @@ echo "getting topics..."
 topic_list=$(kafkactl get topics | tail -n +2)
 if [[ -z $topic_list ]]; then
   echo "nothing to backup"
-  exit 1
+  DP_save_backup_status_info 0
+  exit 0
 fi
 echo $topic_list | grep -v __consumer_offsets | datasafed push - topics.txt
 readarray -t topics < <(kafkactl get topics -o compact | grep -v  __consumer_offsets)
diff --git a/addons/kafka/dataprotection/common.sh b/addons/kafka/dataprotection/common.sh
index 6c07959cd..1f9f4d3ad 100644
--- a/addons/kafka/dataprotection/common.sh
+++ b/addons/kafka/dataprotection/common.sh
@@ -23,7 +23,10 @@ function DP_save_backup_status_info() {
   fi
 }
 
-export BROKERS="$DP_DB_HOST:$DP_DB_PORT"
+# don't let kb's env affect kafkactl's config
+export TLS_ENABLED="false"
+# we'll use the internal listener to avoid using ssl
+export BROKERS="$DP_DB_HOST:9094"
 export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
 export DATASAFED_BACKEND_BASE_PATH=${DP_BACKUP_BASE_PATH}
 
diff --git a/addons/kafka/dataprotection/restore.sh b/addons/kafka/dataprotection/restore.sh
index 112dfb881..07975da11 100644
--- a/addons/kafka/dataprotection/restore.sh
+++ b/addons/kafka/dataprotection/restore.sh
@@ -1,6 +1,10 @@
 #!/bin/bash
 
 echo "getting topics..."
+if [[ -z $(datasafed list topics.txt) ]]; then
+  echo "restore from an empty backup! doing nothing..."
+  exit 0
+fi
 readarray -t lines < <(datasafed pull topics.txt -)
 for line in "${lines[@]}"; do
   read -r topic partitions replication <<< "$line"
diff --git a/addons/kafka/scripts/common.sh b/addons/kafka/scripts/common.sh
index 7243f990a..9c4a6506f 100644
--- a/addons/kafka/scripts/common.sh
+++ b/addons/kafka/scripts/common.sh
@@ -33,8 +33,13 @@ build_zk_server_sasl_properties() {
         INTER_BROKER_PROTOCOL=${KB_KAFKA_SASL_INTER_BROKER_PROTOCOL}
     fi
 
-    export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT
-    echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    if [[ "$TLS_ENABLED" == "true" ]]; then
+        export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP="INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_SSL"
+        echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    else
+        export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP="INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT"
+        echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    fi
     export KAFKA_CFG_SASL_ENABLED_MECHANISMS="${ENABLED_MECHANISMS}"
     echo "[sasl]export KAFKA_CFG_SASL_ENABLED_MECHANISMS=${KAFKA_CFG_SASL_ENABLED_MECHANISMS}"
     export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="${INTER_BROKER_PROTOCOL}"
@@ -50,8 +55,13 @@ build_kraft_server_sasl_properties() {
         INTER_BROKER_PROTOCOL=${KB_KAFKA_SASL_INTER_BROKER_PROTOCOL}
     fi
 
-    export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT
-    echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    if [[ "$TLS_ENABLED" == "true" ]]; then
+        export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP="CONTROLLER:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_SSL"
+        echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    else
+        export KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP="CONTROLLER:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT"
+        echo "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=$KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
+    fi
     export KAFKA_CFG_SASL_ENABLED_MECHANISMS="${ENABLED_MECHANISMS}"
     echo "[sasl]export KAFKA_CFG_SASL_ENABLED_MECHANISMS=${KAFKA_CFG_SASL_ENABLED_MECHANISMS}"
     export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="${INTER_BROKER_PROTOCOL}"
@@ -119,7 +129,7 @@ build_if_build_in_enabled() {
 
 get_client_default_mechanism() {
     isZkOrNot="$1"
-    if [[ "$(is_sasl_enabled)" == "false" ]]; then
+    if [[ "$(is_sasl_enabled $isZkOrNot)" == "false" ]]; then
         echo ""
         return 0
     fi
diff --git a/addons/kafka/scripts/kafka-server-setup.sh b/addons/kafka/scripts/kafka-server-setup.sh
index 861499329..e6e6e73de 100644
--- a/addons/kafka/scripts/kafka-server-setup.sh
+++ b/addons/kafka/scripts/kafka-server-setup.sh
@@ -72,8 +72,6 @@ set_tls_configuration_if_needed() {
   fi
   export KAFKA_TLS_TRUSTSTORE_FILE="$kafka_config_certs_path/kafka.truststore.pem"
   echo "[tls]KAFKA_TLS_TRUSTSTORE_FILE=$KAFKA_TLS_TRUSTSTORE_FILE"
-  echo "[tls]ssl.endpoint.identification.algorithm=" >> $kafka_kraft_config_path/server.properties
-  echo "[tls]ssl.endpoint.identification.algorithm=" >> $kafka_config_path/server.properties
   return 0
 }