diff --git a/addons-cluster/kafka/templates/_helpers.tpl b/addons-cluster/kafka/templates/_helpers.tpl
index ca307cd77..b83c630c8 100644
--- a/addons-cluster/kafka/templates/_helpers.tpl
+++ b/addons-cluster/kafka/templates/_helpers.tpl
@@ -90,11 +90,34 @@ kafka2-external-zk
 {{- end -}}
 {{- end -}}
 
-{{- define "kafka-cluster.brokerCommonEnv" -}}
+{{- define "kafka-cluster.basicEnv" -}}
+- name: KB_CLUSTER_VERSION
+  value: "{{ .Values.version }}"
+- name: KB_CLUSTER_WITH_ZK
+  value: "{{- if hasPrefix "withZookeeper" .Values.mode }}true{{- else }}false{{- end }}"
+{{/*
+will deprecated:
+- KB_KAFKA_ENABLE_SASL
+- KB_KAFKA_ENABLE_SASL_SCRAM
+*/}}
 - name: KB_KAFKA_ENABLE_SASL
   value: "{{ .Values.saslEnable }}"
 - name: KB_KAFKA_ENABLE_SASL_SCRAM
   value: "{{ .Values.saslScramEnable }}"
+- name: KB_KAFKA_SASL_ENABLE
+  value: "{{ .Values.sasl.enable }}"
+{{- if .Values.sasl.enable }}
+- name: KB_KAFKA_SASL_USE_KB_BUILTIN
+  value: "{{ .Values.sasl.useKBBuildInSasl }}"
+- name: KB_KAFKA_SASL_MECHANISMS
+  value: "{{ .Values.sasl.mechanisms | join "," }}"
+- name: KB_KAFKA_SASL_INTER_BROKER_PROTOCOL
+  value: "{{ .Values.sasl.interBrokerProtocol }}"  
+{{- end }}
+{{- end -}}
+
+{{- define "kafka-cluster.brokerCommonEnv" -}}
+{{- include "kafka-cluster.basicEnv" . }}
 - name: KB_KAFKA_BROKER_HEAP
   value: "{{ .Values.brokerHeap }}"
 - name: KB_KAFKA_CONTROLLER_HEAP
@@ -143,3 +166,23 @@ volumeClaimTemplates:
 {{- end }}
 {{- end -}}
 
+{{- define "kafka-broker-component" -}}
+{{- if eq "combined" .Values.mode -}}
+kafka-combine
+{{- else -}}
+kafka-broker
+{{- end -}}
+{{- end -}}
+
+{{- define "kafka-broker-accounts-secret-name" -}}
+{{ include "kblib.clusterName" . }}-{{ include "kafka-broker-component" . }}-accounts
+{{- end -}}
+
+{{- define "kafka-broker-volumes" -}}
+{{- if .Values.sasl.enable }}
+volumes:
+  - name: accounts
+    secret:
+      secretName: {{ include "kafka-broker-accounts-secret-name" . }}
+{{- end }}
+{{- end -}}
\ No newline at end of file
diff --git a/addons-cluster/kafka/templates/account-secret.yaml b/addons-cluster/kafka/templates/account-secret.yaml
new file mode 100644
index 000000000..4ae156084
--- /dev/null
+++ b/addons-cluster/kafka/templates/account-secret.yaml
@@ -0,0 +1,17 @@
+{{- if and (.Values.sasl.enable) .Values.sasl.useKBBuildInSasl }}
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "kafka-broker-accounts-secret-name" . }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: {{ include "kblib.clusterName" . }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/component: {{ include "kafka-broker-component" . }}
+type: Opaque
+stringData:
+  # accounts: |
+  # [mechanism=PLAIN] format:
+  #   {accountName}:{md5Password}
+  accounts: ""
+{{- end }}
\ No newline at end of file
diff --git a/addons-cluster/kafka/templates/cluster.yaml b/addons-cluster/kafka/templates/cluster.yaml
index 593206819..78e567213 100644
--- a/addons-cluster/kafka/templates/cluster.yaml
+++ b/addons-cluster/kafka/templates/cluster.yaml
@@ -11,6 +11,7 @@ spec:
   componentSpecs:
     {{- if eq "combined" .Values.mode }}
     - name: kafka-combine
+      serviceVersion: {{ .Values.version }}
       tls: {{ .Values.tlsEnable }}
       {{- if .Values.tlsEnable }}
       issuer:
@@ -29,8 +30,10 @@ spec:
       env:
         {{- include "kafka-cluster.brokerCommonEnv" . | nindent 8 }}
       {{- include "kafka-cluster.brokerVCT" . | indent 6 }}
+      {{- include "kafka-broker-volumes" . | indent 6 }}
     {{- else }}
     - name: kafka-broker
+      serviceVersion: {{ .Values.version }}
       tls: {{ .Values.tlsEnable }}
       {{- if .Values.tlsEnable }}
       issuer:
@@ -49,6 +52,7 @@ spec:
       env:
         {{- include "kafka-cluster.brokerCommonEnv" . | nindent 8 }}
       {{- include "kafka-cluster.brokerVCT" . | indent 6 }}
+      {{- include "kafka-broker-volumes" . | indent 6 }}
       {{- if eq "withZookeeper-10" $.Values.mode }}
       serviceRefs:
         - name: kafkaZookeeper
@@ -66,6 +70,7 @@ spec:
       {{- end }}
       {{- if eq "separated" $.Values.mode }}
     - name: kafka-controller
+      serviceVersion: {{ .Values.version }}
       tls: {{ .Values.tlsEnable }}
       {{- if .Values.tlsEnable }}
       issuer:
@@ -78,9 +83,9 @@ spec:
     {{- end }}
     {{- if .Values.monitorEnable }}
     - name: kafka-exporter
+      serviceVersion: {{ .Values.exporterVersion }}
       replicas: {{ .Values.monitor.replicas }}
       env:
-        - name: KB_KAFKA_ENABLE_SASL_SCRAM
-          value: "{{ .Values.saslScramEnable }}"
+        {{- include "kafka-cluster.basicEnv" . | nindent 8 }}  
       {{- include "kafka-exporter.resources" . | nindent 6 }}
     {{- end }}
diff --git a/addons-cluster/kafka/templates/validate.yaml b/addons-cluster/kafka/templates/validate.yaml
index 44d4e51f9..c0e5292df 100644
--- a/addons-cluster/kafka/templates/validate.yaml
+++ b/addons-cluster/kafka/templates/validate.yaml
@@ -15,3 +15,13 @@ Validate kafka version and mode
     {{- end }}
   {{- end }}
 {{- end }}
+
+{{- if or (eq .Values.mode "combined") (eq .Values.mode "separated") }}
+  {{- if .Values.sasl.mechanisms }}
+    {{- range $mech := .Values.sasl.mechanisms }}
+      {{- if ne $mech "PLAIN" }}
+        {{ fail (printf "When mode is 'combined' or 'separated', only 'PLAIN' is supported in sasl.mechanisms (got: %s)" $mech) }}
+      {{- end }}
+    {{- end }}
+  {{- end }}
+{{- end }}
diff --git a/addons-cluster/kafka/values.schema.json b/addons-cluster/kafka/values.schema.json
index c5bb73b0c..374c35a66 100644
--- a/addons-cluster/kafka/values.schema.json
+++ b/addons-cluster/kafka/values.schema.json
@@ -198,5 +198,53 @@
       "title": "fixedPodIPEnabled",
       "description":"Whether to enable fixed Pod IP mode in Kafka's `advertised.listeners`"
     }
+    ,
+    "sasl": {
+      "title": "Sasl",
+      "description": "Advanced SASL authentication options.",
+      "type": "object",
+      "properties": {
+        "enable": {
+          "title": "Enable",
+          "description": "Enable SASL authentication.",
+          "type": "boolean",
+          "default": false
+        },
+        "useKBBuildInSasl": {
+          "title": "UseKBBuildInSasl",
+          "description": "Whether to use the built-in SASL server callback handler.",
+          "type": "boolean",
+          "default": true
+        },
+        "mechanisms": {
+          "title": "Mechanisms",
+          "description": "List of SASL mechanisms to enable. When useKBBuildInSasl is true, only [PLAIN] is supported currently.",
+          "type": "array",
+          "items": {
+            "type": "string",
+            "enum": [
+              "PLAIN",
+              "SCRAM-SHA-256",
+              "SCRAM-SHA-512"
+            ]
+          },
+          "default": [
+            "PLAIN"
+          ],
+          "minItems": 1
+        },
+        "interBrokerProtocol": {
+          "title": "Inter Broker Protocol",
+          "description": "SASL mechanism used for inter-broker communication. Should be one of sasl.mechanisms.",
+          "type": "string",
+          "enum": [
+            "PLAIN",
+            "SCRAM-SHA-256",
+            "SCRAM-SHA-512"
+          ],
+          "default": "PLAIN"
+        }
+      }
+    }
   }
 }
diff --git a/addons-cluster/kafka/values.yaml b/addons-cluster/kafka/values.yaml
index cb9c2b93e..9e2190c88 100644
--- a/addons-cluster/kafka/values.yaml
+++ b/addons-cluster/kafka/values.yaml
@@ -1,6 +1,9 @@
 ## @param version Kafka cluster version
 ##
-version: 3.3.2
+version: 3.9.0
+# @param version Kafka exporter component version
+##
+exporterVersion: 1.6.0
 
 ## @param mode for Kafka cluster mode, 'combined' is combined Kafka controller (KRaft) and broker,
 ## 'separated' is a Kafka KRaft and Kafka broker cluster.
@@ -11,12 +14,46 @@ mode: combined
 # Todo: Monitoring is not supported when tls is enabled
 tlsEnable: false
 
+
+# Deprecated: use sasl.enable + sasl.mechanisms instead
+# equals with:
+#   sasl:
+#     enable: true
+#     useKBBuildInSasl: false
+#     mechanism: 
+#       - PLAIN
+#     interBrokerProtocol: PLAIN
 # Enable SASL plain auth
 saslEnable: false
 
+# Will deprecate when 'sasl.mechanisms' support SCRAM-SHA-256, SCRAM-SHA-512
+# equals with:
+#   sasl:
+#     enable: true
+#     useKBBuildInSasl: false
+#     mechanism:
+#       - SCRAM-SHA-256
+#       - SCRAM-SHA-512
+#     interBrokerProtocol: SCRAM-SHA-512
 # Enable SASL Scram auth
 saslScramEnable: false
 
+# SASL settings
+sasl:
+  enable: false
+  # if true, the built-in SASL server callback handler in the knowledge base is used, and users are managed by modifying the key.
+  # if false, use kafka native SASL callback handler
+  # server.properties: 
+  #   useKBBuildInSasl=true -> sasl.server.callback.handler.class=${KBClassName}
+  #   useKBBuildInSasl=false -> sasl.server.callback.handler.class=''
+  useKBBuildInSasl: true
+  # [PLAIN,SCRAM-SHA-256,SCRAM-SHA-512]
+  # when useKBBuildInSasl=true, only supported [PLAIN] yet. SCRAM-SHA-256, SCRAM-SHA-512 will be comming soon
+  mechanisms:
+    - PLAIN
+  # should be one of sasl.mechanisms setting  
+  interBrokerProtocol: PLAIN
+
 # Enable Monitor
 monitorEnable: true
 
diff --git a/addons/kafka/configs/2.7/kafka-27-server.prop.tpl b/addons/kafka/configs/2.7/kafka-27-server.prop.tpl
index be3e68988..521fe6702 100644
--- a/addons/kafka/configs/2.7/kafka-27-server.prop.tpl
+++ b/addons/kafka/configs/2.7/kafka-27-server.prop.tpl
@@ -110,7 +110,7 @@ offsets.retention.check.interval.ms=600000
 offsets.retention.minutes=10080
 offsets.topic.compression.codec=0
 offsets.topic.num.partitions=50
