feat: support Kafka protocol for functions

.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:
@@ -63,7 +64,7 @@ zookeeper:
       cpu: 50m
 
 bookkeeper:
-  replicaCount: 0
+  replicaCount: 1
   metadata:
     image:
       repository: streamnative/sn-platform
@@ -113,6 +114,14 @@ broker:
     ## Enable `autoSkipNonRecoverableData` since bookkeeper is running
     ## without persistence
     autoSkipNonRecoverableData: "true"
+    messagingProtocols: "kafka"
+    protocolHandlerDirectory: "./protocols"
+    PULSAR_PREFIX_kafkaTransactionCoordinatorEnabled: "true"
+    PULSAR_PREFIX_brokerDeduplicationEnabled: "true"
+    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/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
 }
 
@@ -371,6 +380,121 @@ function ci::verify_exclamation_function() {
     return 1
 }
 
+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}" \
+        --replication-factor 1 \
+        --partitions 1 \
+        --command-config "/opt/bitnami/kafka/config/${properties_file}"
+}
+
+function ci::wait_kafka_topic_ready() {
+    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}" \
+            --describe \
+            --topic "${topic}" \
+            --command-config "/opt/bitnami/kafka/config/${properties_file}"; then
+            return 0
+        fi
+        echo "Kafka topic ${topic} is not ready, retry ${attempt}/30"
+        sleep 5
+    done
+    return 1
+}
+
+function ci::wait_kafka_group_coordinator_ready() {
+    consumer_group=$1
+    kafka_bootstrap_server=$2
+    properties_file=$3
+    output_file=$(mktemp)
+    for attempt in $(seq 1 30); do
+        if kubectl exec -n ${NAMESPACE} kafka-client -- kafka-consumer-groups.sh \
+            --bootstrap-server "${kafka_bootstrap_server}" \
+            --describe \
+            --group "${consumer_group}" \
+            --command-config "/opt/bitnami/kafka/config/${properties_file}" > "${output_file}" 2>&1; then
+            cat "${output_file}"
+            rm -f "${output_file}"
+            return 0
+        fi
+        cat "${output_file}"
+        if ! grep -q "CoordinatorLoadInProgress" "${output_file}"; then
+            if grep -Eiq "does not exist|not exist|not found|no active members" "${output_file}"; then
+                rm -f "${output_file}"
+                return 0
+            fi
+            rm -f "${output_file}"
+            return 1
+        fi
+        echo "Kafka consumer group coordinator for ${consumer_group} is loading, retry ${attempt}/30"
+        sleep 5
+    done
+    rm -f "${output_file}"
+    return 1
+}
+
+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/crypto-function/manifests.yaml

@@ -16,6 +16,7 @@ spec:
     typeClassName: java.lang.String
     sourceSpecs:
       "persistent://public/default/java-function-crypto-input-topic":
+        schemaType: STRING
         cryptoConfig:
           cryptoKeyReaderClassName: "org.apache.pulsar.functions.api.examples.RawFileKeyReader"
           cryptoKeyReaderConfig:

.ci/tests/integration/cases/crypto-function/verify.sh

@@ -33,19 +33,69 @@ if [ ! "$KUBECONFIG" ]; then
 fi
 
 manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/crypto-function/manifests.yaml
-kubectl apply -f ${manifests_file} > /dev/null 2>&1
+input_topic="persistent://public/default/java-function-crypto-input-topic"
+output_topic="persistent://public/default/java-function-crypto-output-topic"
 
-verify_fm_result=$(ci::verify_function_mesh java-function-crypto-sample 2>&1)
-if [ $? -ne 0 ]; then
-  echo "$verify_fm_result"
+function delete_topic() {
+  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 exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin schemas delete "${topic}" > /dev/null 2>&1 || true
+}
+
+function cleanup() {
   kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  delete_topic "${input_topic}"
+  delete_topic "${output_topic}"
+}
+
+function upload_string_schema() {
+  kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- sh -c \
+    'printf "%s\n" "{\"type\":\"STRING\",\"schema\":\"\",\"properties\":{}}" > /tmp/function-mesh-string-schema.json && bin/pulsar-admin schemas upload --filename /tmp/function-mesh-string-schema.json "$1"' \
+    sh "${input_topic}" > /dev/null
+}
+
+function wait_string_schema() {
+  for attempt in $(seq 1 30); do
+    if kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- \
+      bin/pulsar-admin schemas get "${input_topic}" > /dev/null 2>&1; then
+      return 0
+    fi
+    echo "Schema for ${input_topic} is not ready, retry ${attempt}/30" >&2
+    sleep 2
+  done
+  return 1
+}
+
+trap cleanup EXIT
+cleanup
+
+if ! create_topic_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::create_topic "${input_topic}" 2>&1); then
+  echo "$create_topic_result" >&2
+  exit 1
+fi
+
+if ! upload_schema_result=$(upload_string_schema 2>&1); then
+  echo "$upload_schema_result" >&2
   exit 1
 fi
 
-verify_crypto_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_crypto_function 2>&1)
-if [ $? -eq 0 ]; then
+if ! wait_schema_result=$(wait_string_schema 2>&1); then
+  echo "$wait_schema_result" >&2
+  exit 1
+fi
+
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+if ! verify_fm_result=$(ci::verify_function_mesh java-function-crypto-sample 2>&1); then
+  echo "$verify_fm_result" >&2
+  exit 1
+fi
+
+if verify_crypto_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_crypto_function 2>&1); then
   echo "e2e-test: ok" | yq eval -
 else
-  echo "$verify_crypto_result"
+  echo "$verify_crypto_result" >&2
+  exit 1
 fi
-kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

.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/manifests.yaml

@@ -0,0 +1,56 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: generic-kafka-function
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-generic-python-runner:latest
+  className: exclamation_function.ExclamationFunction
+  forwardSourceMessageProperty: true
+  replicas: 1
+  input:
+    topics:
+    - input-kafka-topic
+    typeClassName: string
+  output:
+    topic: output-kafka-topic
+    typeClassName: string
+  resources:
+    requests:
+      cpu: 100m
+      memory: 256Mi
+    limits:
+      cpu: 500m
+      memory: 512Mi
+  kafka:
+    bootstrapServers: sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092
+    consumerConfig:
+      auto.offset.reset: earliest
+    inputSchemaConfigs:
+      input-kafka-topic:
+        type: string
+    outputSchemaConfig:
+      type: string
+  genericRuntime:
+    functionFile: exclamation_function.py
+    functionFileLocation: function://public/default/test-py-function
+    language: python
+  pulsarPackageService:
+    pulsarConfig: generic-kafka-package-service
+  clusterName: test
+  subscriptionName: generic-kafka-function-sub
+  subscriptionPosition: earliest
+  autoAck: true
+  pod:
+    env:
+    - name: RUST_LOG
+      value: info
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: generic-kafka-package-service
+  namespace: default
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650