Add Kafka integration tests and deploy-kafka EaaS task

Agent · Agent · commit cd02582371a8 · 2026-07-01T09:18:20.000Z
- Add three self-verifying integration tests that verify Kafka messages
  are published for compose-created, compose-tagged, and compose-untagged
  events. Tests are skipped when KAFKA_URL is unset.
- Add _get_kafka_end_offset() helper to snapshot the partition end offset
  before each action so only messages from that specific test are consumed.
- Add _consume_kafka_message() with configurable timeout and optional
  predicate for draining stale in-flight messages from prior tests.
- Use group_id=None in KafkaConsumer within _consume_kafka_message so
  the group-coordinator protocol is not triggered; with manual partition
  assignment and explicit seek the coordinator is not needed, and its
  extra TCP connections caused spurious ECONNREFUSED failures on the
  third test when the Kafka broker briefly shed connections.
- Catch KafkaConnectionError during message iteration and re-raise as
  AssertionError with a clear diagnostic message.
- Use a proper kafka.serializer.Deserializer subclass for JSON
  deserialization; subclassing the ABC prevents KafkaConsumer from
  wrapping it in DeserializeWrapper, which broke when the object was
  not callable.
- Add deploy-kafka task to integration-test-eaas.yaml using the official
  apache/kafka:3.9.2 image with emptyDir volumes for config, logs, and
  data so the container runs under OpenShift's restricted-v2 SCC.
- Configure CTS to use the Kafka backend in the EaaS test environment.
- Fix messaging.py to convert the string 'none' to Python None for the
  compression_type argument to KafkaProducer.
- Add kafka-python to test-requirements.txt.
- Enable CI triggers on the feature/integration-tests branch.

Generated-By: OpenCode (google-vertex-anthropic/claude-sonnet-4-6@default)
diff --git a/.github/workflows/gating.yaml b/.github/workflows/gating.yaml
@@ -2,9 +2,9 @@ name: Gating
 
 on:
   push:
-    branches: [ "main" ]
+    branches: [ "main", "feature/integration-tests" ]
   pull_request:
-    branches: [ "main" ]
+    branches: [ "main", "feature/integration-tests" ]
 
 jobs:
   tests:
diff --git a/.tekton/cts-pull-request.yaml b/.tekton/cts-pull-request.yaml
@@ -7,7 +7,7 @@ metadata:
     build.appstudio.redhat.com/pull_request_number: '{{pull_request_number}}'
     build.appstudio.redhat.com/target_branch: '{{target_branch}}'
     pipelinesascode.tekton.dev/max-keep-runs: "3"
-    pipelinesascode.tekton.dev/on-cel-expression: event == "pull_request" && target_branch == "main"
+    pipelinesascode.tekton.dev/on-cel-expression: event == "pull_request" && target_branch in ["main", "feature/integration-tests"]
   creationTimestamp:
   labels:
     appstudio.openshift.io/application: cts
diff --git a/.tekton/integration-test-eaas.yaml b/.tekton/integration-test-eaas.yaml
@@ -457,6 +457,145 @@ spec:
     - name: kubeconfig-secret
       value: $(tasks.provision-environment.results.secretRef)
 
+  - name: deploy-kafka
+    runAfter:
+    - provision-environment
+    taskSpec:
+      params:
+      - name: kubeconfig-secret
+        type: string
+      steps:
+      - name: create-kafka
+        image: quay.io/konflux-ci/appstudio-utils:latest
+        script: |
+          #!/usr/bin/env bash
+          set -euo pipefail
+
+          KUBECONFIG=/tmp/kubeconfig
+          kubectl get secret $(params.kubeconfig-secret) -o jsonpath='{.data.kubeconfig}' | base64 -d > $KUBECONFIG
+          export KUBECONFIG
+
+          echo "=========================================="
+          echo "Deploying Kafka (Apache KRaft single-node)"
+          echo "=========================================="
+
+          kubectl apply -f - <<'EOFYAML'
+          apiVersion: apps/v1
+          kind: Deployment
+          metadata:
+            name: kafka
+            labels:
+              app: kafka
+          spec:
+            replicas: 1
+            selector:
+              matchLabels:
+                app: kafka
+            template:
+              metadata:
+                labels:
+                  app: kafka
+              spec:
+                initContainers:
+                - name: copy-kafka-config
+                  image: docker.io/apache/kafka:3.9.2
+                  command: ["/bin/sh", "-c", "cp -r /opt/kafka/config/. /mnt/kafka-config/"]
+                  volumeMounts:
+                  - name: kafka-config
+                    mountPath: /mnt/kafka-config
+                containers:
+                - name: kafka
+                  image: docker.io/apache/kafka:3.9.2
+                  ports:
+                  - containerPort: 9092
+                    name: client
+                  - containerPort: 9093
+                    name: controller
+                  env:
+                  - name: KAFKA_NODE_ID
+                    value: "1"
+                  - name: KAFKA_PROCESS_ROLES
+                    value: "broker,controller"
+                  - name: KAFKA_CONTROLLER_QUORUM_VOTERS
+                    value: "1@localhost:9093"
+                  - name: KAFKA_LISTENERS
+                    value: "PLAINTEXT://:9092,CONTROLLER://:9093"
+                  - name: KAFKA_ADVERTISED_LISTENERS
+                    value: "PLAINTEXT://kafka:9092"
+                  - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
+                    value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
+                  - name: KAFKA_CONTROLLER_LISTENER_NAMES
+                    value: "CONTROLLER"
+                  - name: KAFKA_AUTO_CREATE_TOPICS_ENABLE
+                    value: "true"
+                  - name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
+                    value: "1"
+                  - name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
+                    value: "1"
+                  - name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
+                    value: "1"
+                  - name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
+                    value: "0"
+                  resources:
+                    requests:
+                      memory: "256Mi"
+                      cpu: "100m"
+                    limits:
+                      memory: "512Mi"
+                      cpu: "500m"
+                  readinessProbe:
+                    tcpSocket:
+                      port: 9092
+                    initialDelaySeconds: 20
+                    periodSeconds: 5
+                    timeoutSeconds: 3
+                    failureThreshold: 24
+                  volumeMounts:
+                  - name: kafka-config
+                    mountPath: /opt/kafka/config
+                  - name: kafka-logs
+                    mountPath: /tmp/kafka-logs
+                  - name: kafka-gc-logs
+                    mountPath: /opt/kafka/logs
+                volumes:
+                - name: kafka-config
+                  emptyDir: {}
+                - name: kafka-logs
+                  emptyDir: {}
+                - name: kafka-gc-logs
+                  emptyDir: {}
+          ---
+          apiVersion: v1
+          kind: Service
+          metadata:
+            name: kafka
+            labels:
+              app: kafka
+          spec:
+            ports:
+            - port: 9092
+              targetPort: 9092
+              name: client
+            - port: 9093
+              targetPort: 9093
+              name: controller
+            selector:
+              app: kafka
+          EOFYAML
+
+          echo "Waiting for Kafka to be ready..."
+          if ! kubectl wait --for=condition=available --timeout=300s deployment/kafka; then
+            echo "Kafka deployment failed! Debug info:"
+            kubectl describe deployment kafka
+            kubectl describe pod -l app=kafka
+            kubectl logs -l app=kafka --tail=50 || echo "No logs available"
+            exit 1
+          fi
+          echo "✓ Kafka is ready"
+    params:
+    - name: kubeconfig-secret
+      value: $(tasks.provision-environment.results.secretRef)
+
   - name: deploy-database
     runAfter:
     - provision-environment
