Fix Kafka generic function e2e verification

Author: jiangpengcheng

.ci/helm.sh

@@ -371,6 +371,56 @@ function ci::verify_exclamation_function() {
     return 1
 }
 
+function ci::ensure_kafka_topic() {
+    topic=$1
+    kafka_bootstrap_server=$2
+    properties_file=$3
+    kubectl exec -n ${NAMESPACE} kafka-client -- kafka-topics.sh \
+        --bootstrap-server "${kafka_bootstrap_server}" \
+        --create \
+        --if-not-exists \
+        --topic "${topic}" \
+        --command-config "/opt/bitnami/kafka/config/${properties_file}" || true
+}
+
+function ci::verify_kafka_exclamation_function() {
+    inputtopic=$1
+    outputtopic=$2
+    inputmessage=$3
+    outputmessage=$4
+    kafka_bootstrap_server=$5
+    properties_file=$6
+    consumer_group="function-mesh-${RANDOM}-$(date +%s)"
+    output_file=$(mktemp)
+
+    kubectl exec -n ${NAMESPACE} kafka-client -- kafka-console-consumer.sh \
+        --bootstrap-server "${kafka_bootstrap_server}" \
+        --consumer.config "/opt/bitnami/kafka/config/${properties_file}" \
+        --topic "${outputtopic}" \
+        --group "${consumer_group}" \
+        --timeout-ms 30000 \
+        --max-messages 1 > "${output_file}" &
+    consumer_pid=$!
+
+    sleep 3
+    kubectl exec -n ${NAMESPACE} kafka-client -- bash -c \
+        "echo \"${inputmessage}\" | kafka-console-producer.sh --bootstrap-server ${kafka_bootstrap_server} --producer.config /opt/bitnami/kafka/config/${properties_file} --topic ${inputtopic}"
+
+    if ! wait "${consumer_pid}"; then
+        cat "${output_file}" || true
+        rm -f "${output_file}"
+        return 1
+    fi
+
+    MESSAGE=$(cat "${output_file}")
+    rm -f "${output_file}"
+    echo "$MESSAGE"
+    if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
+        return 0
+    fi
+    return 1
+}
+
 function ci::verify_exclamation_function_with_auth() {
     inputtopic=$1
     outputtopic=$2

.ci/tests/integration/cases/generic-kafka-function/kafka-client.yaml

@@ -0,0 +1,29 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: generic-kafka-client-config
+  namespace: default
+data:
+  kafka.properties: |
+    security.protocol=PLAINTEXT
+---
+apiVersion: v1
+kind: Pod
+metadata:
+  name: kafka-client
+  namespace: default
+spec:
+  containers:
+  - name: kafka
+    image: bitnamilegacy/kafka:3.4.1
+    command: ["/bin/sh", "-c", "--"]
+    args: ["while true; do sleep 30; done;"]
+    volumeMounts:
+    - name: config-volume
+      mountPath: /opt/bitnami/kafka/config/kafka.properties
+      subPath: kafka.properties
+      readOnly: true
+  volumes:
+  - name: config-volume
+    configMap:
+      name: generic-kafka-client-config

.ci/tests/integration/cases/generic-kafka-function/verify.sh

@@ -33,26 +33,49 @@ if [ ! "$KUBECONFIG" ]; then
 fi
 
 manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/generic-kafka-function/manifests.yaml
+kafka_client_file="${BASE_DIR}"/.ci/tests/integration/cases/generic-kafka-function/kafka-client.yaml
+kafka_bootstrap_server=sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092
+kafka_properties_file=kafka.properties
+input_topic=input-kafka-topic
+output_topic=output-kafka-topic
+input_message="test-message-${RANDOM}-$(date +%s)"
+output_message="${input_message}!"
 
+kubectl apply -f "${kafka_client_file}" > /dev/null 2>&1
+kubectl wait pod kafka-client --for=condition=Ready --timeout=2m || {
+  kubectl get pod kafka-client -o yaml || true
+  kubectl delete -f "${kafka_client_file}" > /dev/null 2>&1 || true
+  exit 1
+}
+
+ci::ensure_kafka_topic "${input_topic}" "${kafka_bootstrap_server}" "${kafka_properties_file}" > /dev/null 2>&1
+ci::ensure_kafka_topic "${output_topic}" "${kafka_bootstrap_server}" "${kafka_properties_file}" > /dev/null 2>&1
 kubectl apply -f "${manifests_file}" > /dev/null 2>&1
 
 kubectl wait -l compute.functionmesh.io/name=generic-kafka-function --for=condition=Ready pod --timeout=2m || {
   kubectl get pods -l compute.functionmesh.io/name=generic-kafka-function
   kubectl logs -l compute.functionmesh.io/name=generic-kafka-function --tail=100 || true
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  kubectl delete -f "${kafka_client_file}" > /dev/null 2>&1 || true
   exit 1
 }
 
+set +e
 verify_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} \
-  ci::verify_exclamation_function "persistent://public/default/input-kafka-topic" \
-  "persistent://public/default/output-kafka-topic" "test-message" "test-message!" 10 2>&1)
-if [ $? -eq 0 ]; then
+  ci::verify_kafka_exclamation_function "${input_topic}" "${output_topic}" \
+  "${input_message}" "${output_message}" "${kafka_bootstrap_server}" "${kafka_properties_file}" 2>&1)
+verify_status=$?
+set -e
+
+if [ ${verify_status} -eq 0 ]; then
   echo "e2e-test: ok" | yq eval -
 else
   echo "$verify_result"
   kubectl logs -l compute.functionmesh.io/name=generic-kafka-function --tail=100 || true
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  kubectl delete -f "${kafka_client_file}" > /dev/null 2>&1 || true
   exit 1
 fi
 
 kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+kubectl delete -f "${kafka_client_file}" > /dev/null 2>&1 || true