Enable Kafka transactions in generic e2e

Author: jiangpengcheng

.ci/clusters/values_skywalking_e2e_cluster.yaml

@@ -64,7 +64,7 @@ zookeeper:
       cpu: 50m
 
 bookkeeper:
-  replicaCount: 0
+  replicaCount: 1
   metadata:
     image:
       repository: streamnative/sn-platform
@@ -116,8 +116,10 @@ broker:
     autoSkipNonRecoverableData: "true"
     messagingProtocols: "kafka"
     protocolHandlerDirectory: "./protocols"
-    kafkaListeners: "PLAINTEXT://0.0.0.0:9092"
-    kafkaAdvertisedListeners: "PLAINTEXT://sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092"
+    PULSAR_PREFIX_kafkaListeners: "PLAINTEXT://0.0.0.0:9092"
+    PULSAR_PREFIX_kafkaAdvertisedListeners: "PLAINTEXT://sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092"
+    transactionCoordinatorEnabled: "true"
+    systemTopicEnabled: "true"
     # storage settings
     managedLedgerDefaultEnsembleSize: "1"
     managedLedgerDefaultWriteQuorum: "1"

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

@@ -32,7 +32,7 @@ spec:
     outputSchemaConfig:
       type: string
   genericRuntime:
-    functionFile: /pulsar/functions/exclamation_function.py
+    functionFile: exclamation_function.py
     functionFileLocation: function://public/default/test-py-function
     language: python
   pulsarPackageService:

.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}"
-ci::ensure_kafka_topic "${output_topic}" "${kafka_bootstrap_server}" "${kafka_properties_file}"
+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 || {