Geti doesn't work properly after restart of Kafka

[jcchr]

🧩 Issue type

Bug 🐞

📄 Description

When a pod with Kafka is restarted some basic functionalities of Geti stop working - like creation of a new projects, starting training jobs etc. This is caused by missing topics - as in Geti 2.13.0 kafka's data are not kept in an ephemeral storage. The same problem appears when the whole server or kubernetes are restarted.

📋 Steps to reproduce (if applicable)

1. Delete pod with kafka - and wait until it starts again
2. Try to create a new project

💻 Expected behavior

Restart of kafka (caused by any reason - manual restart, failure, restart of a whole machine etc) shouldn't lead to problems with creating of a new project.

🖼️ Screenshots, video recordings or logs

No response

📦 Geti version

2.13.0

Hardware

No response

🖥️ Operating System

No response

Browser (if applicable)

No response

Python Version (if relevant)

No response

➕ Additional context

No response

[jcchr]

When this issue appears - recreation of missing kafka topics is required. To do so - please execute the following steps:

1. Create the file with the following content on the machine where Geti is installed - name it kafka-provisioning.yaml:

apiVersion: batch/v1
kind: Job
metadata:
  labels:
    app.kubernetes.io/instance: impt
    app.kubernetes.io/name: kafka-provisioning-retry
    app.kubernetes.io/version: 4.0.0
  name: impt-kafka-provisioning-retry
  namespace: impt
