|
| 1 | +--- |
| 2 | +apiVersion: batch/v1 |
| 3 | +kind: Job |
| 4 | +metadata: |
| 5 | + name: start-kafka-job |
| 6 | +spec: |
| 7 | + template: |
| 8 | + spec: |
| 9 | + containers: |
| 10 | + - name: start-kafka-job |
| 11 | + image: oci.stackable.tech/sdp/tools:1.0.0-stackable0.0.0-dev |
| 12 | + env: |
| 13 | + - name: NAMESPACE |
| 14 | + valueFrom: |
| 15 | + fieldRef: |
| 16 | + fieldPath: metadata.namespace |
| 17 | + # N.B. it is possible for the scheduler to report that a DAG exists, |
| 18 | + # only for the worker task to fail if a pod is unexpectedly |
| 19 | + # restarted. The wait/watch steps below are not "water-tight" but add |
| 20 | + # a layer of stability by at least ensuring that the cluster is |
| 21 | + # initialized and ready and that all pods are reachable (albeit |
| 22 | + # independent of each other). |
| 23 | + command: |
| 24 | + - bash |
| 25 | + - -euo |
| 26 | + - pipefail |
| 27 | + - -c |
| 28 | + - | |
| 29 | + # Kafka: wait for cluster |
| 30 | + kubectl rollout status --watch statefulset/kafka-broker-default |
| 31 | + kubectl rollout status --watch statefulset/kafka-controller-default |
| 32 | +
|
| 33 | + # Kafka: create consumer offsets topics (required for group coordinator) |
| 34 | + kubectl exec kafka-broker-default-0 -c kafka -- \ |
| 35 | + /stackable/kafka/bin/kafka-topics.sh \ |
| 36 | + --bootstrap-server kafka-broker-default-0-listener-broker.$(NAMESPACE).svc.cluster.local:9093 \ |
| 37 | + --create \ |
| 38 | + --topic __consumer_offsets \ |
| 39 | + --partitions 50 \ |
| 40 | + --replication-factor 1 \ |
| 41 | + --config cleanup.policy=compact \ |
| 42 | + --command-config /stackable/config/client.properties |
| 43 | +
|
| 44 | + # Airflow: wait for cluster |
| 45 | + kubectl rollout status --watch statefulset/airflow-webserver-default |
| 46 | + kubectl rollout status --watch statefulset/airflow-scheduler-default |
| 47 | +
|
| 48 | + # Airflow: activate DAG |
| 49 | + AIRFLOW_ADMIN_PASSWORD=$(cat /airflow-credentials/adminUser.password) |
| 50 | + ACCESS_TOKEN=$(curl -XPOST http://airflow-webserver-default-headless:8080/auth/token -H 'Content-Type: application/json' -d '{"username": "admin", "password": "'$AIRFLOW_ADMIN_PASSWORD'"}' | jq -r .access_token) |
| 51 | + curl -H "Authorization: Bearer $ACCESS_TOKEN" -H 'Content-Type: application/json' -XPATCH http://airflow-webserver-default-headless:8080/api/v2/dags/kafka_watcher -d '{"is_paused": false}' | jq |
| 52 | +
|
| 53 | + # Kafka: produce a message to create the topic |
| 54 | + kubectl exec kafka-broker-default-0 -c kafka -- bash -c \ |
| 55 | + 'echo "Hello World at: $(date)" | /stackable/kafka/bin/kafka-console-producer.sh \ |
| 56 | + --bootstrap-server kafka-broker-default-0-listener-broker.$(NAMESPACE).svc.cluster.local:9093 \ |
| 57 | + --producer.config /stackable/config/client.properties \ |
| 58 | + --topic test-topic' |
| 59 | + volumeMounts: |
| 60 | + - name: airflow-credentials |
| 61 | + mountPath: /airflow-credentials |
| 62 | + volumes: |
| 63 | + - name: airflow-credentials |
| 64 | + secret: |
| 65 | + secretName: airflow-credentials |
| 66 | + restartPolicy: OnFailure |
| 67 | + backoffLimit: 20 # give some time for the Airflow cluster to be available |
0 commit comments