Skip to content

fix: kafka backup when both sasl and tls enabled #2289

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
cjc7373 merged 2 commits into from
Dec 3, 2025
Merged

fix: kafka backup when both sasl and tls enabled #2289

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a444daf
fix: kafka backup when both sasl and tls enabled
cjc7373 Dec 2, 2025
75e954a
allow empty backup
cjc7373 Dec 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions addons/kafka/configs/kafka-server.prop.tpl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -225,15 +225,15 @@ 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=
# ssl.keystore.key=
# 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=
Expand Down
3 changes: 2 additions & 1 deletion addons/kafka/dataprotection/backup.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
5 changes: 4 additions & 1 deletion addons/kafka/dataprotection/common.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}

Expand Down
4 changes: 4 additions & 0 deletions addons/kafka/dataprotection/restore.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
Expand Down
20 changes: 15 additions & 5 deletions addons/kafka/scripts/common.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}"
Expand All @@ -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}"
Expand Down Expand Up @@ -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
Expand Down
2 changes: 0 additions & 2 deletions addons/kafka/scripts/kafka-server-setup.sh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand Down