spec:
  backoffLimit: 100
  completionMode: NonIndexed
  completions: 1
  manualSelector: false
  parallelism: 1
  podReplacementPolicy: TerminatingOrFailed
  suspend: false
  template:
    metadata:
      creationTimestamp: null
      labels:
        app.kubernetes.io/instance: impt
        app.kubernetes.io/name: kafka-provisioning-retry
        app.kubernetes.io/version: 4.0.0
        batch.kubernetes.io/job-name: impt-kafka-provisioning-retry
        job-name: impt-kafka-provisioning-retry
    spec:
      containers:
      - args:
        - |
          echo "Starting Kafka provisioning"

          # Grant admin user read & describe permissions to be able to describe all topics
          echo "Grant admin user read & describe permissions to be able to describe all topics"
          /opt/kafka/bin/kafka-acls.sh \
            --add \
            --bootstrap-server $KAFKA_ADDRESS \
            --command-config /tmp/kafka/client.properties \
            --allow-principal User:$KAFKA_ADMIN_USERNAME \
            --allow-host '*' \
            --operation Read --operation Describe --operation DescribeConfigs \
            --operation Alter --operation AlterConfigs --operation Create \
            --topic '*'

          exit_status=$?
          if [ $exit_status -ne 0 ]; then
              echo "Kafka admin authorization command failed with exit status $exit_status"
              wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
              exit 1
          fi

          echo "Getting kafka description"
          describe_result=$(/opt/kafka/bin/kafka-topics.sh \
              --exclude-internal \
              --describe \
              --bootstrap-server $KAFKA_ADDRESS \
              --command-config /tmp/kafka/client.properties)

          exit_status=$?
          if [ $exit_status -ne 0 ]; then
              echo "Kafka topic listing command failed with exit status $exit_status"
              wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
              exit 1
          fi

          declare -A TOPICS=()
          while read -r line; do
            [[ ! "$line" =~ ^Topic:\ *([a-z0-9\_]*).*TopicId.*PartitionCount:\ *([0-9]*).*ReplicationFactor.*$ ]] && continue
            topic="${BASH_REMATCH[1]}"
            partition_count="${BASH_REMATCH[2]}"
            TOPICS[$topic]=$partition_count
          done < <(echo "$describe_result")

          echo "Existing topics"
          for x in "${!TOPICS[@]}"; do
            [[ ! "$x" =~ $KAFKA_TOPIC_PREFIX.* ]] && continue
            printf "%-100s | %s\n" "$x" "${TOPICS[$x]}"
          done

          while IFS=$';' read -r topic replicationFactor numPartitions config || [ -n "$topic" ]; do
            [[ "$topic" =~ ^#.*$ ]] && continue # Skip comment lines

            prefixedTopicName="${KAFKA_TOPIC_PREFIX}${topic}"
            echo "Checking topic '$prefixedTopicName'"

            if [[ -n "${TOPICS[$prefixedTopicName]}" ]] && [[ $numPartitions == ${TOPICS[$prefixedTopicName]} ]]; then
              echo "Topic '$prefixedTopicName' is already created with $numPartitions partitions, skipping"
            elif [[ -n "${TOPICS[$prefixedTopicName]}" ]] && [[ ! $numPartitions == ${TOPICS[$prefixedTopicName]} ]]; then
              if [ "$numPartitions" -lt "${TOPICS[$prefixedTopicName]}" ]; then
                echo "Cannot decrease number of partitions per topic, skipping topic update"
                continue
              fi
              echo "Updating topic '$prefixedTopicName' to $numPartitions partitions"

              /opt/kafka/bin/kafka-topics.sh \
                --alter \
                --if-exists \
                --bootstrap-server $KAFKA_ADDRESS \
                --partitions $numPartitions \
                --command-config /tmp/kafka/client.properties \
                --topic $prefixedTopicName

              exit_status=$?
              if [ $exit_status -ne 0 ]; then
                  echo "Kafka topic altering command failed with exit status $exit_status"
                  wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
                  exit 1
              fi
            else
              echo "Topic '$prefixedTopicName' doesn't exist, creating it with replication factor '$replicationFactor', partitions number '$numPartitions' and config '$config'"

              /opt/kafka/bin/kafka-topics.sh \
                --create \
                --if-not-exists \
                --bootstrap-server $KAFKA_ADDRESS \
                --replication-factor $replicationFactor \
                --partitions $numPartitions \
                --command-config /tmp/kafka/client.properties \
                $config \
                --topic $prefixedTopicName

              exit_status=$?
              if [ $exit_status -ne 0 ]; then
                  echo "Kafka topic creating command failed with exit status $exit_status"
                  wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
                  exit 1
              fi
            fi

          done < /tmp/topics/topics.properties

          failed=0

          # Wait for all background processes to finish and check their exit status
          wait_for_subprocesses() {
              pids=$*
              for pid in ${pids[*]}; do
                  wait $pid
                  exit_status=$?
                  if [ $exit_status -ne 0 ]; then
                      echo "Command failed with exit status $exit_status"
                      failed=1
                  fi
              done
          }

          batch_size=20

          counter=0
          pids=()

          while IFS=$';' read -r topic user operations || [ -n "$topic" ]; do
            [[ "$topic" =~ ^#.*$ ]] && continue # Skip comment lines

            prefixedTopicName="${KAFKA_TOPIC_PREFIX}${topic}"
            echo "Setup ACL for '$user' on '$prefixedTopicName' topic for operations '$operations'"

            # Trying to update topic if it already exists
            /opt/kafka/bin/kafka-acls.sh \
              --add \
              --bootstrap-server $KAFKA_ADDRESS \
              --command-config /tmp/kafka/client.properties \
              --allow-principal User:$user \
              --allow-host '*' \
              $operations \
              --topic $prefixedTopicName &

            pid=$!
            pids+=("$pid")

            let "counter=counter+1"
            if ((counter >= batch_size)); then
                echo "batch finished"
                wait_for_subprocesses $pids

                counter=0
                pids=()
            fi

          done < /tmp/topics/acl.properties

          wait_for_subprocesses $pids

          # Check if any command failed and handle accordingly
          if [ $failed -eq 1 ]; then
              echo "ACL provisioning failed"
              wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
              exit 1
          fi

          echo "Provisioning succeeded"
          wget --post-data="" -q -O/dev/null 127.0.0.1:15020/quitquitquit;
        command:
        - /bin/bash
        - -c
        env:
        - name: KAFKA_ADDRESS
          value: impt-kafka:9092
        - name: KAFKA_TOPIC_PREFIX
        - name: KAFKA_LOG4J_OPTS
          value: -Dlog4j.configuration=file:/tmp/kafka/tools-log4j.properties
        - name: KAFKA_ADMIN_USERNAME
          value: admin
        image: docker.io/apache/kafka:4.1.0
        imagePullPolicy: IfNotPresent
        name: kafka-provisioning-retry
        resources: {}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          readOnlyRootFilesystem: true
          runAsNonRoot: true
          runAsUser: 10001
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        volumeMounts:
        - mountPath: /tmp/kafka
          name: client-config
          readOnly: true
        - mountPath: /tmp/topics
          name: kafka-topics
        - mountPath: /opt/kafka/logs
          name: kafka-logs
      dnsPolicy: ClusterFirst
      initContainers:
      - command:
        - sh
        - -c
        - until nc -w 5 -z $KAFKA_HOST $KAFKA_PORT; do echo wait...; sleep 2; done;
        env:
        - name: KAFKA_HOST
          value: impt-kafka
        - name: KAFKA_PORT
          value: "9092"
        image: quay.io/prometheus/busybox:glibc
        imagePullPolicy: IfNotPresent
        name: wait-for-kafka
        resources: {}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          readOnlyRootFilesystem: true
          runAsNonRoot: true
          runAsUser: 10001
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      - args:
        - |
          if [ "$KAFKA_SASL_MECHANISM" = "PLAIN" ]; then
            login_module=org.apache.kafka.common.security.plain.PlainLoginModule
          else
            login_module=org.apache.kafka.common.security.scram.ScramLoginModule
          fi

          cat - > /tmp/kafka/client.properties <<EOF
          security.protocol=$KAFKA_SECURITY_PROTOCOL
          sasl.mechanism=$KAFKA_SASL_MECHANISM
          sasl.jaas.config=$login_module required username="$KAFKA_ADMIN_USERNAME" password="$KAFKA_ADMIN_PASSWORD";
          request.timeout.ms=60000
          EOF

          cat - > /tmp/kafka/tools-log4j.properties <<EOF
          log4j.rootLogger=INFO, stderr

          log4j.appender.stderr=org.apache.log4j.ConsoleAppender
          log4j.appender.stderr.layout=org.apache.log4j.PatternLayout
          log4j.appender.stderr.layout.ConversionPattern=[%d] %p %m (%c)%n
          log4j.appender.stderr.Target=System.err

          # for connect-plugin-path
          log4j.logger.org.reflections=ERROR
          EOF
        command:
        - /bin/sh
        - -c
        env:
        - name: KAFKA_ADMIN_USERNAME
          value: admin
        - name: KAFKA_ADMIN_PASSWORD
          valueFrom:
            secretKeyRef:
              key: system-user-password
              name: impt-kafka-jaas
        - name: KAFKA_SECURITY_PROTOCOL
          value: SASL_PLAINTEXT
        - name: KAFKA_SASL_MECHANISM
          value: PLAIN
        image: quay.io/prometheus/busybox:glibc
        imagePullPolicy: IfNotPresent
        name: store-credentials
        resources: {}
        securityContext:
          allowPrivilegeEscalation: false
          capabilities:
            drop:
            - ALL
          readOnlyRootFilesystem: true
          runAsNonRoot: true
          runAsUser: 10001
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
        volumeMounts:
        - mountPath: /tmp/kafka
          name: client-config
      restartPolicy: Never
      schedulerName: default-scheduler
      securityContext: {}
      serviceAccount: impt-kafka-provisioning
      serviceAccountName: impt-kafka-provisioning
      terminationGracePeriodSeconds: 30
      volumes:
      - emptyDir: {}
        name: kafka-logs
      - emptyDir: {}
        name: client-config
      - configMap:
          defaultMode: 420
          name: impt-kafka-topics
        name: kafka-topics

2. Start the job by executing the following command: sudo kubectl apply -n impt -f kafka-provisioning.yaml

3. Wait until the job is finished. You can check it by executing the following command : sudo kubectl get jobs -n impt | grep impt-kafka-provisioning-retry and wait until the status of this job is set as Complete

4. When the jobs is finished, delete it - by issuing the following command: sudo kubectl delete job impt-kafka-provisioning-retry -n impt