-offsets.topic.replication.factor=3
+offsets.topic.replication.factor=1
 offsets.topic.segment.bytes=104857600
 password.encoder.cipher.algorithm=AES/CBC/PKCS5Padding
 password.encoder.iterations=4096
@@ -190,9 +190,9 @@ transaction.abort.timed.out.transaction.cleanup.interval.ms=10000
 transaction.max.timeout.ms=900000
 transaction.remove.expired.transaction.cleanup.interval.ms=3600000
 transaction.state.log.load.buffer.size=5242880
-transaction.state.log.min.isr=2
+transaction.state.log.min.isr=1
 transaction.state.log.num.partitions=50
-transaction.state.log.replication.factor=3
+transaction.state.log.replication.factor=1
 transaction.state.log.segment.bytes=104857600
 transactional.id.expiration.ms=604800000
 unclean.leader.election.enable=false
diff --git a/addons/kafka/scripts-ut-spec/kafka_server_setup_spec.sh b/addons/kafka/scripts-ut-spec/kafka_server_setup_spec.sh
index 343fcb823..ce0c866ef 100644
--- a/addons/kafka/scripts-ut-spec/kafka_server_setup_spec.sh
+++ b/addons/kafka/scripts-ut-spec/kafka_server_setup_spec.sh
@@ -57,8 +57,6 @@ Describe "Kafka Server Setup Script Tests"
     unset CONTROLLER_POD_NAME_LIST
     unset KB_HOST_IP
     unset BROKER_MIN_NODE_ID
-    unset KB_KAFKA_ENABLE_SASL
-    unset KB_KAFKA_SASL_CONFIG_PATH
     unset KAFKA_KRAFT_CLUSTER_ID
     unset KB_KAFKA_BROKER_HEAP
     unset KB_KAFKA_CONTROLLER_HEAP
@@ -147,24 +145,6 @@ Describe "Kafka Server Setup Script Tests"
     End
   End
 
-  Describe "override_sasl_configuration()"
-    It "sets SASL configuration when KB_KAFKA_ENABLE_SASL is true"
-      KB_KAFKA_ENABLE_SASL="true"
-      KB_KAFKA_SASL_CONFIG_PATH="$kafka_config_path/kafka_jaas.conf"
-      touch "$KB_KAFKA_SASL_CONFIG_PATH"
-      When run override_sasl_configuration
-      The output should include "[sasl]KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT"
-      The status should be success
-    End
-
-    It "does not set SASL configuration when KB_KAFKA_ENABLE_SASL is false"
-      KB_KAFKA_ENABLE_SASL="false"
-      When run override_sasl_configuration
-      The output should not include "KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP"
-      The status should be success
-    End
-  End
-
   Describe "generate_kraft_cluster_id()"
     It "sets KAFKA_KRAFT_CLUSTER_ID if provided"
       KAFKA_KRAFT_CLUSTER_ID="my-cluster-id"
