apiVersion: batch/v1
kind: Job
metadata:
  name: kafka-smoke-test
  namespace: kafka
spec:
  ttlSecondsAfterFinished: 600
  backoffLimit: 3
  template:
    spec:
      restartPolicy: Never
      volumes:
        - name: ca-cert
          secret:
            secretName: kafka-cluster-ca-cert
            items:
              - key: ca.crt
                path: ca.crt
      containers:
        - name: smoke-test
          image: apache/kafka:4.1.0
          command: ["/bin/bash", "-c"]
          args:
            - |
              set -euo pipefail
              BOOTSTRAP="kafka-kafka-bootstrap.kafka.svc.cluster.local:9093"
              TOPIC="smoke-test"

              # Build client.properties injecting sasl.jaas.config from the Strimzi-managed secret
              cat > /tmp/client.properties <<EOF
              security.protocol=SASL_SSL
              ssl.truststore.type=PEM
              ssl.truststore.location=/ca/ca.crt
              sasl.mechanism=SCRAM-SHA-512
              sasl.jaas.config=${SASL_JAAS_CONFIG}
              EOF

              echo "==> [1/3] Creating topic '${TOPIC}' (idempotent)"
              /opt/kafka/bin/kafka-topics.sh \
                --bootstrap-server "${BOOTSTRAP}" \
                --command-config /tmp/client.properties \
                --create \
                --topic "${TOPIC}" \
                --partitions 1 \
                --replication-factor 1 \
                --if-not-exists
              echo "    Topic ready."

              echo "==> [2/3] Producing 5 messages"
              printf 'message-%s\n' 1 2 3 4 5 | \
                /opt/kafka/bin/kafka-console-producer.sh \
                  --bootstrap-server "${BOOTSTRAP}" \
                  --producer.config /tmp/client.properties \
                  --topic "${TOPIC}"
              echo "    5 messages produced."

              echo "==> [3/3] Consuming (from-beginning, max 5)"
              MESSAGES=$(/opt/kafka/bin/kafka-console-consumer.sh \
                --bootstrap-server "${BOOTSTRAP}" \
                --consumer.config /tmp/client.properties \
                --topic "${TOPIC}" \
                --from-beginning \
                --max-messages 5 \
                --timeout-ms 30000 2>/dev/null || true)

              echo "${MESSAGES}"
              COUNT=$(echo "${MESSAGES}" | grep -c '^message-' || true)

              if [ "${COUNT}" -eq 5 ]; then
                echo "==> SMOKE TEST PASSED (${COUNT}/5 messages)"
              else
                echo "==> SMOKE TEST FAILED: expected 5, got ${COUNT}"
                exit 1
              fi
          env:
            - name: SASL_JAAS_CONFIG
              valueFrom:
                secretKeyRef:
                  name: kafka-client
                  key: sasl.jaas.config
          volumeMounts:
            - name: ca-cert
              mountPath: /ca
              readOnly: true
          resources:
            requests:
              memory: 256Mi
              cpu: "100m"
            limits:
              memory: 512Mi
              cpu: "500m"