feat: kafka support custom sasl

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 -}}

addons-cluster/kafka/templates/account-secret.yaml

@@ -0,0 +1,15 @@
+{{- if and (.Values.sasl.enable) .Values.sasl.useKBBuildInSasl }}
+apiVersion: v1
+kind: Secret
+metadata:
+  name: {{ include "kafka-broker-accounts-secret-name" . }}
+  labels:
+    app.kubernetes.io/name: {{ include "kblib.clusterName" . }}
+    app.kubernetes.io/component: {{ include "kafka-broker-component" . }}
+type: Opaque
+stringData:
+  # accounts: |
+  # [mechanism=PLAIN] format:
+  #   {accountName}:{md5Password}
+  accounts: ""
+{{- end }}

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 }}

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 }}

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"
+        }
+      }
+    }
   }
 }

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
 

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

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"