From ef7b54f323efc296b8930608d774e8a1ea46ad92 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Tue, 27 Oct 2026 12:47:40 +0300 Subject: [PATCH 1/6] test: add Python Kafka sidecar delivery E2E The Kafka runtime sidecar is language-agnostic by design, but only Go was validated end to end. Add an integration test that proves Python: a plain cloudevents Python function, built and deployed through func on the keda deployer, receives Kafka records via the injected sidecar as CloudEvents over HTTP -- the function itself contains no Kafka code. Unlike the existing TestInt_KafkaScaling, which pins a prebuilt image and only checks the created KEDA objects, this exercises the real build+deploy pipeline and asserts delivery: it produces a record to the broker, waits for the function to log receipt, and confirms the consumer group drains to lag 0 (the sidecar commits the offset only on a 2xx). The test runs against the same Strimzi broker the Kafka integration job already provisions and is picked up by the existing test-integration-kafka target, so no CI changes are needed. --- pkg/keda/kafka_delivery_int_test.go | 384 ++++++++++++++++++++++++++++ 1 file changed, 384 insertions(+) create mode 100644 pkg/keda/kafka_delivery_int_test.go diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go new file mode 100644 index 0000000000..877b980538 --- /dev/null +++ b/pkg/keda/kafka_delivery_int_test.go @@ -0,0 +1,384 @@ +//go:build integration_kafka + +package keda_test + +import ( + "bytes" + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/rand" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + + fn "knative.dev/func/pkg/functions" + "knative.dev/func/pkg/k8s" + testingk8s "knative.dev/func/pkg/testing/k8s" +) + +// TestInt_KafkaPythonDelivery proves the language-agnostic claim for Python: a +// plain Python cloudevents function, built and deployed through func on the keda +// deployer, receives Kafka records via the injected kafka-runtime sidecar as +// CloudEvents over HTTP -- the function itself contains no Kafka code. +// +// Unlike TestInt_KafkaScaling (which pins a prebuilt image and only asserts the +// KEDA objects), this test exercises the real build+deploy pipeline for Python +// and asserts end-to-end delivery: +// 1. func create -l python -t cloudevents, add run.kafka + a kafka KEDA trigger +// 2. func deploy (builds the Python image, pushes to the in-cluster registry, +// injects the sidecar) against the real Strimzi broker +// 3. produce one record to test-topic +// 4. assert the Python user-container logged receipt ("Request Received") +// 5. assert the consumer group drains to lag 0 (the sidecar commits the offset +// only on a 2xx from the function, so lag 0 confirms delivery succeeded) +// +// Environment: the Strimzi broker provisioned by hack/cluster.sh with +// FUNC_CLUSTER_KAFKA=true (TLS + SCRAM-SHA-512 on 9093, topic test-topic, user +// test-kafka-user) plus KEDA and the in-cluster registry mirror -- i.e. the slim +// cluster the Kafka integration-tests CI job already stands up. The CA cert and +// SASL password are copied from the operator-managed secrets into the per-test +// namespace so the sidecar can authenticate, exactly as TestInt_KafkaScaling +// does. +// +// Redelivery-on-failure is intentionally out of scope here (it needs the +// function to fail then recover deterministically); this test covers the +// delivery + lag-drain half of the acceptance criteria. +func TestInt_KafkaPythonDelivery(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Minute*25) + t.Cleanup(cancel) + + name := "func-int-kafka-py-" + rand.String(5) + group := name + "-group" + ns := testingk8s.Namespace(t, ctx) + + cliSet, err := k8s.NewKubernetesClientset() + if err != nil { + t.Fatal(err) + } + + // Copy the broker CA cert and the user's SASL password from the operator- + // managed secrets in the kafka namespace into the per-test namespace so the + // injected sidecar can authenticate against the real broker (SCRAM over TLS). + caCert := readClusterSecretKey(t, ctx, cliSet, kafkaNamespace, kafkaCASecretName, "ca.crt") + caSecretName := name + "-ca" + createSecretForTest(t, ctx, cliSet, ns, caSecretName, map[string][]byte{"ca.crt": caCert}) + + saslPassword := readClusterSecretKey(t, ctx, cliSet, kafkaNamespace, kafkaUserSecretName, "password") + userSecretName := name + "-user" + createSecretForTest(t, ctx, cliSet, ns, userSecretName, map[string][]byte{"password": saslPassword}) + + // Build the func CLI from the repo under test, then scaffold a Python + // cloudevents function with it. + funcBin := buildFuncCLI(t) + funcDir := t.TempDir() + runFunc(t, ctx, funcBin, funcDir, nil, "create", "-l", "python", "-t", "cloudevents", funcDir) + + // Configure the generated function: deploy via keda with the sidecar, invoke + // as cloudevent, keep one replica up so the consumer stays joined, and point + // run.kafka at the real broker over SASL_SSL. + f, err := fn.NewFunction(funcDir) + if err != nil { + t.Fatalf("loading created function: %v", err) + } + f.Name = name + f.Namespace = ns + f.Deployer = "keda" + f.Invoke = "cloudevent" + f.Expose = "none" + minScale := int64(1) + maxScale := int64(1) + lagThreshold := int64(5) + f.Scale = &fn.ScaleOptions{ + Min: &minScale, + Max: &maxScale, + KEDA: &fn.KEDAScaleOptions{ + Triggers: []fn.KEDATrigger{ + {Type: "kafka", LagThreshold: &lagThreshold}, + }, + }, + } + f.Run.Kafka = &fn.KafkaConfig{ + Brokers: kafkaBootstrap, + Topic: "test-topic", + ConsumerGroup: group, + SecurityProtocol: "SASL_SSL", + TLS: &fn.KafkaTLS{ + CACert: "/etc/kafka/ca/ca.crt", + }, + SASL: &fn.KafkaSASL{ + Mechanism: "SCRAM-SHA-512", + User: "test-kafka-user", + Password: "{{ secret:" + userSecretName + ":password }}", + }, + } + f.Run.Volumes = []fn.Volume{ + {Secret: ptrTo(caSecretName), Path: ptrTo("/etc/kafka/ca")}, + } + if err := f.Write(); err != nil { + t.Fatalf("writing function config: %v", err) + } + + // Build + push + deploy. Use the in-cluster registry mirror (insecure), and + // honor a FUNC_KAFKA_RUNTIME_IMAGE override if the caller set one (otherwise + // the deployer's built-in default sidecar image is used). + deployEnv := []string{ + "FUNC_REGISTRY=registry.localtest.me/func", + "FUNC_REGISTRY_INSECURE=true", + } + if img := os.Getenv("FUNC_KAFKA_RUNTIME_IMAGE"); img != "" { + deployEnv = append(deployEnv, "FUNC_KAFKA_RUNTIME_IMAGE="+img) + } + runFunc(t, ctx, funcBin, funcDir, deployEnv, "deploy", "--namespace", ns, "--verbose") + t.Cleanup(func() { + // Best effort; the per-test namespace cleanup (testingk8s.Namespace) is + // the real backstop. + _ = runFuncErr(context.Background(), funcBin, funcDir, deployEnv, "delete", "--namespace", ns) + }) + + // A successful keda deploy only reports ready once the sidecar's readiness + // probe passes, which requires the consumer to have joined the group and been + // assigned partitions. So by now the consumer is live and -- because it reads + // from the newest offset -- a record produced now will be seen. + strimziImg := strimziKafkaImage(t, ctx, cliSet) + produceRecord(t, ctx, cliSet, strimziImg, "hello from python e2e") + + // Assert the Python function received the CloudEvent. + waitForFunctionLog(t, ctx, cliSet, ns, name, "Request Received") + + // Assert the consumer group drains to lag 0 (offset committed => 2xx + // delivery confirmed). + waitForZeroLag(t, ctx, cliSet, strimziImg, group) +} + +// buildFuncCLI compiles the func binary from the repo under test and returns its +// path. +func buildFuncCLI(t *testing.T) string { + t.Helper() + _, thisFile, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot determine source file path") + } + repoRoot := filepath.Join(filepath.Dir(thisFile), "..", "..") + bin := filepath.Join(t.TempDir(), "func") + + cmd := exec.Command("go", "build", "-o", bin, "./cmd/func") + cmd.Dir = repoRoot + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("building func CLI: %v\n%s", err, out) + } + return bin +} + +// runFunc runs the func CLI and fails the test on error. extraEnv entries +// (KEY=VALUE) are appended to the inherited environment. +func runFunc(t *testing.T, ctx context.Context, bin, dir string, extraEnv []string, args ...string) { + t.Helper() + if err := runFuncErr(ctx, bin, dir, extraEnv, args...); err != nil { + t.Fatalf("func %s failed: %v", strings.Join(args, " "), err) + } +} + +func runFuncErr(ctx context.Context, bin, dir string, extraEnv []string, args ...string) error { + cmd := exec.CommandContext(ctx, bin, args...) + cmd.Dir = dir + cmd.Env = append(os.Environ(), extraEnv...) + var buf bytes.Buffer + cmd.Stdout = &buf + cmd.Stderr = &buf + if err := cmd.Run(); err != nil { + return fmt.Errorf("%w\n%s", err, buf.String()) + } + return nil +} + +// strimziKafkaImage returns the image the running broker uses, so the client +// tool pods below run the exact same Kafka version as the broker. +func strimziKafkaImage(t *testing.T, ctx context.Context, cliSet *kubernetes.Clientset) string { + t.Helper() + pods, err := cliSet.CoreV1().Pods(kafkaNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: "strimzi.io/cluster=" + kafkaClusterName + ",strimzi.io/broker-role=true", + }) + if err != nil || len(pods.Items) == 0 { + // Fall back to the generic kafka-pool label if the role label is absent. + pods, err = cliSet.CoreV1().Pods(kafkaNamespace).List(ctx, metav1.ListOptions{ + LabelSelector: "strimzi.io/cluster=" + kafkaClusterName + ",strimzi.io/kind=Kafka", + }) + } + if err != nil { + t.Fatalf("listing broker pods: %v", err) + } + if len(pods.Items) == 0 { + t.Fatal("no Strimzi broker pod found (is Kafka installed? FUNC_CLUSTER_KAFKA=true)") + } + for _, c := range pods.Items[0].Spec.Containers { + if c.Name == "kafka" { + return c.Image + } + } + // Single-container broker pod: use the first container's image. + return pods.Items[0].Spec.Containers[0].Image +} + +// clientPropsScript emits the shell that writes an SASL_SSL client.properties to +// /tmp/client.properties, authenticating as test-kafka-user with the password in +// $PASSWORD and trusting the broker CA mounted (PEM) at /opt/ca/ca.crt. +const clientPropsScript = `cat > /tmp/client.properties <", ns, pod, err) + } + return string(b) +} From 0b628e063306c2178b435487b04e40503f0a0cf7 Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Tue, 27 Oct 2026 12:58:29 +0300 Subject: [PATCH 2/6] test: verify Python receives exact Kafka payload The delivery check only asserted the function logged "Request Received", which proves a request reached the function but not that it was our record with the payload intact. Produce a record carrying a unique per-run token and instrument the scaffolded handler to echo the received data, then assert that token appears in the function logs. Delivery of the specific record (delivered and decoded intact) is now verified, alongside the existing lag-0 check that independently confirms the function acked it. --- pkg/keda/kafka_delivery_int_test.go | 53 +++++++++++++++++++++++++---- 1 file changed, 47 insertions(+), 6 deletions(-) diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go index 877b980538..1a77cf3d18 100644 --- a/pkg/keda/kafka_delivery_int_test.go +++ b/pkg/keda/kafka_delivery_int_test.go @@ -36,10 +36,14 @@ import ( // 1. func create -l python -t cloudevents, add run.kafka + a kafka KEDA trigger // 2. func deploy (builds the Python image, pushes to the in-cluster registry, // injects the sidecar) against the real Strimzi broker -// 3. produce one record to test-topic -// 4. assert the Python user-container logged receipt ("Request Received") +// 3. produce one record carrying a unique token to test-topic +// 4. assert the Python user-container logged that exact token (the handler is +// instrumented to echo the received payload, so this confirms our specific +// record was delivered and decoded intact -- not merely that a request +// arrived) // 5. assert the consumer group drains to lag 0 (the sidecar commits the offset -// only on a 2xx from the function, so lag 0 confirms delivery succeeded) +// only on a 2xx from the function, so lag 0 independently confirms the +// function acked the delivery) // // Environment: the Strimzi broker provisioned by hack/cluster.sh with // FUNC_CLUSTER_KAFKA=true (TLS + SCRAM-SHA-512 on 9093, topic test-topic, user @@ -58,6 +62,11 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { name := "func-int-kafka-py-" + rand.String(5) group := name + "-group" + // A unique per-run token carried in the record payload. The instrumented + // handler echoes the received data, so finding this token in the function's + // logs proves it received *our* specific record (payload intact), not merely + // that some request arrived. + token := "e2e-" + rand.String(10) ns := testingk8s.Namespace(t, ctx) cliSet, err := k8s.NewKubernetesClientset() @@ -82,6 +91,10 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { funcDir := t.TempDir() runFunc(t, ctx, funcBin, funcDir, nil, "create", "-l", "python", "-t", "cloudevents", funcDir) + // Echo the received payload from the handler so delivery of our specific + // record can be verified (see token above). + instrumentPythonHandler(t, funcDir) + // Configure the generated function: deploy via keda with the sidecar, invoke // as cloudevent, keep one replica up so the consumer stays joined, and point // run.kafka at the real broker over SASL_SSL. @@ -149,16 +162,44 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { // assigned partitions. So by now the consumer is live and -- because it reads // from the newest offset -- a record produced now will be seen. strimziImg := strimziKafkaImage(t, ctx, cliSet) - produceRecord(t, ctx, cliSet, strimziImg, "hello from python e2e") + produceRecord(t, ctx, cliSet, strimziImg, "payload-"+token) - // Assert the Python function received the CloudEvent. - waitForFunctionLog(t, ctx, cliSet, ns, name, "Request Received") + // Assert the Python function received our specific record: the instrumented + // handler echoes the payload, so the unique token appears in its logs only if + // the record was delivered and decoded intact. + waitForFunctionLog(t, ctx, cliSet, ns, name, token) // Assert the consumer group drains to lag 0 (offset committed => 2xx // delivery confirmed). waitForZeroLag(t, ctx, cliSet, strimziImg, group) } +// instrumentPythonHandler edits the scaffolded Python handler to log the +// received CloudEvent's data. This is test-only instrumentation of our own +// scaffold (not the shipped template): it lets the test assert that the function +// received the exact payload we produced, rather than only that a request +// arrived. +func instrumentPythonHandler(t *testing.T, funcDir string) { + t.Helper() + p := filepath.Join(funcDir, "function", "func.py") + b, err := os.ReadFile(p) + if err != nil { + t.Fatalf("reading scaffolded handler: %v", err) + } + // Anchor on the line that binds the decoded event; append an echo right after + // it so `request` is in scope. + const anchor = `request = scope["event"]` + src := string(b) + if !strings.Contains(src, anchor) { + t.Fatalf("scaffolded handler missing %q; template changed, update this test", anchor) + } + echo := anchor + "\n logging.info(f\"FUNC-E2E-RECEIVED data={request.get_data()!r}\")" + src = strings.Replace(src, anchor, echo, 1) + if err := os.WriteFile(p, []byte(src), 0o644); err != nil { + t.Fatalf("writing instrumented handler: %v", err) + } +} + // buildFuncCLI compiles the func binary from the repo under test and returns its // path. func buildFuncCLI(t *testing.T) string { From 7fda8c902a4582e0098be39267f205482fd4d79b Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Tue, 27 Oct 2026 13:12:57 +0300 Subject: [PATCH 3/6] test: use committed fixture for Kafka payload echo The Python Kafka delivery test instrumented the scaffolded handler in place with a strings.Replace on func.py, anchored to a template line. That coupled the test to the template's exact source and hid the only part that matters -- the echo handler -- inside Go string surgery. Replace it with a committed handler fixture under testdata that is copied over the scaffold after func create. func create still provides func.yaml and the dependencies; only the handler is swapped. The fixture is a reviewable Python file and the test no longer breaks when the template's wording changes. --- pkg/keda/kafka_delivery_int_test.go | 59 +++++++++---------- .../testdata/kafka_python_delivery/func.py | 35 +++++++++++ 2 files changed, 63 insertions(+), 31 deletions(-) create mode 100644 pkg/keda/testdata/kafka_python_delivery/func.py diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go index 1a77cf3d18..b6bcb81882 100644 --- a/pkg/keda/kafka_delivery_int_test.go +++ b/pkg/keda/kafka_delivery_int_test.go @@ -37,10 +37,9 @@ import ( // 2. func deploy (builds the Python image, pushes to the in-cluster registry, // injects the sidecar) against the real Strimzi broker // 3. produce one record carrying a unique token to test-topic -// 4. assert the Python user-container logged that exact token (the handler is -// instrumented to echo the received payload, so this confirms our specific -// record was delivered and decoded intact -- not merely that a request -// arrived) +// 4. assert the Python user-container logged that exact token (a test handler +// fixture echoes the received payload, so this confirms our specific record +// was delivered and decoded intact -- not merely that a request arrived) // 5. assert the consumer group drains to lag 0 (the sidecar commits the offset // only on a 2xx from the function, so lag 0 independently confirms the // function acked the delivery) @@ -62,10 +61,10 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { name := "func-int-kafka-py-" + rand.String(5) group := name + "-group" - // A unique per-run token carried in the record payload. The instrumented - // handler echoes the received data, so finding this token in the function's - // logs proves it received *our* specific record (payload intact), not merely - // that some request arrived. + // A unique per-run token carried in the record payload. The handler fixture + // echoes the received data, so finding this token in the function's logs + // proves it received *our* specific record (payload intact), not merely that + // some request arrived. token := "e2e-" + rand.String(10) ns := testingk8s.Namespace(t, ctx) @@ -91,9 +90,10 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { funcDir := t.TempDir() runFunc(t, ctx, funcBin, funcDir, nil, "create", "-l", "python", "-t", "cloudevents", funcDir) - // Echo the received payload from the handler so delivery of our specific - // record can be verified (see token above). - instrumentPythonHandler(t, funcDir) + // Replace the scaffolded handler with a test fixture that echoes the received + // payload, so delivery of our specific record can be verified (see token + // above). The stock template only logs "Request Received" without the payload. + installHandlerFixture(t, funcDir) // Configure the generated function: deploy via keda with the sidecar, invoke // as cloudevent, keep one replica up so the consumer stays joined, and point @@ -174,29 +174,26 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { waitForZeroLag(t, ctx, cliSet, strimziImg, group) } -// instrumentPythonHandler edits the scaffolded Python handler to log the -// received CloudEvent's data. This is test-only instrumentation of our own -// scaffold (not the shipped template): it lets the test assert that the function -// received the exact payload we produced, rather than only that a request -// arrived. -func instrumentPythonHandler(t *testing.T, funcDir string) { +// installHandlerFixture overwrites the scaffolded Python handler with the +// committed test fixture (testdata/kafka_python_delivery/func.py), which echoes +// the received CloudEvent's data to the log. This lets the test assert the +// function received the exact payload we produced, rather than only that a +// request arrived. The fixture is a reviewable file rather than inline source +// surgery; func create still scaffolds everything else (func.yaml, deps). +func installHandlerFixture(t *testing.T, funcDir string) { t.Helper() - p := filepath.Join(funcDir, "function", "func.py") - b, err := os.ReadFile(p) - if err != nil { - t.Fatalf("reading scaffolded handler: %v", err) + _, thisFile, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot determine source file path") } - // Anchor on the line that binds the decoded event; append an echo right after - // it so `request` is in scope. - const anchor = `request = scope["event"]` - src := string(b) - if !strings.Contains(src, anchor) { - t.Fatalf("scaffolded handler missing %q; template changed, update this test", anchor) + src := filepath.Join(filepath.Dir(thisFile), "testdata", "kafka_python_delivery", "func.py") + b, err := os.ReadFile(src) + if err != nil { + t.Fatalf("reading handler fixture: %v", err) } - echo := anchor + "\n logging.info(f\"FUNC-E2E-RECEIVED data={request.get_data()!r}\")" - src = strings.Replace(src, anchor, echo, 1) - if err := os.WriteFile(p, []byte(src), 0o644); err != nil { - t.Fatalf("writing instrumented handler: %v", err) + dst := filepath.Join(funcDir, "function", "func.py") + if err := os.WriteFile(dst, b, 0o644); err != nil { + t.Fatalf("installing handler fixture into scaffold: %v", err) } } diff --git a/pkg/keda/testdata/kafka_python_delivery/func.py b/pkg/keda/testdata/kafka_python_delivery/func.py new file mode 100644 index 0000000000..74990ecfae --- /dev/null +++ b/pkg/keda/testdata/kafka_python_delivery/func.py @@ -0,0 +1,35 @@ +# Test handler fixture for TestInt_KafkaPythonDelivery. +# +# A plain CloudEvents-over-HTTP Python function (no Kafka code) that echoes the +# received payload to its log, so the integration test can confirm the exact +# record produced to Kafka was delivered to this function -- decoded intact -- +# via the injected kafka-runtime sidecar. It returns an OK response CloudEvent +# (a 2xx) so the sidecar commits the offset and the consumer group drains. +# +# This is deliberately a copy-over-the-scaffold fixture rather than the shipped +# template: the stock template only logs "Request Received" with no payload, so +# verifying specific content requires a handler that echoes it. +import logging +from cloudevents.core.v1.event import CloudEvent + + +def new(): + return Function() + + +class Function: + async def handle(self, scope, receive, send): + event = scope["event"] + # The test produces a record whose value carries a unique token and + # asserts this line (with that token) appears in the function's logs. + logging.info(f"FUNC-E2E-RECEIVED data={event.get_data()!r}") + + response = CloudEvent( + attributes={ + "type": "function.response", + "source": "function", + "id": f"response-{event.get_id() or 'unknown'}", + }, + data={"message": "OK"}, + ) + await send(response) From 7648c57971f51e7f7559cdd5f77c9b00e207505c Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Tue, 27 Oct 2026 14:34:25 +0300 Subject: [PATCH 4/6] test: fix function name derived from temp dir func create derives the function name from the target directory's basename. t.TempDir() returns a path whose leaf segment is numeric (e.g. ".../002"), which fails function-name validation (a name must start with an alphabetic character). Nest the function in a validly named subdirectory so create succeeds. --- pkg/keda/kafka_delivery_int_test.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go index b6bcb81882..9daf9581ce 100644 --- a/pkg/keda/kafka_delivery_int_test.go +++ b/pkg/keda/kafka_delivery_int_test.go @@ -87,7 +87,13 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { // Build the func CLI from the repo under test, then scaffold a Python // cloudevents function with it. funcBin := buildFuncCLI(t) - funcDir := t.TempDir() + // func create derives the function name from the directory's basename, and + // t.TempDir()'s leaf is numeric (e.g. "002"), which is not a valid function + // name. Nest the function in a validly-named subdirectory. + funcDir := filepath.Join(t.TempDir(), name) + if err := os.MkdirAll(funcDir, 0o755); err != nil { + t.Fatalf("creating function dir: %v", err) + } runFunc(t, ctx, funcBin, funcDir, nil, "create", "-l", "python", "-t", "cloudevents", funcDir) // Replace the scaffolded handler with a test fixture that echoes the received From 77a2f588c3f2211e537be21b4070c14d5921beff Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Tue, 27 Oct 2026 22:34:17 +0300 Subject: [PATCH 5/6] fix: reach the in-cluster registry on the slim Kafka cluster The Kafka integration job runs a slim cluster with serving (and thus the Contour ingress) disabled to save CPU. Pushing a built function image to registry.localtest.me only works through that ingress, so the Python delivery test's build+push failed with a connection reset -- the first Kafka test to actually build and push, since the scaling test pins a prebuilt image. Expose the registry pod's hostPort (5000) to the host and push to localhost:5000 instead. containerd treats localhost as insecure, and the node already pulls it through the same hostPort the containerd mirror uses, so no ingress is needed. This is the canonical kind-local-registry pattern. --- hack/cluster.sh | 10 ++++++++++ pkg/keda/kafka_delivery_int_test.go | 9 ++++++--- 2 files changed, 16 insertions(+), 3 deletions(-) diff --git a/hack/cluster.sh b/hack/cluster.sh index b138af6490..7b9261ca2b 100755 --- a/hack/cluster.sh +++ b/hack/cluster.sh @@ -157,6 +157,13 @@ nodes: - containerPort: 30022 hostPort: 30022 listenAddress: "127.0.0.1" + # Expose the in-cluster registry's hostPort (5000) to the host. Pushing to + # registry.localtest.me normally goes through the Contour ingress (installed + # with serving); on slim clusters that skip serving (e.g. the Kafka job), the + # ingress is absent, so a host-side build+push needs localhost:5000 directly. + - containerPort: 5000 + hostPort: 5000 + listenAddress: "127.0.0.1" - containerPort: 80 hostPort: 80 listenAddress: "::1" @@ -166,6 +173,9 @@ nodes: - containerPort: 30022 hostPort: 30022 listenAddress: "::1" + - containerPort: 5000 + hostPort: 5000 + listenAddress: "::1" containerdConfigPatches: - |- [plugins."io.containerd.grpc.v1.cri".registry.mirrors."registry.localtest.me"] diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go index 9daf9581ce..75e6cf466c 100644 --- a/pkg/keda/kafka_delivery_int_test.go +++ b/pkg/keda/kafka_delivery_int_test.go @@ -146,11 +146,14 @@ func TestInt_KafkaPythonDelivery(t *testing.T) { t.Fatalf("writing function config: %v", err) } - // Build + push + deploy. Use the in-cluster registry mirror (insecure), and - // honor a FUNC_KAFKA_RUNTIME_IMAGE override if the caller set one (otherwise + // Build + push + deploy. Push to the in-cluster registry via localhost:5000 + // (its hostPort), not registry.localtest.me: the latter is only reachable + // through the Contour ingress, which the slim Kafka cluster skips (no + // serving). The node still pulls localhost:5000 through the same hostPort. + // Honor a FUNC_KAFKA_RUNTIME_IMAGE override if the caller set one (otherwise // the deployer's built-in default sidecar image is used). deployEnv := []string{ - "FUNC_REGISTRY=registry.localtest.me/func", + "FUNC_REGISTRY=localhost:5000/func", "FUNC_REGISTRY_INSECURE=true", } if img := os.Getenv("FUNC_KAFKA_RUNTIME_IMAGE"); img != "" { From a9781d8ab567bc62e8adcba48f6ad46df0247f8f Mon Sep 17 00:00:00 2001 From: Ali Ok Date: Thu, 29 Oct 2026 23:07:18 +0300 Subject: [PATCH 6/6] test: fail fast when a Kafka tool pod fails runKafkaToolPod waited for the pod to reach either Succeeded or Failed and then returned its logs regardless of which. A failed producer pod was therefore treated as success and only surfaced minutes later as a misleading "function did not log" delivery timeout; a failed consumer-group command was parsed as a lag observation and reported as a false "did not drain to lag 0". Capture the terminal phase and fail the test immediately with the pod logs when it is not Succeeded, so the real failure is reported at its source. --- pkg/keda/kafka_delivery_int_test.go | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/pkg/keda/kafka_delivery_int_test.go b/pkg/keda/kafka_delivery_int_test.go index 75e6cf466c..b8110544f4 100644 --- a/pkg/keda/kafka_delivery_int_test.go +++ b/pkg/keda/kafka_delivery_int_test.go @@ -378,17 +378,26 @@ func runKafkaToolPod(t *testing.T, ctx context.Context, cliSet *kubernetes.Clien _ = cliSet.CoreV1().Pods(kafkaNamespace).Delete(context.Background(), podName, metav1.DeleteOptions{}) }() + var phase corev1.PodPhase err := wait.PollUntilContextTimeout(ctx, time.Second*2, time.Minute*2, true, func(ctx context.Context) (bool, error) { p, err := cliSet.CoreV1().Pods(kafkaNamespace).Get(ctx, podName, metav1.GetOptions{}) if err != nil { return false, err } - return p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed, nil + phase = p.Status.Phase + return phase == corev1.PodSucceeded || phase == corev1.PodFailed, nil }) logs := podLogs(ctx, cliSet, kafkaNamespace, podName, "") if err != nil { t.Fatalf("tool pod %s did not complete: %v\n%s", podName, err, logs) } + // Treat a failed pod as a test failure, surfacing its logs immediately. + // Otherwise a failed producer or consumer-group command would return here + // as if it succeeded, only to resurface minutes later as a misleading + // delivery timeout or a false "did not drain to lag 0". + if phase != corev1.PodSucceeded { + t.Fatalf("tool pod %s failed (phase %s):\n%s", podName, phase, logs) + } return logs }