@@ -597,6 +736,7 @@ spec:
     - deploy-database
     - deploy-openldap
     - deploy-dex
+    - deploy-kafka
     taskSpec:
       params:
       - name: kubeconfig-secret
@@ -643,6 +783,14 @@ spec:
                   ]
                   ADMINS = {"groups": [], "users": ["builder@example.com"]}
                   ALLOWED_BUILDERS = {"groups": [], "users": ["builder@example.com"]}
+                  MESSAGING_BACKEND = "kafka"
+                  MESSAGING_BROKER_URLS = ["kafka:9092"]
+                  MESSAGING_KAFKA_SECURITY_PROTOCOL = "PLAINTEXT"
+                  MESSAGING_KAFKA_SASL_MECHANISM = ""
+                  MESSAGING_KAFKA_USERNAME = ""
+                  MESSAGING_KAFKA_PASSWORD = ""
+                  MESSAGING_KAFKA_COMPRESSION_TYPE = "none"
+                  MESSAGING_TOPIC_PREFIX = "cts."
             httpd.conf: |
               ServerRoot "/etc/httpd"
               PidFile /tmp/httpd.pid
@@ -916,9 +1064,9 @@ spec:
             echo 'Installing Dex CA certificate...'
             echo '$DEX_CA_B64' | base64 -d > /tmp/dex-ca.crt
 
-            echo 'Installing pytest and requests...'
+            echo 'Installing pytest, requests, and kafka-python...'
             python3 -m ensurepip
-            python3 -m pip install --target /tmp/test-deps --quiet pytest requests
+            python3 -m pip install --target /tmp/test-deps --quiet pytest requests kafka-python
 
             echo ''
             echo 'Cloning repository...'
@@ -930,7 +1078,7 @@ spec:
 
             echo ''
             echo 'Running pytest...'
-            PYTHONPATH=/tmp/test-deps REQUESTS_CA_BUNDLE=/tmp/dex-ca.crt CTS_URL=http://cts:8080 AUTH_BACKEND=oidc_or_kerberos DEX_URL=https://dex:5556 python3 -m pytest tests/test_integration_api.py -v -s -o addopts=
+            PYTHONPATH=/tmp/test-deps REQUESTS_CA_BUNDLE=/tmp/dex-ca.crt CTS_URL=http://cts:8080 AUTH_BACKEND=oidc_or_kerberos DEX_URL=https://dex:5556 KAFKA_URL=kafka:9092 python3 -m pytest tests/test_integration_api.py -v -s -o addopts=
           "
           TEST_RESULT=$?
           set -e
diff --git a/cts/messaging.py b/cts/messaging.py
@@ -102,9 +102,15 @@ def _kafka_send_msg(msgs):
 
     def _send():
         """Inner function to send messages (will be retried on failure)"""
+        compression = conf.messaging_kafka_compression_type
+        # kafka-python uses Python None to mean "no compression"; the string
+        # "none" (which may come from a config file) is not accepted.
+        if compression and compression.lower() == "none":
+            compression = None
+
         config = {
             "bootstrap_servers": conf.messaging_broker_urls,
-            "compression_type": conf.messaging_kafka_compression_type,
+            "compression_type": compression,
             "security_protocol": conf.messaging_kafka_security_protocol,
             "sasl_mechanism": conf.messaging_kafka_sasl_mechanism,
             "sasl_plain_username": conf.messaging_kafka_username,
diff --git a/test-requirements.txt b/test-requirements.txt
@@ -9,3 +9,5 @@ tox
 # Let's update this package to avoid this problem.
 itsdangerous>=1.1.0
 freezegun
+# Required for Kafka integration tests (test_integration_api.py)
+kafka-python
diff --git a/tests/test_integration_api.py b/tests/test_integration_api.py