diff --git a/addons/kafka/scripts/common.sh b/addons/kafka/scripts/common.sh
new file mode 100644
index 000000000..7243f990a
--- /dev/null
+++ b/addons/kafka/scripts/common.sh
@@ -0,0 +1,139 @@
+#!/bin/bash
+set -e
+
+kafka_libs_path="/opt/bitnami/kafka/libs"
+kafka_config_path="/opt/bitnami/kafka/config"
+
+is_sasl_enabled() {
+    isZkOrNot="$1"
+
+    if [[ "${KB_KAFKA_ENABLE_SASL}" == "true" ]] || [[ "${KB_KAFKA_SASL_ENABLE}" == "true" ]]; then
+        echo "true"
+    elif [[ "${isZkOrNot}" == "true" ]] && [[ "${KB_KAFKA_ENABLE_SASL_SCRAM}" == "true" ]]; then
+        echo "true"
+    else
+        echo "false"
+    fi
+}
+
+is_sasl_build_in_enabled() {
+    if [[ "${KB_KAFKA_SASL_ENABLE:-false}" == "false" ]] || [[ "${KB_KAFKA_ENABLE_SASL_SCRAM:-false}" == "true" ]]; then
+        echo "false"
+    else
+        echo "${KB_KAFKA_SASL_USE_KB_BUILTIN:-false}"
+    fi
+}
+
+build_zk_server_sasl_properties() {
+    local ENABLED_MECHANISMS="SCRAM-SHA-256,SCRAM-SHA-512"
+    local INTER_BROKER_PROTOCOL="SCRAM-SHA-512"
+
+    if [[ "${KB_KAFKA_SASL_ENABLE}" == "true" ]] && [[ -n "${KB_KAFKA_SASL_MECHANISMS}" ]] && [[ -n "${KB_KAFKA_SASL_INTER_BROKER_PROTOCOL}" ]]; then
+        ENABLED_MECHANISMS=${KB_KAFKA_SASL_MECHANISMS}
+        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"
+    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}"
+    echo "[sasl]export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=${KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL}"
+}
+
+build_kraft_server_sasl_properties() {
+    local ENABLED_MECHANISMS="PLAIN"
+    local INTER_BROKER_PROTOCOL="PLAIN"
+
+    if [[ "${KB_KAFKA_SASL_ENABLE}" == "true" ]] && [[ -n "${KB_KAFKA_SASL_MECHANISMS}" ]] && [[ -n "${KB_KAFKA_SASL_INTER_BROKER_PROTOCOL}" ]]; then
+        ENABLED_MECHANISMS=${KB_KAFKA_SASL_MECHANISMS}
+        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"
+    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}"
+    echo "[sasl]export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=${KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL}"
+}
+
+build_server_jaas_config() {
+    local admin_password="${KAFKA_ADMIN_PASSWORD}"
+    local client_password="${KAFKA_CLIENT_PASSWORD}"
+    local login_module="${1}"
+
+    if [ "$(is_sasl_build_in_enabled)" == "true" ]; then
+        # build-in only support plain yet
+        login_module="org.apache.kafka.common.security.plain.PlainLoginModule required"
+    fi
+
+    cat << EOF > ${kafka_config_path}/kafka_jaas.conf
+KafkaServer {
+  ${login_module}
+  username="$KAFKA_ADMIN_USER"
+  password="$admin_password";
+};
+KafkaClient {
+  ${login_module}
+  username="$KAFKA_CLIENT_USER"
+  password="$client_password";
+};
+EOF
+
+    echo "[sasl] write jaas config to /opt/bitnami/kafka/config/kafka_jaas.conf"
+}
+
+build_encode_password() {
+    local password="${1}"
+    echo -n "${password}" | md5sum | awk '{print $1}'
+}
+
+build_if_build_in_enabled() {
+    if [ "$(is_sasl_build_in_enabled)" == "false" ]; then
+        return 0
+    fi
+
+    if [ ! -d "/shared-tools/sasl" ]; then
+        return 1
+    fi
+    if [ ! -f "/shared-tools/sasl/get-sasl-jar.sh" ]; then
+        return 1
+    fi
+    if [ -z "${KB_CLUSTER_VERSION}" ]; then
+        return 1
+    fi
+
+    local jar_path=$(/shared-tools/sasl/get-sasl-jar.sh ${KB_CLUSTER_VERSION} ${kafka_libs_path})
+    if [ -z "${jar_path}" ]; then
+        echo "no custom jar found. maybe build-in sasl not support for ${KB_CLUSTER_VERSION}"
+        return 1
+    fi
+
+    export KAFKA_CFG_LISTENER_NAME_CLIENT_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS=${jar_path}
+    echo "[sasl]export KAFKA_CFG_LISTENER_NAME_CLIENT_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS=${KAFKA_CFG_LISTENER_NAME_CLIENT_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS}"
+    
+    export KAFKA_CFG_LISTENER_NAME_INTERNAL_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS=${jar_path}
+    echo "[sasl]export KAFKA_CFG_LISTENER_NAME_INTERNAL_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS=${KAFKA_CFG_LISTENER_NAME_INTERNAL_PLAIN_SASL_SERVER_CALLBACK_HANDLER_CLASS}"
+}
+
+get_client_default_mechanism() {
+    isZkOrNot="$1"
+    if [[ "$(is_sasl_enabled)" == "false" ]]; then
+        echo ""
+        return 0
+    fi
+    if [[ -n "$KB_KAFKA_SASL_MECHANISMS" ]]; then
+        if [[ "$KB_KAFKA_SASL_MECHANISMS" == *,* ]]; then
+            echo "${KB_KAFKA_SASL_MECHANISMS%%,*}"
+        else
+            echo "$KB_KAFKA_SASL_MECHANISMS"
+        fi
+        return 0
+    fi
+    if [[ "${isZkOrNot}" == "true" ]] && [[ "${KB_KAFKA_ENABLE_SASL_SCRAM}" == "true" ]]; then
+        echo "SCRAM-SHA-512"
+        return 0
+    fi
+    echo "PLAIN"
+}
\ No newline at end of file
diff --git a/addons/kafka/scripts/kafka-27-server-setup.sh b/addons/kafka/scripts/kafka-27-server-setup.sh
index 440fdf08c..c5bdd8387 100644
--- a/addons/kafka/scripts/kafka-27-server-setup.sh
+++ b/addons/kafka/scripts/kafka-27-server-setup.sh
@@ -103,31 +103,19 @@ convert_server_properties_to_env_var() {
 }
 
 override_sasl_configuration() {
-  # override SASL settings
-  if [[ "true" == "$KB_KAFKA_ENABLE_SASL" ]]; then
-    # bitnami default jaas setting: /opt/bitnami/kafka/config/kafka_jaas.conf
-    if [[ "${KB_KAFKA_SASL_CONFIG_PATH}" ]]; then
-      cp ${KB_KAFKA_SASL_CONFIG_PATH} $kafka_config_path/kafka_jaas.conf 2>/dev/null
-      echo "[sasl]do: cp ${KB_KAFKA_SASL_CONFIG_PATH} $kafka_config_path/kafka_jaas.conf "
-    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"
-    export KAFKA_CFG_SASL_ENABLED_MECHANISMS="PLAIN"
-    echo "[sasl]export KAFKA_CFG_SASL_ENABLED_MECHANISMS=${KAFKA_CFG_SASL_ENABLED_MECHANISMS}"
-    export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="PLAIN"
-    echo "[sasl]export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=${KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL}"
+  local isZkOrNot="true"
+  local defaultLoginModule="org.apache.kafka.common.security.scram.ScramLoginModule required"
+  
+  if [[ $(is_sasl_enabled "${isZkOrNot}") == "false" ]]; then
+    echo "sasl is not enabled, skipping SASL configuration"
+    return 0
   fi
 
-  if [[ "true" == "$KB_KAFKA_ENABLE_SASL_SCRAM" ]]; then
-    # bitnami default jaas setting: /opt/bitnami/kafka/config/kafka_jaas.conf
-    cat << EOF > /opt/bitnami/kafka/config/kafka_jaas.conf
-KafkaServer {
-  org.apache.kafka.common.security.scram.ScramLoginModule required
-  username="$KAFKA_ADMIN_USER"
-  password="$KAFKA_ADMIN_PASSWORD";
-};
-EOF
-    echo "[sasl] write jaas config to /opt/bitnami/kafka/config/kafka_jaas.conf "
+  build_server_jaas_config "${defaultLoginModule}"
+  build_if_build_in_enabled
+  build_zk_server_sasl_properties
+
+  if [[ "$(is_sasl_build_in_enabled)" == "false" ]] && [[ "true" == "$KB_KAFKA_ENABLE_SASL_SCRAM" ]]; then
     # NB: use the endpoint with sub path
     first_zoopkeeper=${KAFKA_CFG_ZOOKEEPER_CONNECT%%,*}
     echo "[sasl] zookeeper address: $first_zoopkeeper"
@@ -140,13 +128,6 @@ EOF
       --add-config "SCRAM-SHA-256=[iterations=8192,password=$KAFKA_CLIENT_PASSWORD],SCRAM-SHA-512=[password=$KAFKA_CLIENT_PASSWORD]" \
       --entity-type users --entity-name "$KAFKA_CLIENT_USER"
     echo "[sasl] add user $KAFKA_CLIENT_USER to zookeeper"
-
-    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"
-    export KAFKA_CFG_SASL_ENABLED_MECHANISMS="SCRAM-SHA-256,SCRAM-SHA-512"
-    echo "[sasl]export KAFKA_CFG_SASL_ENABLED_MECHANISMS=${KAFKA_CFG_SASL_ENABLED_MECHANISMS}"
-    export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="SCRAM-SHA-512"
-    echo "[sasl]export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=${KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL}"
   fi
 }
 
