feat: kafka backup topics (#2260)

Author: cjc7373

addons/kafka/dataprotection/backup.sh

@@ -0,0 +1,21 @@
+#!/bin/bash
+
+# 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
+readarray -t topics < <(kafkactl get topics -o compact | grep -v  __consumer_offsets)
+
+for topic in "${topics[@]}"; do
+  echo "backing up ${topic}..."
+  kafkactl consume "${topic}" --from-beginning --print-keys --print-timestamps --exit --print-headers -o json-raw | datasafed push - "data/${topic}.json"
+done
+
+# use datasafed to get backup size
+# if we do not write into $DP_BACKUP_INFO_FILE, the backup job will stuck
+TOTAL_SIZE=$(datasafed stat / | grep TotalSize | awk '{print $2}')
+DP_save_backup_status_info "$TOTAL_SIZE"

addons/kafka/dataprotection/common.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+set -eo pipefail
+
+# Save backup status info file for syncing progress.
+# timeFormat: %Y-%m-%dT%H:%M:%SZ
+function DP_save_backup_status_info() {
+  local totalSize=$1
+  local startTime=$2
+  local stopTime=$3
+  local timeZone=$4
+  local extras=$5
+  local timeZoneStr=""
+  if [ -n "${timeZone}" ]; then
+    timeZoneStr=$(printf ',"timeZone":"%s"' "${timeZone}")
+  fi
+  if [ -z "${stopTime}" ]; then
+    printf '{"totalSize":"%s"}' "${totalSize}" > "${DP_BACKUP_INFO_FILE}"
+  elif [ -z "${startTime}" ]; then
+    printf '{"totalSize":"%s","extras":[%s],"timeRange":{"end":"%s"%s}}' "${totalSize}" "${extras}" "${stopTime}" "${timeZoneStr}" > "${DP_BACKUP_INFO_FILE}"
+  else
+    printf '{"totalSize":"%s","extras":[%s],"timeRange":{"start":"%s","end":"%s"%s}}' "${totalSize}" "${extras}" "${startTime}" "${stopTime}" "${timeZoneStr}" > "${DP_BACKUP_INFO_FILE}"
+  fi
+}
+
+export BROKERS="$DP_DB_HOST:$DP_DB_PORT"
+export PATH="$PATH:$DP_DATASAFED_BIN_PATH"
+export DATASAFED_BACKEND_BASE_PATH=${DP_BACKUP_BASE_PATH}

addons/kafka/dataprotection/restore.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+echo "getting topics..."
+readarray -t lines < <(datasafed pull topics.txt -)
+for line in "${lines[@]}"; do
+  read -r topic partitions replication <<< "$line"
+  echo "restoring ${topic}..."
+  kafkactl create topic "$topic" --partitions "$partitions" --replication-factor "$replication"
+  datasafed pull "data/${topic}.json" - | kafkactl produce "$topic" --input-format=json
+done

addons/kafka/scripts/kafka-env.sh

@@ -42,6 +42,7 @@ kafka_env_vars=(
     KAFKA_CFG_MAX_PARTITION_FETCH_BYTES
     KAFKA_ENABLE_KRAFT
     KAFKA_KRAFT_CLUSTER_ID
+    KAFKA_SKIP_KRAFT_STORAGE_INIT
     KAFKA_ZOOKEEPER_PROTOCOL
     KAFKA_ZOOKEEPER_PASSWORD
     KAFKA_ZOOKEEPER_USER
@@ -107,6 +108,7 @@ export KAFKA_CFG_INTER_BROKER_LISTENER_NAME="${KAFKA_CFG_INTER_BROKER_LISTENER_N
 export KAFKA_CFG_MAX_REQUEST_SIZE="${KAFKA_CFG_MAX_REQUEST_SIZE:-}"
 export KAFKA_CFG_MAX_PARTITION_FETCH_BYTES="${KAFKA_CFG_MAX_PARTITION_FETCH_BYTES:-}"
 export KAFKA_ENABLE_KRAFT="${KAFKA_ENABLE_KRAFT:-yes}"
+export KAFKA_SKIP_KRAFT_STORAGE_INIT="${KAFKA_SKIP_KRAFT_STORAGE_INIT:-false}"
 export KAFKA_KRAFT_CLUSTER_ID="${KAFKA_KRAFT_CLUSTER_ID:-}"
 
 # ZooKeeper connection settings

addons/kafka/templates/actionset.yaml

@@ -0,0 +1,30 @@
+apiVersion: dataprotection.kubeblocks.io/v1alpha1
+kind: ActionSet
+metadata:
+  name: kafka-topics
+  labels:
+    {{- include "kafka.labels" . | nindent 4 }}
+spec:
+  backupType: Full
+  backup:
+    backupData:
+      image: {{ .Values.images.registry | default "docker.io" }}/{{ .Values.images.kafkactl.repository }}:{{ .Values.images.kafkactl.tag }}
+      syncProgress:
+        enabled: false
+        intervalSeconds: 5
+      command:
+      - bash
+      - -c
+      - |
+        {{- .Files.Get "dataprotection/common.sh" | nindent 8 }}
+        {{- .Files.Get "dataprotection/backup.sh" | nindent 8 }}
+  restore:
+    postReady:
+    - job:
+        image: {{ .Values.images.registry | default "docker.io" }}/{{ .Values.images.kafkactl.repository }}:{{ .Values.images.kafkactl.tag }}
+        command:
+        - bash
+        - -c
+        - |
+          {{- .Files.Get "dataprotection/common.sh" | nindent 12 }}
+          {{- .Files.Get "dataprotection/restore.sh" | nindent 12 }}

addons/kafka/templates/backuppolicytemplate.yaml

@@ -0,0 +1,21 @@
+apiVersion: dataprotection.kubeblocks.io/v1alpha1
+kind: BackupPolicyTemplate
+metadata:
+  name: kafka-backup-policy-template
+  labels:
+    {{- include "kafka.labels" . | nindent 4 }}
+spec:
+  serviceKind: Kafka
+  compDefs:
+    - {{ include "kafka2-broker.cmpdRegexpPattern" . }}
+    - {{ include "kafka-broker.cmpdRegexpPattern" . }}
+    - {{ include "kafka-combine.cmpdRegexpPattern" . }}
+  backupMethods:
+    - name: topics
+      snapshotVolumes: false
+      actionSetName: kafka-topics
+  schedules:
+    - backupMethod: topics
+      enabled: false
+      cronExpression: "0 18 * * *"
+      retentionPeriod: 7d

addons/kafka/templates/cmpd-broker.yaml

@@ -11,6 +11,7 @@ spec:
   description: Kafka broker component definition
   serviceKind: kafka
   serviceVersion: {{ .Values.defaultServiceVersion.broker }}
+  podManagementPolicy: Parallel
   services:
     - name: advertised-listener
       serviceName: advertised-listener

addons/kafka/templates/cmpd-controller.yaml

@@ -11,6 +11,7 @@ spec:
   description: Kafka controller that act as controllers (kraft) only server.
   serviceKind: kafka-controller
   serviceVersion: {{ .Values.defaultServiceVersion.controller }}
+  podManagementPolicy: Parallel
   vars:
     - name: CLUSTER_UID
       valueFrom:
@@ -96,6 +97,10 @@ spec:
             value: "CONTROLLER://:9093"
           - name: KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP
             value: "CONTROLLER:PLAINTEXT"
+          # FIXME: why this config is needed for controller? 
+          # kafka 3.8/3.9 controller failed to start without this 
+          - name: KAFKA_CFG_INTER_BROKER_LISTENER_NAME
+            value: "CONTROLLER"
           - name: ALLOW_PLAINTEXT_LISTENER
             value: "yes"
           - name: JMX_PORT

addons/kafka/values.yaml

@@ -41,6 +41,10 @@ images:
   jmxExporter:
     repository: apecloud/jmx-exporter
     tag: 0.18.0-debian-11-r20
+  kafkactl:
+    # mirrored from deviceinsight/kafkactl
+    repository: apecloud/kafkactl
+    tag: v5.15.0
 
 ## @param define default serviceVersion of each Component
 defaultServiceVersion: