backport: lake demo, wait for resources (#268)

* Improve resource/topic wait script (#261)

Co-authored-by: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>

* update broker url

---------

Co-authored-by: Malte Sander <malte.sander.it@gmail.com>

Author: razvan

demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml

@@ -11,9 +11,85 @@ spec:
     spec:
       serviceAccountName: demo-serviceaccount
       initContainers:
-        - name: wait-for-kafka
+        - name: wait-for-resources
           image: oci.stackable.tech/sdp/tools:1.0.0-stackable25.3.0
-          command: ["bash", "-c", "echo 'Waiting for all kafka brokers to be ready' && kubectl wait --for=condition=ready --timeout=30m pod -l app.kubernetes.io/name=kafka -l app.kubernetes.io/instance=kafka"]
+          command:
+            - bash
+            - -euo
+            - pipefail
+            - -c
+            - |
+              echo 'Waiting for all minio instances to be ready'
+              kubectl wait --for=condition=ready --timeout=30m pod -l app=minio,release=minio,stackable.tech/vendor=Stackable
+              echo 'Waiting for all kafka brokers to be ready'
+              kubectl wait --for=condition=ready --timeout=30m pod -l app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka
+              echo 'Waiting for all nifi instances to be ready'
+              kubectl wait --for=condition=ready --timeout=30m pod -l app.kubernetes.io/name=nifi,app.kubernetes.io/instance=nifi
+        - name: wait-for-kafka-topics
+          image: oci.stackable.tech/sdp/kafka:3.9.0-stackable25.3.0
+          command:
+            - bash
+            - -euo
+            - pipefail
+            - -c
+            - |
+              #!/usr/bin/bash
+              #
+              # Wait for Kafka topics created by the Nifi workflows to be ready.
+              # Also wait for all topic partitions to have a leader.
+              # This is required for the Spark streaming job to be able to read from the topics.
+              # Without this check, the Spark job might hang indefinitely without processing any events.
+
+              BROKER="${BROKER:-kafka-broker-default:9093}"
+
+              log() {
+                level="$1"
+                shift
+                echo "[$level] $*"
+              }
+
+              check_leaders() {
+                local topic=$1
+                local failed=0
+
+                log INFO "Starting leader check on Kafka broker: $BROKER for topic: $topic"
+                metadata=$(kcat -b "$BROKER" -X security.protocol=SSL -X ssl.key.location=/stackable/tls-kcat/tls.key -X ssl.certificate.location=/stackable/tls-kcat/tls.crt -X ssl.ca.location=/stackable/tls-kcat/ca.crt -L -t "$topic" 2>/dev/null)
+
+                if [[ -z "$metadata" ]]; then
+                  log ERROR "Failed to retrieve metadata for topic: $topic"
+                  return 1
+                fi
+
+                log DEBUG "Metadata for $topic:"
+                echo "$metadata"
+
+                if echo "$metadata" | grep -q 'leader: -1'; then
+                  log ERROR "Found 'leader: -1' in topic '$topic'; topic not ready yet!"
+                  return 1
+                fi
+
+                if echo "$metadata" | grep -q 'Broker: Leader not available'; then
+                  log ERROR "Topic '$topic' not available yet"
+                  return 1
+                fi
+
+                log INFO "Check topic '$topic' was successful"
+                return 0
+              }
+
+              for topic in "shared_bikes_bike_status" "shared_bikes_station_status" "shared_bikes_station_information" "water_levels_measurements" "water_levels_stations"
+              do
+                result=$(check_leaders "$topic")
+                echo "$result"
+                if [ "$result" == "1" ]
+                then
+                  exit 1
+                fi
+              done
+              exit 0
+          volumeMounts:
+            - name: tls-kcat
+              mountPath: /stackable/tls-kcat
       containers:
         - name: create-spark-ingestion-job
           image: oci.stackable.tech/sdp/tools:1.0.0-stackable25.3.0
@@ -25,7 +101,26 @@ spec:
         - name: manifest
           configMap:
             name: create-spark-ingestion-job-manifest
+        - name: tls-kcat
+          ephemeral:
+            volumeClaimTemplate:
+              metadata:
+                annotations:
+                  secrets.stackable.tech/backend.autotls.cert.lifetime: "1d"
+                  secrets.stackable.tech/class: "tls"
+                  secrets.stackable.tech/format: "tls-pem"
+                  secrets.stackable.tech/scope: "pod"
+              spec:
+                accessModes:
+                  - ReadWriteOnce
+                resources:
+                  requests:
+                    storage: "1"
+                storageClassName: secrets.stackable.tech
+                volumeMode: Filesystem
       restartPolicy: OnFailure
+      securityContext:
+        fsGroup: 1000
   backoffLimit: 50
 ---
 apiVersion: v1