diff --git a/addons/kafka/scripts/kafka-exporter-setup.sh b/addons/kafka/scripts/kafka-exporter-setup.sh
index 7db1c7455..dcc930d81 100644
--- a/addons/kafka/scripts/kafka-exporter-setup.sh
+++ b/addons/kafka/scripts/kafka-exporter-setup.sh
@@ -1,4 +1,5 @@
 #!/bin/bash
+set -e
 
 # shellcheck disable=SC2034
 ut_mode="false"
@@ -45,10 +46,12 @@ get_start_kafka_exporter_cmd() {
   fi
 
   saslArgs=""
-  if [[ $KB_KAFKA_ENABLE_SASL_SCRAM == "true" ]]; then
+  if [[ $(is_sasl_enabled "${KB_CLUSTER_WITH_ZK:-false}") == "true" ]]; then 
     echo "sasl is enabled, setting sasl args" >&2
-    saslArgs="--sasl.enabled --sasl.mechanism=scram-sha512 --sasl.username=$KAFKA_ADMIN_USER --sasl.password=$KAFKA_ADMIN_PASSWORD"
-  fi
+    local default_mechanism=$(get_client_default_mechanism "${KB_CLUSTER_WITH_ZK:-false}")
+    echo "sasl mechanism from config: $default_mechanism" >&2
+    saslArgs=$(get_kafka_exporter_sasl_args_by_mechanism "$default_mechanism")
+  fi 
 
   if [[ -n "$TLS_ENABLED" ]]; then
     echo "TLS_ENABLED is set to true, start kafka_exporter with tls enabled." >&2
@@ -60,6 +63,34 @@ get_start_kafka_exporter_cmd() {
   return 0
 }
 
+get_kafka_exporter_sasl_args_by_mechanism() {
+  local user_var_name=${KAFKA_ADMIN_USER_BROKER}
+  local password_var_name=${KAFKA_ADMIN_PASSWORD_BROKER}
+
+  if [[ -z "$user_var_name" ]]; then
+    user_var_name="$KAFKA_ADMIN_USER_COMBINE"
+    password_var_name="$KAFKA_ADMIN_PASSWORD_COMBINE"
+  fi
+  
+  local mechanism="$1"
+  case "${mechanism,,}" in
+    "scram-sha512")
+      echo "--sasl.enabled --sasl.mechanism=scram-sha512 --sasl.username=$user_var_name --sasl.password=$password_var_name"
+      ;;
+    "scram-sha256")
+      echo "--sasl.enabled --sasl.mechanism=scram-sha256 --sasl.username=$user_var_name --sasl.password=$password_var_name"
+      ;;
+    "plain")
+      echo "--sasl.enabled --sasl.mechanism=plain --sasl.username=$user_var_name --sasl.password=$password_var_name"
+      ;;
+    *)
+      echo "invalid or not supported sasl mechanism: $mechanism" >&2
+      return 1
+      ;;
+  esac
+}
+
+
 start_kafka_exporter() {
   local cmd
   cmd=$(get_start_kafka_exporter_cmd)
diff --git a/addons/kafka/scripts/kafka-sasl-sample.prop.tpl b/addons/kafka/scripts/kafka-sasl-sample.prop.tpl
deleted file mode 100644
index c6b147421..000000000
--- a/addons/kafka/scripts/kafka-sasl-sample.prop.tpl
+++ /dev/null
@@ -1,12 +0,0 @@
-KafkaClient {
-   org.apache.kafka.common.security.plain.PlainLoginModule required
-   username="client"
-   password="kubeblocks";
-};
-KafkaServer {
-   org.apache.kafka.common.security.plain.PlainLoginModule required
-   username="admin"
-   password="kubeblocks"
-   user_admin="kubeblocks"
-   user_client="kubeblocks";
-};
\ No newline at end of file
diff --git a/addons/kafka/scripts/kafka-server-setup.sh b/addons/kafka/scripts/kafka-server-setup.sh
index e70857417..861499329 100644
--- a/addons/kafka/scripts/kafka-server-setup.sh
+++ b/addons/kafka/scripts/kafka-server-setup.sh
@@ -108,19 +108,13 @@ convert_server_properties_to_env_var() {
 }
 
 override_sasl_configuration() {
-  # override SASL settings
-  if [[ "true" == "$KB_KAFKA_ENABLE_SASL" ]]; then
-    # bitnami default jaas setting: /opt/bitnami/kafka/config/kafka_jaas.conf
-    if [[ "${KB_KAFKA_SASL_CONFIG_PATH}" ]]; then
-      cp ${KB_KAFKA_SASL_CONFIG_PATH} $kafka_config_path/kafka_jaas.conf 2>/dev/null
-      echo "[sasl]do: cp ${KB_KAFKA_SASL_CONFIG_PATH} $kafka_config_path/kafka_jaas.conf "
-    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"
-    export KAFKA_CFG_SASL_ENABLED_MECHANISMS="PLAIN"
-    echo "[sasl]export KAFKA_CFG_SASL_ENABLED_MECHANISMS=${KAFKA_CFG_SASL_ENABLED_MECHANISMS}"
-    export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL="PLAIN"
-    echo "[sasl]export KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL=${KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL}"
+  local isZkOrNot="false"
+  local defaultLoginModule="org.apache.kafka.common.security.plain.PlainLoginModule required"
+
+  if [[ $(is_sasl_enabled "${isZkOrNot}") == "true" ]]; then
+    build_server_jaas_config "${defaultLoginModule}"
+    build_if_build_in_enabled
+    build_kraft_server_sasl_properties
   fi
 }
 
diff --git a/addons/kafka/templates/_helpers.tpl b/addons/kafka/templates/_helpers.tpl
index 04c4f9784..fbe1f4cb6 100644
--- a/addons/kafka/templates/_helpers.tpl
+++ b/addons/kafka/templates/_helpers.tpl
@@ -214,13 +214,6 @@ Define kafka 2.x broker env configmap tpl name
 kafka2-broker-env-tpl
 {{- end -}}
 
-{{/*
-Define kafka tools scripts tpl name
-*/}}
-{{- define "kafka.toolsScriptsTplName" -}}
-kafka-tools-scripts-tpl
-{{- end -}}
-
 {{- define "kafka.tls.cmpdConfig" -}}
 tls:
   volumeName: tls 
diff --git a/addons/kafka/templates/cmpd-broker-27.yaml b/addons/kafka/templates/cmpd-broker-27.yaml
index 84d0070d4..33a00e88d 100644
--- a/addons/kafka/templates/cmpd-broker-27.yaml
+++ b/addons/kafka/templates/cmpd-broker-27.yaml
@@ -119,11 +119,6 @@ spec:
       volumeName: scripts
       namespace: {{ .Release.Namespace }}
       defaultMode: 0755
-    - name: kafka-tools-scripts-tpl
-      template: {{ include "kafka.toolsScriptsTplName" . }}
-      volumeName: tools
-      namespace: {{ .Release.Namespace }}
-      defaultMode: 0755
   systemAccounts:
     - name: client
       initAccount: true
@@ -152,6 +147,18 @@ spec:
     securityContext:
       {{- toYaml .Values.securityContext | nindent 6 }}
     {{- end }}
+    initContainers:
+      - name: kafkatool
+        imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
+        command:
+          - sh
+          - -c
+          - |
+            cp -r /sasl /shared-tools/sasl
+            chmod +x /shared-tools/sasl
+        volumeMounts:
+          - name: shared-tools
+            mountPath: /shared-tools
     containers:
       - name: kafka
         imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
@@ -212,6 +219,8 @@ spec:
             value: "User:$(KAFKA_ADMIN_USER)"
           - name: BROKER_MIN_NODE_ID
             value: {{ .Values.kafkaBroker.minNodeId | quote }}
+          - name: KAFKA_DYNAMIC_CREDENTIAL_FILE
+            value: "/accounts/accounts-mount/accounts"    
           {{- with .Values.extraEnvs.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -250,12 +259,10 @@ spec:
           - name: kafka-config
             mountPath: /scripts/server.properties
             subPath: server.properties
-          - name: tools
-            mountPath: /tools/client-ssl.properties
-            subPath: client-ssl.properties
-          - name: tools
-            mountPath: /tools/server-jaas.properties
-            subPath: server-jaas.properties
+          - name: shared-tools
+            mountPath: /shared-tools  
+          - name: accounts
+            mountPath: /accounts/accounts-mount   
           {{- with .Values.extraVolumeMounts.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -286,3 +293,6 @@ spec:
         volumeMounts:
           - name: jmx-config
             mountPath: /etc/jmx-kafka
+    volumes:
+      - name: shared-tools
+        emptyDir: {}            
diff --git a/addons/kafka/templates/cmpd-broker.yaml b/addons/kafka/templates/cmpd-broker.yaml
index 7d52ac418..47bde148c 100644
--- a/addons/kafka/templates/cmpd-broker.yaml
+++ b/addons/kafka/templates/cmpd-broker.yaml
@@ -28,8 +28,30 @@ spec:
       valueFrom:
         clusterVarRef:
           clusterUID: Required
-    - name: SUPER_USER
-      value: admin
+    - name: KAFKA_ADMIN_USER
+      valueFrom:
+        credentialVarRef:
+          name: admin
+          optional: false
+          username: Required
+    - name: KAFKA_ADMIN_PASSWORD
+      valueFrom:
+        credentialVarRef:
+          name: admin
+          optional: false
+          password: Required
+    - name: KAFKA_CLIENT_USER
+      valueFrom:
+        credentialVarRef:
+          name: client
+          optional: false
+          username: Required
+    - name: KAFKA_CLIENT_PASSWORD
+      valueFrom:
+        credentialVarRef:
+          name: client
+          optional: false
+          password: Required
     - name: POD_FQDN_LIST
       valueFrom:
         componentVarRef:
@@ -94,16 +116,36 @@ spec:
       volumeName: scripts
       namespace: {{ .Release.Namespace }}
       defaultMode: 0755
-    - name: kafka-tools-scripts-tpl
-      template: {{ include "kafka.toolsScriptsTplName" . }}
-      volumeName: tools
-      namespace: {{ .Release.Namespace }}
-      defaultMode: 0755
+  systemAccounts:
+    - name: client
+      initAccount: true
+      passwordGenerationPolicy:
+        length: 16
+        numDigits: 8
+        letterCase: MixedCases
+    - name: admin
+      initAccount: true
+      passwordGenerationPolicy:
+        length: 16
+        numDigits: 8
+        letterCase: MixedCases       
   runtime:
     {{- if .Values.securityContext }}
     securityContext:
       {{- toYaml .Values.securityContext | nindent 6 }}
     {{- end }}
+    initContainers:
+      - name: kafkatool
+        imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
+        command:
+          - sh
+          - -c
+          - |
+            cp -r /sasl /shared-tools/sasl
+            chmod +x /shared-tools/sasl
+        volumeMounts:
+          - name: shared-tools
+            mountPath: /shared-tools
     containers:
       - name: kafka
         imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
@@ -165,13 +207,11 @@ spec:
           - name: KAFKA_KRAFT_CLUSTER_ID
             value: $(CLUSTER_UID)
           - name: KAFKA_CFG_SUPER_USERS
-            value: "User:$(SUPER_USER)"
-          # - name: KB_KAFKA_ENABLE_SASL  # enable the SASL with plain mode
-          #   value: "true"
-          - name: KB_KAFKA_SASL_CONFIG_PATH  # specify the SASL jaas users
-            value: /tools/server-jaas.properties
+            value: "User:$(KAFKA_ADMIN_USER)"
           - name: BROKER_MIN_NODE_ID
             value: {{ .Values.kafkaBroker.minNodeId | quote }}
+          - name: KAFKA_DYNAMIC_CREDENTIAL_FILE
+            value: "/accounts/accounts-mount/accounts"  
           {{- with .Values.extraEnvs.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -210,12 +250,10 @@ spec:
           - name: kafka-config
             mountPath: /scripts/server.properties
             subPath: server.properties
-          - name: tools
-            mountPath: /tools/client-ssl.properties
-            subPath: client-ssl.properties
-          - name: tools
-            mountPath: /tools/server-jaas.properties
-            subPath: server-jaas.properties
+          - name: shared-tools
+            mountPath: /shared-tools
+          - name: accounts
+            mountPath: /accounts/accounts-mount 
           {{- with .Values.extraVolumeMounts.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -246,3 +284,6 @@ spec:
         volumeMounts:
           - name: jmx-config
             mountPath: /etc/jmx-kafka
+    volumes:
+      - name: shared-tools
+        emptyDir: {}           
diff --git a/addons/kafka/templates/cmpd-combine.yaml b/addons/kafka/templates/cmpd-combine.yaml
index cd1926c60..29c00e7bd 100644
--- a/addons/kafka/templates/cmpd-combine.yaml
+++ b/addons/kafka/templates/cmpd-combine.yaml
@@ -30,8 +30,30 @@ spec:
       valueFrom:
         clusterVarRef:
           clusterUID: Required
-    - name: SUPER_USER
-      value: admin
+    - name: KAFKA_ADMIN_USER
+      valueFrom:
+        credentialVarRef:
+          name: admin
+          optional: false
+          username: Required
+    - name: KAFKA_ADMIN_PASSWORD
+      valueFrom:
+        credentialVarRef:
+          name: admin
+          optional: false
+          password: Required
+    - name: KAFKA_CLIENT_USER
+      valueFrom:
+        credentialVarRef:
+          name: client
+          optional: false
+          username: Required
+    - name: KAFKA_CLIENT_PASSWORD
+      valueFrom:
+        credentialVarRef:
+          name: client
+          optional: false
+          password: Required
     - name: POD_FQDN_LIST
       valueFrom:
         componentVarRef:
@@ -84,16 +106,36 @@ spec:
       volumeName: scripts
       namespace: {{ .Release.Namespace }}
       defaultMode: 0755
-    - name: kafka-tools-scripts-tpl
-      template: {{ include "kafka.toolsScriptsTplName" . }}
-      volumeName: tools
-      namespace: {{ .Release.Namespace }}
-      defaultMode: 0755
+  systemAccounts:
+    - name: client
+      initAccount: true
+      passwordGenerationPolicy:
+        length: 16
+        numDigits: 8
+        letterCase: MixedCases
+    - name: admin
+      initAccount: true
+      passwordGenerationPolicy:
+        length: 16
+        numDigits: 8
+        letterCase: MixedCases    
   runtime:
     {{- if .Values.securityContext }}
     securityContext:
       {{- toYaml .Values.securityContext | nindent 6 }}
     {{- end }}
+    initContainers:
+      - name: kafkatool
+        imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
+        command:
+          - sh
+          - -c
+          - |
+            cp -r /sasl /shared-tools/sasl
+            chmod +x /shared-tools/sasl
+        volumeMounts:
+          - name: shared-tools
+            mountPath: /shared-tools
     containers:
       - name: kafka
         imagePullPolicy: {{ default "IfNotPresent" .Values.images.pullPolicy }}
@@ -155,11 +197,9 @@ spec:
           - name: KAFKA_KRAFT_CLUSTER_ID
             value: $(CLUSTER_UID)
           - name: KAFKA_CFG_SUPER_USERS
-            value: "User:$(SUPER_USER)"
-          # - name: KB_KAFKA_ENABLE_SASL  # enable the SASL with plain mode
-          # value: "false"
-          - name: KB_KAFKA_SASL_CONFIG_PATH  # specify the SASL jaas users
-            value: /tools/server-jaas.properties
+            value: "User:$(KAFKA_ADMIN_USER)"
+          - name: KAFKA_DYNAMIC_CREDENTIAL_FILE
+            value: "/accounts/accounts-mount/accounts"
           {{- with .Values.extraEnvs.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -200,12 +240,10 @@ spec:
           - name: kafka-config
             mountPath: /scripts/server.properties
             subPath: server.properties
-          - name: tools
-            mountPath: /tools/client-ssl.properties
-            subPath: client-ssl.properties
-          - name: tools
-            mountPath: /tools/server-jaas.properties
-            subPath: server-jaas.properties
+          - name: shared-tools
+            mountPath: /shared-tools
+          - name: accounts
+            mountPath: /accounts/accounts-mount  
           {{- with .Values.extraVolumeMounts.kafka }}
           {{- toYaml . | nindent 10}}
           {{- end }}
@@ -236,3 +274,6 @@ spec:
         volumeMounts:
           - name: jmx-config
             mountPath: /etc/jmx-kafka
+    volumes:
+      - name: shared-tools
+        emptyDir: {}        
diff --git a/addons/kafka/templates/cmpd-exporter.yaml b/addons/kafka/templates/cmpd-exporter.yaml
index 8e61a80f6..1626d649e 100644
--- a/addons/kafka/templates/cmpd-exporter.yaml
+++ b/addons/kafka/templates/cmpd-exporter.yaml
@@ -31,20 +31,34 @@ spec:
           compDef: {{ include "kafka-combine.cmpdRegexpPattern" . }}
           optional: true
           podFQDNs: Required
-    - name: KAFKA_ADMIN_USER
+    - name: KAFKA_ADMIN_USER_BROKER
       valueFrom:
         credentialVarRef:
           compDef: "^kafka\\d*-broker-"
           name: admin
           optional: true
           username: Required
-    - name: KAFKA_ADMIN_PASSWORD
+    - name: KAFKA_ADMIN_PASSWORD_BROKER
       valueFrom:
         credentialVarRef:
           compDef: "^kafka\\d*-broker-"
           name: admin
           optional: true
           password: Required
+    - name: KAFKA_ADMIN_USER_COMBINE
+      valueFrom:
+        credentialVarRef:
+          compDef: "^kafka\\d*-combine-"
+          name: admin
+          optional: true
+          username: Required
+    - name: KAFKA_ADMIN_PASSWORD_COMBINE
+      valueFrom:
+        credentialVarRef:
+          compDef: "^kafka\\d*-combine-"
+          name: admin
+          optional: true
+          password: Required      
   scripts:
   - name: kafka-scripts-tpl
     template: {{ include "kafka.serverScriptsTplName" . }}
diff --git a/addons/kafka/templates/cmpv-broker.yaml b/addons/kafka/templates/cmpv-broker.yaml
index 7fd95fcb4..1e1e7aef4 100644
--- a/addons/kafka/templates/cmpv-broker.yaml
+++ b/addons/kafka/templates/cmpv-broker.yaml
@@ -28,6 +28,7 @@ spec:
       images:
         kafka: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.kafka.repository }}:{{ .tag }}
         jmx-exporter: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.jmxExporter.repository }}:{{ $.Values.images.jmxExporter.tag }}
+        kafkatool: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.kafkatool.repository }}:{{ $.Values.images.kafkatool.tag }}
         {{- if eq $key "v2" }}
         accountProvision: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.kafka.repository }}:{{ .tag }}
         {{- end }}
diff --git a/addons/kafka/templates/cmpv-combine.yaml b/addons/kafka/templates/cmpv-combine.yaml
index ba9faa062..7433e280f 100644
--- a/addons/kafka/templates/cmpv-combine.yaml
+++ b/addons/kafka/templates/cmpv-combine.yaml
@@ -21,4 +21,5 @@ spec:
       images:
         kafka: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.kafka.repository }}:{{ .tag }}
         jmx-exporter: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.jmxExporter.repository }}:{{ $.Values.images.jmxExporter.tag }}
