From f0e327418d5a53e297e08188b41585ed86ebb439 Mon Sep 17 00:00:00 2001 From: Sebastian Bernauer Date: Wed, 11 Nov 2026 16:24:55 +0100 Subject: [PATCH 1/2] chore(lakehouse): Use new KafkaTopic resource --- .../create-kafka-topics.yaml | 58 ++++++++++ .../create-nifi-ingestion-job.yaml | 11 +- .../create-spark-ingestion-job.yaml | 100 ++---------------- .../serviceaccount.yaml | 29 +++-- demos/demos-v2.yaml | 1 + .../kafka.yaml | 20 +++- 6 files changed, 110 insertions(+), 109 deletions(-) create mode 100644 demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml diff --git a/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml b/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml new file mode 100644 index 00000000..8023894d --- /dev/null +++ b/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml @@ -0,0 +1,58 @@ +# We can't simply create the KafkaTopic objects here (or in the stack), as Kafka only distributes the partitions of a +# topic among the brokers that are running when the topic is created. If the topics are created before all brokers are +# up, some brokers end up without any partitions. +# +# This is a known issue for kafka-operator, we will document this (TODO link to docs PR) . +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: create-kafka-topics +spec: + template: + spec: + serviceAccountName: demo-serviceaccount + containers: + - name: create-kafka-topics + image: oci.stackable.tech/sdp/tools:1.0.0-stackable0.0.0-dev + command: + - bash + - -euo + - pipefail + - -c + - | + echo 'Waiting for the kafka cluster to be available' + kubectl wait --for=condition=available --timeout=30m kafkaclusters.kafka.stackable.tech/kafka -n {{ NAMESPACE }} + echo 'Creating kafka topics' + kubectl apply -f /tmp/manifest/kafka-topics.yaml -n {{ NAMESPACE }} + echo 'Waiting for all kafka topics to be ready' + kubectl wait --for=condition=ready --timeout=30m -f /tmp/manifest/kafka-topics.yaml -n {{ NAMESPACE }} + volumeMounts: + - name: manifest + mountPath: /tmp/manifest + volumes: + - name: manifest + configMap: + name: create-kafka-topics-manifest + restartPolicy: OnFailure + backoffLimit: 50 +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: create-kafka-topics-manifest +data: + kafka-topics.yaml: | + # TODO: Switch to KafkaTopicTemplate (or however it's called) once it's supported + # {% for topic in ["shared_bikes_bike_status", "water_levels_measurements", "shared_bikes_station_status", "water_levels_stations", "shared_bikes_station_information"] %} + --- + apiVersion: kafka.stackable.tech/v1alpha1 + kind: KafkaTopic + metadata: + name: {{ topic | replace(from="_", to="-") }} + spec: + clusterRef: + name: kafka + topicName: {{ topic }} + deletionPolicy: Delete + # {% endfor %} diff --git a/demos/data-lakehouse-iceberg-trino-spark/create-nifi-ingestion-job.yaml b/demos/data-lakehouse-iceberg-trino-spark/create-nifi-ingestion-job.yaml index dfe3cf19..27efc213 100644 --- a/demos/data-lakehouse-iceberg-trino-spark/create-nifi-ingestion-job.yaml +++ b/demos/data-lakehouse-iceberg-trino-spark/create-nifi-ingestion-job.yaml @@ -8,7 +8,7 @@ spec: spec: serviceAccountName: demo-serviceaccount initContainers: - - name: wait-for-kafka + - name: wait-for-resources image: oci.stackable.tech/sdp/tools:1.0.0-stackable0.0.0-dev command: - bash @@ -16,10 +16,11 @@ spec: - pipefail - -c - | - echo 'Waiting for all nifi nodes to be ready' - kubectl wait --for=condition=ready --timeout=30m pod -l app.kubernetes.io/instance=nifi,app.kubernetes.io/name=nifi -n {{ NAMESPACE }} - echo 'Waiting for all kafka brokers to be ready' - kubectl wait --for=condition=ready --timeout=30m pod -l app.kubernetes.io/instance=kafka,app.kubernetes.io/name=kafka -n {{ NAMESPACE }} + echo 'Waiting for the nifi cluster to be available' + kubectl wait --for=condition=available --timeout=30m nificlusters.nifi.stackable.tech/nifi -n {{ NAMESPACE }} + # The NiFi flow can only publish once the topics exist, as Kafka doesn't auto-create them + echo 'Waiting for the kafka topics to be created' + kubectl wait --for=condition=complete --timeout=30m job/create-kafka-topics -n {{ NAMESPACE }} containers: - name: create-nifi-ingestion-job image: oci.stackable.tech/sdp/testing-tools/nifi:0.3.0-stackable0.0.0-dev diff --git a/demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml b/demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml index 95424376..756e5d75 100644 --- a/demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml +++ b/demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml @@ -1,6 +1,5 @@ -# We can't simply create the SparkApplication object here as we have to wait for Kafka to be ready because -# * We currently don't restart failed Spark applications (see https://github.com/stackabletech/spark-k8s-operator/issues/157) -# * We currently auto-create topics and we need all the brokers to be available so that the topic is distributed among all the brokers +# We can't simply create the SparkApplication object here as we have to wait for the Kafka topics to be ready because +# we currently don't restart failed Spark applications (see https://github.com/stackabletech/spark-k8s-operator/issues/157) --- apiVersion: batch/v1 kind: Job @@ -21,75 +20,9 @@ spec: - | echo 'Waiting for Garage to be ready' kubectl wait --for=condition=ready --timeout=30m pod -l app=garage -n {{ NAMESPACE }} - 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 -n {{ NAMESPACE }} - 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 -n {{ NAMESPACE }} - - name: wait-for-kafka-topics - image: oci.stackable.tech/sdp/kafka:4.3.1-stackable0.0.0-dev - 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-0-listener-broker: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 + # Without the topics, the Spark job might hang indefinitely without processing any events + echo 'Waiting for the kafka topics to be created' + kubectl wait --for=condition=complete --timeout=30m job/create-kafka-topics -n {{ NAMESPACE }} containers: - name: create-spark-ingestion-job image: oci.stackable.tech/sdp/tools:1.0.0-stackable0.0.0-dev @@ -107,26 +40,7 @@ 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 @@ -145,13 +59,13 @@ data: stackable.tech/demo: {{ DEMO }} spec: sparkImage: - # Iceberg 1.11.0 only supports Spark 4.1.x + # Iceberg 1.12.0 only supports Spark 4.1.x productVersion: 4.1.3 mode: cluster mainApplicationFile: local:///stackable/spark/jobs/spark-ingest-into-lakehouse.py deps: packages: - - org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.11.0 + - org.apache.iceberg:iceberg-spark-runtime-4.1_2.13:1.12.0 - org.apache.spark:spark-sql-kafka-0-10_2.13:4.1.2 s3connection: reference: garage diff --git a/demos/data-lakehouse-iceberg-trino-spark/serviceaccount.yaml b/demos/data-lakehouse-iceberg-trino-spark/serviceaccount.yaml index 5b50b941..32ef8190 100644 --- a/demos/data-lakehouse-iceberg-trino-spark/serviceaccount.yaml +++ b/demos/data-lakehouse-iceberg-trino-spark/serviceaccount.yaml @@ -31,27 +31,44 @@ rules: - list - watch - apiGroups: - - apps + - batch resources: - - statefulsets + - jobs verbs: - get - list - watch - apiGroups: - - batch + - spark.stackable.tech resources: - - jobs + - sparkapplications verbs: - get - list - watch + - create - apiGroups: - - spark.stackable.tech + - kafka.stackable.tech resources: - - sparkapplications + - kafkaclusters + verbs: + - get + - list + - watch + - apiGroups: + - kafka.stackable.tech + resources: + - kafkatopics verbs: - get - list - watch - create + - apiGroups: + - nifi.stackable.tech + resources: + - nificlusters + verbs: + - get + - list + - watch diff --git a/demos/demos-v2.yaml b/demos/demos-v2.yaml index 4d56eed1..c23f1916 100644 --- a/demos/demos-v2.yaml +++ b/demos/demos-v2.yaml @@ -213,6 +213,7 @@ demos: - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/serviceaccount.yaml - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/load-test-data.yaml - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/create-trino-tables.yaml + - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/create-nifi-ingestion-job.yaml - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/create-spark-ingestion-job.yaml - plainYaml: https://raw.githubusercontent.com/stackabletech/demos/main/demos/data-lakehouse-iceberg-trino-spark/setup-superset.yaml diff --git a/stacks/data-lakehouse-iceberg-trino-spark/kafka.yaml b/stacks/data-lakehouse-iceberg-trino-spark/kafka.yaml index db576c30..b909c896 100644 --- a/stacks/data-lakehouse-iceberg-trino-spark/kafka.yaml +++ b/stacks/data-lakehouse-iceberg-trino-spark/kafka.yaml @@ -9,7 +9,20 @@ spec: clusterConfig: authentication: - authenticationClass: kafka-client-tls + managementAccess: + enabled: true + authentication: + userProvided: # TODO: Switch to automatic once supported + tls: + secretClassName: tls controllers: + configOverrides: + controller.properties: &sharedProperties + # We want to create the Topics using the KafkaTopic CRD + auto.create.topics.enable: "false" + num.partitions: "32" + log.segment.bytes: "50000000" # 50 MB + log.retention.bytes: "2000000000" # 2 GB. Should keep between 2.0 and 2.5GB roleGroups: default: replicas: 1 @@ -26,14 +39,11 @@ spec: max: "3" memory: limit: 2Gi + configOverrides: + broker.properties: *sharedProperties roleGroups: default: replicas: 5 - configOverrides: - broker.properties: - num.partitions: "27" - log.segment.bytes: "50000000" # 50 MB - log.retention.bytes: "2000000000" # 2 GB. Should keep between 2.0 and 2.5GB --- apiVersion: authentication.stackable.tech/v1alpha1 kind: AuthenticationClass From 49de7f3e2cb889cf83beeadf2b0a342d99bd8cd6 Mon Sep 17 00:00:00 2001 From: Sebastian Bernauer Date: Thu, 12 Nov 2026 10:01:32 +0100 Subject: [PATCH 2/2] Apply suggestion from @sbernauer --- .../data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml b/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml index 8023894d..00b8cfc4 100644 --- a/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml +++ b/demos/data-lakehouse-iceberg-trino-spark/create-kafka-topics.yaml @@ -2,7 +2,7 @@ # topic among the brokers that are running when the topic is created. If the topics are created before all brokers are # up, some brokers end up without any partitions. # -# This is a known issue for kafka-operator, we will document this (TODO link to docs PR) . +# This is a known issue for kafka-operator and was documented in https://github.com/stackabletech/kafka-operator/pull/1057. --- apiVersion: batch/v1 kind: Job