Stabilize window function integration verification

Author: jiangpengcheng

.ci/helm.sh

@@ -348,10 +348,19 @@ function ci::verify_backlog() {
     topic=$1
     sub=$2
     expected=$3
-    BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats $topic | grep msgBacklog)
-    if [[ "$BACKLOG" == *"\"msgBacklog\" : $expected"* ]]; then
-       return 0
-    fi
+    for attempt in $(seq 1 30); do
+        if BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats "$topic" 2>&1); then
+            true
+        elif BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats "${topic}-partition-0" 2>&1); then
+            true
+        fi
+        if [[ "$BACKLOG" == *"\"msgBacklog\" : $expected"* ]]; then
+           return 0
+        fi
+        echo "Backlog for ${topic} subscription ${sub} is not ${expected}, retry ${attempt}/30"
+        echo "$BACKLOG"
+        sleep 2
+    done
     return 1
 }
 

.ci/tests/integration/cases/logging-window-function/verify.sh

@@ -33,6 +33,22 @@ if [ ! "$KUBECONFIG" ]; then
 fi
 
 manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/logging-window-function/manifests.yaml
+input_topic="persistent://public/default/window-function-input-topic"
+output_topic="persistent://public/default/window-function-output-topic"
+log_topic="persistent://public/default/window-function-logs"
+
+function delete_topic_if_exists() {
+  topic=$1
+  kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin topics delete "${topic}" -f > /dev/null 2>&1 || true
+  kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin topics delete-partitioned-topic "${topic}" -f > /dev/null 2>&1 || true
+  kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin topics delete "${topic}-partition-0" -f > /dev/null 2>&1 || true
+}
+
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+kubectl wait -l compute.functionmesh.io/name=window-function-sample --for=delete pod --timeout=2m > /dev/null 2>&1 || true
+delete_topic_if_exists "${input_topic}"
+delete_topic_if_exists "${output_topic}"
+delete_topic_if_exists "${log_topic}"
 
 kubectl apply -f "${manifests_file}" > /dev/null 2>&1
 
@@ -51,7 +67,7 @@ if [ $? -ne 0 ]; then
   exit 1
 fi
 
-verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::send_test_data "persistent://public/default/window-function-input-topic" "test-message" 3 2>&1)
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::send_test_data "${input_topic}" "test-message" 3 2>&1)
 if [ $? -ne 0 ]; then
   echo "$verify_java_result"
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
@@ -61,15 +77,15 @@ fi
 sleep 3
 
 # the 3 messages will not be processed, so backlog should be 3
-verify_backlog_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_backlog "persistent://public/default/window-function-input-topic" "public/default/window-function-sample" 3 2>&1)
+verify_backlog_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_backlog "${input_topic}" "public/default/window-function-sample" 3 2>&1)
 if [ $? -ne 0 ]; then
   echo "$verify_backlog_result"
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
   exit 1
 fi
 
 # it will fire the window with first 5 messages when get the 5th message, and then fire again with 10 messages when get 10th message
-verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::send_test_data "persistent://public/default/window-function-input-topic" "test-message" 7 2>&1)
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::send_test_data "${input_topic}" "test-message" 7 2>&1)
 if [ $? -ne 0 ]; then
   echo "$verify_java_result"
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
@@ -87,17 +103,29 @@ fi
 #  exit 1
 #fi
 
-verify_log_result=$(kubectl logs -l compute.functionmesh.io/name=window-function-sample --tail=-1 | grep -e "-window-log" | wc -l)
-if [ $verify_log_result -eq 15 ]; then
+verify_log_result=0
+for attempt in $(seq 1 30); do
+  verify_log_result=$(kubectl logs -l compute.functionmesh.io/name=window-function-sample --tail=-1 | grep -e "-window-log" | wc -l)
+  if [ "$verify_log_result" -eq 15 ]; then
+    break
+  fi
+  sleep 2
+done
+
+if [ "$verify_log_result" -eq 15 ]; then
   sub_name=$(echo $RANDOM | md5sum | head -c 20; echo;)
-  verify_log_topic_result=$(kubectl exec -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-broker-0 -- bin/pulsar-client consume -n 15 -s $sub_name --subscription-position Earliest "persistent://public/default/window-function-logs" | grep  -e "-window-log" | wc -l)
-  if [ $verify_log_topic_result -ne 0 ]; then
+  verify_log_topic_result=$(kubectl exec -n ${PULSAR_NAMESPACE} ${PULSAR_RELEASE_NAME}-pulsar-broker-0 -- bin/pulsar-client consume -n 15 -s $sub_name --subscription-position Earliest "${log_topic}" | grep  -e "-window-log" | wc -l)
+  if [ "$verify_log_topic_result" -eq 15 ]; then
     echo "e2e-test: ok" | yq eval -
   else
-    echo "$verify_log_topic_result"
+    echo "expected 15 window log topic messages, got ${verify_log_topic_result}" >&2
+    kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+    exit 1
   fi
 else
-  echo "$verify_log_result"
+  echo "expected 15 window log lines, got ${verify_log_result}" >&2
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
 fi
 
 kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true