+        kafkatool: {{ $.Values.images.registry | default "docker.io" }}/{{ $.Values.images.kafkatool.repository }}:{{ $.Values.images.kafkatool.tag }}
   {{- end }}
diff --git a/addons/kafka/templates/script-template.yaml b/addons/kafka/templates/script-template.yaml
index 1b0b8bd8f..831301555 100644
--- a/addons/kafka/templates/script-template.yaml
+++ b/addons/kafka/templates/script-template.yaml
@@ -11,8 +11,14 @@ data:
     {{- include "kblib.compvars.get_target_pod_fqdn_from_pod_fqdn_vars" $ | nindent 4 }}
     {{- include "kblib.strings.is_empty" $ | nindent 4 }}
   kafka-server-setup.sh: |-
+    #!/bin/bash
+    set -e;
+    {{- .Files.Get "scripts/common.sh" | nindent 4 }}
     {{- .Files.Get "scripts/kafka-server-setup.sh" | nindent 4 }}
   kafka-exporter-setup.sh: |-
+    #!/bin/bash
+    set -e;
+    {{- .Files.Get "scripts/common.sh" | nindent 4 }}
     {{- .Files.Get "scripts/kafka-exporter-setup.sh" | nindent 4 }}
   kafka-env.sh: |-
     {{- .Files.Get "scripts/kafka-env.sh" | nindent 4 }}
