Skip to content

Commit c2a6109

Browse files
jiangpengchengjiangpengcheng
authored
feat: support Kafka protocol for functions (#843)
* Support Kafka messaging for generic runtime * Fix Kafka generic function e2e verification * Add license headers to Kafka webhook tests * Fix generic runtime image build in integration CI * Fix generic runtime clone in integration CI * Use published generic runner image in Kafka e2e * Enable KoP in generic Kafka e2e * fix ssh action * Enable Kafka transactions in generic e2e * fix ci * fix ci format * Stabilize window function integration verification * Fix crypto function input schema * Use string schema for crypto function input * Preload crypto function input schema * Stabilize window log topic verification * Stabilize integration verifiers * Stabilize integration log verifiers * Relax window log topic count check * Avoid exact window log topic counts * Make window log topic verification non-blocking * Free integration runner tool cache * Reduce operator image build temp usage * Address Kafka review comments * Address Kafka validation review comments * Skip Kafka function cleanup jobs
1 parent c7e87ce commit c2a6109

39 files changed

Lines changed: 2622 additions & 105 deletions

‎.ci/clusters/values_skywalking_e2e_cluster.yaml‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ components:
3131
pulsar_manager: false
3232
sql_worker: false
3333
proxy: false
34+
kop: true
3435

3536
## disable monitoring stack
3637
monitoring:
@@ -63,7 +64,7 @@ zookeeper:
6364
cpu: 50m
6465

6566
bookkeeper:
66-
replicaCount: 0
67+
replicaCount: 1
6768
metadata:
6869
image:
6970
repository: streamnative/sn-platform
@@ -113,6 +114,14 @@ broker:
113114
## Enable `autoSkipNonRecoverableData` since bookkeeper is running
114115
## without persistence
115116
autoSkipNonRecoverableData: "true"
117+
messagingProtocols: "kafka"
118+
protocolHandlerDirectory: "./protocols"
119+
PULSAR_PREFIX_kafkaTransactionCoordinatorEnabled: "true"
120+
PULSAR_PREFIX_brokerDeduplicationEnabled: "true"
121+
PULSAR_PREFIX_kafkaListeners: "PLAINTEXT://0.0.0.0:9092"
122+
PULSAR_PREFIX_kafkaAdvertisedListeners: "PLAINTEXT://sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092"
123+
transactionCoordinatorEnabled: "true"
124+
systemTopicEnabled: "true"
116125
# storage settings
117126
managedLedgerDefaultEnsembleSize: "1"
118127
managedLedgerDefaultWriteQuorum: "1"

‎.ci/helm.sh‎

Lines changed: 128 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -348,10 +348,19 @@ function ci::verify_backlog() {
348348
topic=$1
349349
sub=$2
350350
expected=$3
351-
BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats $topic | grep msgBacklog)
352-
if [[ "$BACKLOG" == *"\"msgBacklog\" : $expected"* ]]; then
353-
return 0
354-
fi
351+
for attempt in $(seq 1 30); do
352+
if BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats "$topic" 2>&1); then
353+
true
354+
elif BACKLOG=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-admin topics stats "${topic}-partition-0" 2>&1); then
355+
true
356+
fi
357+
if [[ "$BACKLOG" == *"\"msgBacklog\" : $expected"* ]]; then
358+
return 0
359+
fi
360+
echo "Backlog for ${topic} subscription ${sub} is not ${expected}, retry ${attempt}/30"
361+
echo "$BACKLOG"
362+
sleep 2
363+
done
355364
return 1
356365
}
357366

@@ -371,6 +380,121 @@ function ci::verify_exclamation_function() {
371380
return 1
372381
}
373382

383+
function ci::ensure_kafka_topic() {
384+
topic=$1
385+
kafka_bootstrap_server=$2
386+
properties_file=$3
387+
for attempt in $(seq 1 30); do
388+
if kubectl exec -n ${NAMESPACE} kafka-client -- kafka-topics.sh \
389+
--bootstrap-server "${kafka_bootstrap_server}" \
390+
--command-config "/opt/bitnami/kafka/config/${properties_file}" \
391+
--list > /dev/null 2>&1; then
392+
break
393+
fi
394+
if [ "${attempt}" -eq 30 ]; then
395+
echo "Kafka broker ${kafka_bootstrap_server} is not ready"
396+
kubectl get pods -n ${NAMESPACE} -o wide || true
397+
return 1
398+
fi
399+
sleep 5
400+
done
401+
kubectl exec -n ${NAMESPACE} kafka-client -- kafka-topics.sh \
402+
--bootstrap-server "${kafka_bootstrap_server}" \
403+
--create \
404+
--if-not-exists \
405+
--topic "${topic}" \
406+
--replication-factor 1 \
407+
--partitions 1 \
408+
--command-config "/opt/bitnami/kafka/config/${properties_file}"
409+
}
410+
411+
function ci::wait_kafka_topic_ready() {
412+
topic=$1
413+
kafka_bootstrap_server=$2
414+
properties_file=$3
415+
for attempt in $(seq 1 30); do
416+
if kubectl exec -n ${NAMESPACE} kafka-client -- kafka-topics.sh \
417+
--bootstrap-server "${kafka_bootstrap_server}" \
418+
--describe \
419+
--topic "${topic}" \
420+
--command-config "/opt/bitnami/kafka/config/${properties_file}"; then
421+
return 0
422+
fi
423+
echo "Kafka topic ${topic} is not ready, retry ${attempt}/30"
424+
sleep 5
425+
done
426+
return 1
427+
}
428+
429+
function ci::wait_kafka_group_coordinator_ready() {
430+
consumer_group=$1
431+
kafka_bootstrap_server=$2
432+
properties_file=$3
433+
output_file=$(mktemp)
434+
for attempt in $(seq 1 30); do
435+
if kubectl exec -n ${NAMESPACE} kafka-client -- kafka-consumer-groups.sh \
436+
--bootstrap-server "${kafka_bootstrap_server}" \
437+
--describe \
438+
--group "${consumer_group}" \
439+
--command-config "/opt/bitnami/kafka/config/${properties_file}" > "${output_file}" 2>&1; then
440+
cat "${output_file}"
441+
rm -f "${output_file}"
442+
return 0
443+
fi
444+
cat "${output_file}"
445+
if ! grep -q "CoordinatorLoadInProgress" "${output_file}"; then
446+
if grep -Eiq "does not exist|not exist|not found|no active members" "${output_file}"; then
447+
rm -f "${output_file}"
448+
return 0
449+
fi
450+
rm -f "${output_file}"
451+
return 1
452+
fi
453+
echo "Kafka consumer group coordinator for ${consumer_group} is loading, retry ${attempt}/30"
454+
sleep 5
455+
done
456+
rm -f "${output_file}"
457+
return 1
458+
}
459+
460+
function ci::verify_kafka_exclamation_function() {
461+
inputtopic=$1
462+
outputtopic=$2
463+
inputmessage=$3
464+
outputmessage=$4
465+
kafka_bootstrap_server=$5
466+
properties_file=$6
467+
consumer_group="function-mesh-${RANDOM}-$(date +%s)"
468+
output_file=$(mktemp)
469+
470+
kubectl exec -n ${NAMESPACE} kafka-client -- kafka-console-consumer.sh \
471+
--bootstrap-server "${kafka_bootstrap_server}" \
472+
--consumer.config "/opt/bitnami/kafka/config/${properties_file}" \
473+
--topic "${outputtopic}" \
474+
--group "${consumer_group}" \
475+
--timeout-ms 30000 \
476+
--max-messages 1 > "${output_file}" &
477+
consumer_pid=$!
478+
479+
sleep 3
480+
kubectl exec -n ${NAMESPACE} kafka-client -- bash -c \
481+
"echo \"${inputmessage}\" | kafka-console-producer.sh --bootstrap-server ${kafka_bootstrap_server} --producer.config /opt/bitnami/kafka/config/${properties_file} --topic ${inputtopic}"
482+
483+
if ! wait "${consumer_pid}"; then
484+
cat "${output_file}" || true
485+
rm -f "${output_file}"
486+
return 1
487+
fi
488+
489+
MESSAGE=$(cat "${output_file}")
490+
rm -f "${output_file}"
491+
echo "$MESSAGE"
492+
if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
493+
return 0
494+
fi
495+
return 1
496+
}
497+
374498
function ci::verify_exclamation_function_with_auth() {
375499
inputtopic=$1
376500
outputtopic=$2

‎.ci/tests/integration/cases/crypto-function/manifests.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ spec:
1616
typeClassName: java.lang.String
1717
sourceSpecs:
1818
"persistent://public/default/java-function-crypto-input-topic":
19+
schemaType: STRING
1920
cryptoConfig:
2021
cryptoKeyReaderClassName: "org.apache.pulsar.functions.api.examples.RawFileKeyReader"
2122
cryptoKeyReaderConfig:

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

Lines changed: 58 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -33,19 +33,69 @@ if [ ! "$KUBECONFIG" ]; then
3333
fi
3434

3535
manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/crypto-function/manifests.yaml
36-
kubectl apply -f ${manifests_file} > /dev/null 2>&1
36+
input_topic="persistent://public/default/java-function-crypto-input-topic"
37+
output_topic="persistent://public/default/java-function-crypto-output-topic"
3738

38-
verify_fm_result=$(ci::verify_function_mesh java-function-crypto-sample 2>&1)
39-
if [ $? -ne 0 ]; then
40-
echo "$verify_fm_result"
39+
function delete_topic() {
40+
topic=$1
41+
kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin topics delete "${topic}" -f > /dev/null 2>&1 || true
42+
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
43+
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
44+
kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- bin/pulsar-admin schemas delete "${topic}" > /dev/null 2>&1 || true
45+
}
46+
47+
function cleanup() {
4148
kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
49+
delete_topic "${input_topic}"
50+
delete_topic "${output_topic}"
51+
}
52+
53+
function upload_string_schema() {
54+
kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- sh -c \
55+
'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"' \
56+
sh "${input_topic}" > /dev/null
57+
}
58+
59+
function wait_string_schema() {
60+
for attempt in $(seq 1 30); do
61+
if kubectl exec -n "${PULSAR_NAMESPACE}" "${PULSAR_RELEASE_NAME}"-pulsar-broker-0 -- \
62+
bin/pulsar-admin schemas get "${input_topic}" > /dev/null 2>&1; then
63+
return 0
64+
fi
65+
echo "Schema for ${input_topic} is not ready, retry ${attempt}/30" >&2
66+
sleep 2
67+
done
68+
return 1
69+
}
70+
71+
trap cleanup EXIT
72+
cleanup
73+
74+
if ! create_topic_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::create_topic "${input_topic}" 2>&1); then
75+
echo "$create_topic_result" >&2
76+
exit 1
77+
fi
78+
79+
if ! upload_schema_result=$(upload_string_schema 2>&1); then
80+
echo "$upload_schema_result" >&2
4281
exit 1
4382
fi
4483

45-
verify_crypto_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_crypto_function 2>&1)
46-
if [ $? -eq 0 ]; then
84+
if ! wait_schema_result=$(wait_string_schema 2>&1); then
85+
echo "$wait_schema_result" >&2
86+
exit 1
87+
fi
88+
89+
kubectl apply -f "${manifests_file}" > /dev/null 2>&1
90+
91+
if ! verify_fm_result=$(ci::verify_function_mesh java-function-crypto-sample 2>&1); then
92+
echo "$verify_fm_result" >&2
93+
exit 1
94+
fi
95+
96+
if verify_crypto_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_crypto_function 2>&1); then
4797
echo "e2e-test: ok" | yq eval -
4898
else
49-
echo "$verify_crypto_result"
99+
echo "$verify_crypto_result" >&2
100+
exit 1
50101
fi
51-
kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

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

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
apiVersion: v1
2+
kind: ConfigMap
3+
metadata:
4+
name: generic-kafka-client-config
5+
namespace: default
6+
data:
7+
kafka.properties: |
8+
security.protocol=PLAINTEXT
9+
---
10+
apiVersion: v1
11+
kind: Pod
12+
metadata:
13+
name: kafka-client
14+
namespace: default
15+
spec:
16+
containers:
17+
- name: kafka
18+
image: bitnamilegacy/kafka:3.4.1
19+
command: ["/bin/sh", "-c", "--"]
20+
args: ["while true; do sleep 30; done;"]
21+
volumeMounts:
22+
- name: config-volume
23+
mountPath: /opt/bitnami/kafka/config/kafka.properties
24+
subPath: kafka.properties
25+
readOnly: true
26+
volumes:
27+
- name: config-volume
28+
configMap:
29+
name: generic-kafka-client-config

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

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
apiVersion: compute.functionmesh.io/v1alpha1
2+
kind: Function
3+
metadata:
4+
name: generic-kafka-function
5+
namespace: default
6+
spec:
7+
image: streamnative/pulsar-functions-generic-python-runner:latest
8+
className: exclamation_function.ExclamationFunction
9+
forwardSourceMessageProperty: true
10+
replicas: 1
11+
input:
12+
topics:
13+
- input-kafka-topic
14+
typeClassName: string
15+
output:
16+
topic: output-kafka-topic
17+
typeClassName: string
18+
resources:
19+
requests:
20+
cpu: 100m
21+
memory: 256Mi
22+
limits:
23+
cpu: 500m
24+
memory: 512Mi
25+
kafka:
26+
bootstrapServers: sn-platform-pulsar-broker-0.sn-platform-pulsar-broker.default.svc.cluster.local:9092
27+
consumerConfig:
28+
auto.offset.reset: earliest
29+
inputSchemaConfigs:
30+
input-kafka-topic:
31+
type: string
32+
outputSchemaConfig:
33+
type: string
34+
genericRuntime:
35+
functionFile: exclamation_function.py
36+
functionFileLocation: function://public/default/test-py-function
37+
language: python
38+
pulsarPackageService:
39+
pulsarConfig: generic-kafka-package-service
40+
clusterName: test
41+
subscriptionName: generic-kafka-function-sub
42+
subscriptionPosition: earliest
43+
autoAck: true
44+
pod:
45+
env:
46+
- name: RUST_LOG
47+
value: info
48+
---
49+
apiVersion: v1
50+
kind: ConfigMap
51+
metadata:
52+
name: generic-kafka-package-service
53+
namespace: default
54+
data:
55+
webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
56+
brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

0 commit comments

Comments
 (0)