Enable KoP in generic Kafka e2e

Author: jiangpengcheng

.ci/clusters/values_skywalking_e2e_cluster.yaml

@@ -31,6 +31,7 @@ components:
   pulsar_manager: false
   sql_worker: false
   proxy: false
+  kop: true
 
 ## disable monitoring stack
 monitoring:

.ci/helm.sh

@@ -375,12 +375,26 @@ function ci::ensure_kafka_topic() {
     topic=$1
     kafka_bootstrap_server=$2
     properties_file=$3
+    for attempt in $(seq 1 30); do
+        if kubectl exec -n ${NAMESPACE} kafka-client -- kafka-topics.sh \
+            --bootstrap-server "${kafka_bootstrap_server}" \
+            --command-config "/opt/bitnami/kafka/config/${properties_file}" \
+            --list > /dev/null 2>&1; then
+            break
+        fi
+        if [ "${attempt}" -eq 30 ]; then
+            echo "Kafka broker ${kafka_bootstrap_server} is not ready"
+            kubectl get pods -n ${NAMESPACE} -o wide || true
+            return 1
+        fi
+        sleep 5
+    done
     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
+        --command-config "/opt/bitnami/kafka/config/${properties_file}"
 }
 
 function ci::verify_kafka_exclamation_function() {

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

@@ -48,8 +48,8 @@ kubectl wait pod kafka-client --for=condition=Ready --timeout=2m || {
   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
+ci::ensure_kafka_topic "${input_topic}" "${kafka_bootstrap_server}" "${kafka_properties_file}"
+ci::ensure_kafka_topic "${output_topic}" "${kafka_bootstrap_server}" "${kafka_properties_file}"
 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 || {