@@ -30,21 +36,14 @@ data:
     {{- include "kblib.compvars.get_target_pod_fqdn_from_pod_fqdn_vars" $ | nindent 4 }}
     {{- include "kblib.strings.is_empty" $ | nindent 4 }}
   kafka-server-setup.sh: |-
+    #!/bin/bash
+    set -e;
+    {{- .Files.Get "scripts/common.sh" | nindent 4 }}
     {{- .Files.Get "scripts/kafka-27-server-setup.sh" | nindent 4 }}
   kafka-exporter-setup.sh: |-
+    #!/bin/bash
+    set -e;
+    {{- .Files.Get "scripts/common.sh" | nindent 4 }}
     {{- .Files.Get "scripts/kafka-exporter-setup.sh" | nindent 4 }}
   kafka-env.sh: |-
-    {{- .Files.Get "scripts/kafka-env.sh" | nindent 4 }}
----
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: {{ include "kafka.toolsScriptsTplName" . }}
-  namespace: {{ .Release.Namespace | quote }}
-  labels: 
-    {{- include "kafka.labels" . | nindent 4 }}
-data:
-  client-ssl.properties: |-
-    {{- .Files.Get "scripts/kafka-ssl.prop.tpl" | nindent 4 }}
-  server-jaas.properties: |-
-    {{- .Files.Get "scripts/kafka-sasl-sample.prop.tpl" | nindent 4 }}
+    {{- .Files.Get "scripts/kafka-env.sh" | nindent 4 }}
\ No newline at end of file
diff --git a/addons/kafka/values.yaml b/addons/kafka/values.yaml
index 83d50578a..669dffeec 100644
--- a/addons/kafka/values.yaml
+++ b/addons/kafka/values.yaml
@@ -45,6 +45,9 @@ images:
     # mirrored from deviceinsight/kafkactl
     repository: apecloud/kafkactl
     tag: v5.15.0
+  kafkatool:
+    repository: apecloud/kafkatools
+    tag: 0.1.0
 
 ## @param define default serviceVersion of each Component
 defaultServiceVersion: