Add KEDA Kafka consumer-lag scaling support

[aliok]

Changes

This PR adds Kafka consumer-lag autoscaling support via KEDA. When deployer: keda is set and Kafka configuration is provided, the deployer creates a KEDA ScaledObject with a Kafka trigger that scales function pods based on consumer group lag, along with a TriggerAuthentication resource for SASL/TLS credentials.

The func.yaml schema has been restructured:

• scale is now a top-level field (previously under deploy.scale) with new KEDA tuning knobs: cooldownPeriod, pollingInterval, lagThreshold, and activationLagThreshold.
• Kafka runtime configuration lives under run.kafka (brokers, topic, consumerGroup, security protocol, SASL/TLS settings).

A migration handles existing func.yaml files automatically, moving deploy.scale to the top level and deploy.kafka to run.kafka.

The PR also includes testing deployment scenarios (A-G) with collected Kubernetes resource snapshots for each deployer/Kafka combination.

/kind enhancement

Supersedes #4045

Closes #2419

Release Note

Functions can now consume from Apache Kafka using KEDA-based consumer-lag autoscaling. Configure Kafka connection details (brokers, topic, consumer group, SASL/TLS) under `run.kafka` in func.yaml and set `deployer: keda` to deploy with a ScaledObject that scales based on Kafka consumer lag. New `scale` top-level config supports KEDA tuning knobs: cooldownPeriod, pollingInterval, lagThreshold, and activationLagThreshold.

Docs

[knative-prow]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by: aliok
Once this PR has been reviewed and has the lgtm label, please assign gauron99 for approval. For more information see the Code Review Process.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[Copilot]

🟡 Changes recommended

There are correctness and UX gaps around KEDA trigger configurations (unsupported trigger types/combos) and TriggerAuthentication construction that can lead to broken Kafka scaling at runtime.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

Adds KEDA-based Kafka consumer-lag autoscaling support and restructures func.yaml autoscaling configuration so KEDA/KPA settings live under a new top-level scale field, with a migration to preserve backwards compatibility.

Changes:

• Introduces top-level scale with keda/kpa sub-keys, updates validation, schema, docs, and a migration from deploy.options.scale.
• Adds KEDA Kafka scaling implementation that creates a ScaledObject + optional TriggerAuthentication, and updates the KEDA remover to clean these up.
• Updates deployers, CLI, and tests to use Deploy.ActiveDeployer/Deploy.ActiveExpose and the new Function.Scale location.

File summaries

File	Description
schema/func_yaml-schema.json	Updates schema for scale top-level move and new KEDA/KPA structures.
pkg/pipelines/tekton/pipelines_provider.go	Renames deployed-state fields to ActiveDeployer/ActiveExpose.
pkg/mock/deployer.go	Updates mock deployer to use Deploy.ActiveDeployer.
pkg/knative/deployer.go	Reads autoscaling annotations from Function.Scale (KPA sub-key).
pkg/keda/remover.go	Adds explicit deletion of Kafka KEDA scaling resources during remove.
pkg/keda/kafka_scaling.go	Implements unstructured KEDA ScaledObject/TriggerAuthentication helpers.
pkg/keda/kafka_scaling_test.go	Unit tests for Kafka scaling object construction helpers.
pkg/keda/kafka_scaling_int_test.go	Integration test verifying KEDA resources created/removed in-cluster.
pkg/keda/deployer.go	Wires trigger selection and creates Kafka scaling resources during deploy.
pkg/k8s/wait.go	Treats Deployments with desired replicas 0 as “available” for waits.
pkg/k8s/deployer.go	Seeds Deployment replicas from Function.Scale.Min.
pkg/functions/function.go	Adds Function.Scale, renames deployed-state fields, hooks new validation.
pkg/functions/function_scale.go	New deployer-aware scale validation entry point.
pkg/functions/function_options.go	Reworks scale types into KEDA/KPA; keeps legacy options-scale for migration only.
pkg/functions/function_options_unit_test.go	Moves scale tests to ValidateScale; retains resource validation tests.
pkg/functions/function_migrations.go	Adds 0.37.0 migration to move scale to top-level + KPA flattening.
pkg/functions/function_migrations_unit_test.go	Adds unit tests covering the new scale migration scenarios.
pkg/functions/client.go	Uses Deploy.ActiveDeployer/ActiveExpose for deploy/remove state tracking.
pkg/functions/client_test.go	Updates tests for renamed deployed-state fields.
pkg/deployer/testing/integration_test_helper.go	Updates integration helper to set Function.Scale and ActiveExpose.
pkg/config/config.go	Uses Deploy.ActiveDeployer when seeding global config.
e2e/e2e_recorder_test.go	Updates E2E recorder test to set Function.Scale.
e2e/e2e_expose_test.go	Updates E2E exposure tests for trigger requirement and ActiveExpose/Deployer.
docs/reference/func_yaml.md	Documents new scale and run.kafka structures and examples.
docs/reference/func_deploy.md	Updates deployer flag help to reflect broader KEDA trigger support.
cmd/func-util/main.go	Uses Deploy.ActiveDeployer as fallback for effective deployer.
cmd/deploy.go	Updates help text and deployed-state field usage; adjusts exposure-record warning logic.
cmd/deploy_test.go	Updates deploy tests for ActiveDeployer/Expose and KEDA trigger requirement.
cmd/delete_test.go	Updates delete tests for ActiveDeployer and KEDA trigger requirement.

Review details

Suppressed comments (1)

docs/reference/func_yaml.md:218

• The KEDA example configures both an HTTP trigger and a Kafka trigger, but the current deployer implementation uses an HTTPScaledObject for HTTP and a separate ScaledObject for Kafka; those generally can’t both target the same Deployment. Update the example to a supported single-trigger configuration (or add an explicit limitation note).

      - type: http
        targetValue: 200
      - type: kafka
        lagThreshold: 5
        activationLagThreshold: 0

• Files reviewed: 29/29 changed files
• Comments generated: 5
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

🔵 Needs a closer look

There are a few correctness gaps (stale Kafka scaler resources not cleaned up on trigger removal, incomplete mutual-TLS wiring for KEDA auth, and an integration test helper still setting legacy scale config) that should be addressed before approval.

Review details

Suppressed comments (6)

Previously missed (5) — in code that hasn't changed since the last review.

pkg/deployer/testing/integration_test_helper.go:746

• This integration test helper still sets scaling under Deploy.Options.Scale, but the deployers now read scaling from the top-level Function.Scale. As written, minScale/maxScale here will no longer affect the deployed resources, so the test isn’t exercising scale bounds as intended.
  pkg/keda/deployer.go:211
• Kafka scaler resources are only created when a Kafka trigger is configured, but there’s no corresponding cleanup when the Kafka trigger is removed (or when switching back to HTTP-only). That can leave a stale ScaledObject/TriggerAuthentication behind and prevent switching triggers cleanly (or keep scaling based on lag unexpectedly).
  pkg/keda/kafka_scaling.go:92
• needsTriggerAuth doesn’t consider mutual-TLS inputs (run.kafka.tls.clientCert/clientKey). If a user configures mTLS without a CA cert and without SASL, the ScaledObject won’t reference a TriggerAuthentication at all, so the scaler can’t be given the cert/key material.

This issue also appears on line 178 of the same file.
docs/reference/func_yaml.md:150

• The documented defaults for scale.min/scale.max don’t match the current deployer implementations. For example, the raw deployer defaults to 1 replica, and the keda deployer defaults HTTP scaler bounds to min=1/max=10 when unset.
  schema/func_yaml-schema.json:263
• schema/func_yaml-schema.json is listed as a generated file in AGENTS.md and shouldn’t be edited directly. Please regenerate it via the repo’s codegen workflow (and commit the generated output) so it stays consistent with the Go struct tags and other generated artifacts.

pkg/keda/kafka_scaling.go:187

• buildTriggerAuth only wires run.kafka.tls.caCert into the TriggerAuthentication. If run.kafka.tls.clientCert/clientKey are set (mutual TLS), they’re currently ignored, so KEDA can’t be configured with the client cert/key even though the func.yaml schema/docs allow them.

	if kafka.TLS != nil && kafka.TLS.CACert != "" {
		caSecretName, caKey := findSecretForPath(kafka.TLS.CACert, f.Run.Volumes)
		if caSecretName != "" {
			secretRefs = append(secretRefs, map[string]interface{}{
				"parameter": "ca",
				"name":      caSecretName,
				"key":       caKey,
			})
		}
	}

• Files reviewed: 29/29 changed files
• Comments generated: 0 new
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Critical migration, deployment, compatibility, and integration issues remain unresolved.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (13)

Previously missed (1) — in code that hasn't changed since the last review.

pkg/keda/kafka_scaling.go:303

• run.kafka.tls.skipVerify is propagated to the function container, but the KEDA scaler connects independently and this trigger metadata never tells it to skip certificate verification. With a self-signed broker, the consumer can connect while the scaler fails TLS and never scales; when SkipVerify is true, also emit KEDA's unsafeSslSkipVerify: "true" metadata.

cmd/deploy.go:171

• This help text advertises cron as a supported KEDA trigger, but ValidateScale rejects type: cron as not implemented (pkg/functions/function_scale.go:86-93), and the reference documentation correctly says it fails validation. Users following this help cannot deploy the advertised mode; remove cron from the help text until it is implemented and regenerate the command reference.

		fmt.Sprintf("Type of deployment to use: '%s' for Knative Service, '%s' for Kubernetes Deployment, or '%s' for Deployment scaled by KEDA (HTTP, Kafka, or cron triggers) ($FUNC_DEPLOYER)", deployers.Knative, deployers.Kubernetes, deployers.Keda))

pkg/functions/function.go:346

• These renames remove the exported DeploySpec.Deployer and DeploySpec.Expose fields. Keeping the same YAML tags preserves func.yaml files, but downstream Go consumers that read or write the old fields will no longer compile. Retain a deprecated compatibility surface or explicitly version and document this breaking Go API change.

	// ActiveDeployer records the deployer the Function is CURRENTLY DEPLOYED
	// with: observed state, written after successful deployment, and cleared
	// on undeploy alongside Namespace. User intent lives on Function.Deployer.
	ActiveDeployer string `yaml:"deployer,omitempty" jsonschema:"enum=knative,enum=raw,enum=keda"`

pkg/functions/function_scale.go:68

• Validation permits multiple triggers of the same type, but kafkaTrigger and the HTTP scaler setup each use only the first matching trigger. A configuration with two Kafka (or two HTTP) entries therefore silently ignores the remaining trigger settings. Reject duplicate trigger types or materialize every configured trigger instead of accepting ineffective configuration.

	var sawHTTP, sawKafka bool
	for i, t := range keda.Triggers {

pkg/functions/function_scale.go:104

• This validation only rejects a list containing both trigger types; it still accepts run.kafka with an HTTP-only KEDA trigger. validateKafka configures the pod as a Kafka consumer, while httpScaledObject scales it from HTTP traffic, so Kafka lag can grow without causing scale-out. Require a Kafka trigger whenever run.kafka is set for KEDA, or reject this combination explicitly.

	if sawHTTP && sawKafka {
		errors = append(errors, "scale.keda.triggers must not combine type http with type kafka: they cannot scale the same Deployment together, not yet supported")
	}

pkg/keda/deployer.go:221

• These defensive Kafka validations run only after d.Deployer.Deploy has already created or updated the raw Deployment and Service. A direct caller of this deployer (which the surrounding comments explicitly support without Function.Validate) can therefore receive an error while leaving a running partial workload with no Kafka scaler; the unsupported-mechanism and unresolved-auth checks below have the same problem. Perform all Kafka/KEDA preflight validation before invoking the raw deployer, or roll back the created resources on failure.

	if wantKafka && f.Run.Kafka == nil {
		// ValidateScale already rejects this combination, but Deploy is
		// reachable without going through Function.Validate first (library
		// callers, tests): fail loudly here too instead of silently treating
		// a misconfigured kafka trigger as "no kafka trigger" and deploying
		// without any Kafka scaling or error.
		return fn.DeploymentResult{}, fmt.Errorf("function %q: scale.keda.triggers has a kafka trigger but run.kafka is not configured", f.Name)

pkg/keda/deployer.go:213

• Kafka-only deployments do not create an HTTPScaledObject, but the existing KEDA Describer and Lister unconditionally fetch one after recognizing the KEDA service. Consequently func describe, func list, and the pipeline's post-deploy status lookup fail with NotFound for the new Kafka-only mode. Provide a Kafka/ScaledObject-aware status path or make these readers handle KEDA functions without an HTTP scaler.

		// No HTTP trigger — URL is the app service. The Service listens on
		// port 80 (routing to the container's DefaultHTTPPort via
		// targetPort), so the URL, like elsewhere in the codebase (e.g.
		// pkg/k8s/describer.go), has no explicit port.
		url = fmt.Sprintf("http://%s.%s.svc", f.Name, namespace)
	}

pkg/keda/deployer.go:254

• When a previous Kafka deployment needed credentials but the new Kafka configuration no longer does, this branch skips ensureTriggerAuth and updates only the ScaledObject, leaving the old TriggerAuthentication behind. Unlike the no-Kafka branch below, no cleanup runs, so stale auth resources persist until undeploy. Delete the named TriggerAuthentication when needsTriggerAuth is false.

		}

pkg/keda/kafka_scaling.go:334

• This owner reference sets blockOwnerDeletion: true, which can make ScaledObject creation fail on OpenShift's OwnerReferencesPermissionEnforcement admission plugin when the deployer or pipeline service account cannot update the Deployment finalizers subresource. The raw Service owner reference explicitly omits this flag for that reason (pkg/k8s/deployer.go:729-737); omit it here as well so background garbage collection still works.

						"blockOwnerDeletion": true,

pkg/keda/kafka_scaling.go:248

• This owner reference sets blockOwnerDeletion: true, which can make TriggerAuthentication creation fail on OpenShift's OwnerReferencesPermissionEnforcement admission plugin when the deployer or pipeline service account cannot update the Deployment finalizers subresource. The raw Service owner reference explicitly omits this flag for that reason (pkg/k8s/deployer.go:729-737); omit it here as well so background garbage collection still works.

						"blockOwnerDeletion": true,

pkg/keda/kafka_scaling.go:390

• The update replaces the existing ScaledObject metadata with this newly built object and only preserves the resource version. KEDA adds its own finalizer to this resource (as the integration test notes at pkg/keda/kafka_scaling_int_test.go:196-205), so a redeploy strips that finalizer and can bypass KEDA cleanup when the scaler is later deleted. Preserve the existing finalizers before updating.

	so.SetResourceVersion(existing.GetResourceVersion())
	if _, err := client.Update(ctx, so, metav1.UpdateOptions{}); err != nil {

pkg/keda/kafka_scaling.go:414

• The update replaces the existing TriggerAuthentication metadata with this newly built object and only preserves the resource version. KEDA adds its own finalizer to this resource (as the integration test notes at pkg/keda/kafka_scaling_int_test.go:196-205), so a redeploy strips that finalizer and can bypass KEDA cleanup when the resource is later deleted. Preserve the existing finalizers before updating.

	ta.SetResourceVersion(existing.GetResourceVersion())
	if _, err := client.Update(ctx, ta, metav1.UpdateOptions{}); err != nil {

pkg/keda/kafka_scaling.go:74

• Only the first Kafka trigger is returned, while validation accepts repeated same-type triggers. A configuration with two Kafka triggers therefore silently drops the later trigger's settings (for example, its lag threshold); the HTTP path likewise takes only the first target value. Reject duplicate trigger types during validation or construct every declared trigger instead of silently applying one.

func kafkaTrigger(triggers []fn.KEDATrigger) fn.KEDATrigger {
	for _, t := range triggers {
		if t.Type == "kafka" {
			return t
		}

• Files reviewed: 30/30 changed files
• Comments generated: 4
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Unresolved moderate findings affect migration, KEDA transitions and status handling, validation, and Kafka authentication.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (6)

Previously missed (1) — in code that hasn't changed since the last review.

pkg/keda/kafka_scaling.go:193

• Each configured TLS path is optionalized independently here. If the CA resolves but a configured client certificate or key does not match a Secret volume, buildTriggerAuth still returns a TriggerAuthentication containing only the resolved fields; needsTriggerAuth then makes the ScaledObject reference it, so deployment succeeds but KEDA cannot perform the configured mTLS authentication. Treat any unresolved configured path as an error.

docs/reference/func_yaml.md:150

• This documentation says scale.max is any non-negative integer, but ValidateScale rejects scale.max: 0 specifically for deployer: keda because KEDA's HPA maxReplicas must be at least 1. Users following this line will get a validation error; document the KEDA-specific positive-value requirement.

- `max`: Maximum number of replicas. Non-negative integer. Default is 0 (no limit) for `deployer: knative`, not enforced for `deployer: raw`, and 10 for `deployer: keda`. See related [Knative docs](https://knative.dev/docs/serving/autoscaling/scale-bounds/#upper-bound).

pkg/functions/function_migrations.go:384

• This migration only deserializes deploy.options.scale and top-level scale. A legacy deploy.kafka field is never read or copied, so existing files using that location lose their Kafka configuration when current YAML unmarshalling runs, despite the migration described for this PR. Capture the legacy field and move it to run.kafka when the new field is absent, with a regression test.

	var disk struct {
		Deploy oldDeploy     `yaml:"deploy,omitempty"`
		Scale  *ScaleOptions `yaml:"scale,omitempty"`
	}

pkg/keda/deployer.go:465

• scale.keda.pollingInterval is documented and used for Kafka ScaledObjects, but this HTTPScaledObject builder never reads it. The documented HTTP example therefore silently uses the add-on default, so changing the setting has no effect for HTTP-only functions. Wire it through if the CRD supports it, or narrow the documented contract/schema to the supported path.

	cooldown := int32(300)
	targetValue := int64(100)
	if scale != nil && scale.KEDA != nil {
		if scale.KEDA.CooldownPeriod != nil {
			cooldown = *scale.KEDA.CooldownPeriod

pkg/keda/deployer.go:121

• Kafka resource names are suffixed with -kafka and -kafka-auth, but the DNS-length guard is now run only for HTTP triggers. A valid Kubernetes function name near the 63-character limit can therefore pass this check and then fail creating the ScaledObject or TriggerAuthentication because the suffixed name is too long. Validate the Kafka resource name budgets before the raw Deployment is created.

	if wantHTTP {
		if err := validateBridgeName(f.Name); err != nil {
			return fn.DeploymentResult{}, err
		}
	}

pkg/pipelines/tekton/pipelines_provider.go:280

• A remote pipeline with a Kafka-only KEDA function reaches this unconditional KEDA Describer call, but Kafka-only deployment does not create the HTTPScaledObject that Describer.Get requires. The pipeline therefore reports deployment failure after the resources were created; use a Kafka-capable describer or avoid the HTTP-only status lookup for this case.

• Files reviewed: 32/32 changed files
• Comments generated: 4
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Unresolved critical and moderate issues affect API compatibility, migrations, Kafka configuration, scaler lifecycle, and deployment behavior.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (7)

pkg/functions/function_migrations.go:463

• If an existing scale.keda object has tuning fields such as pollingInterval or cooldownPeriod but no triggers, replacing the whole object here discards those settings while adding the migration's default HTTP trigger. Initialize KEDA only when nil, then append the HTTP trigger to preserve the user's tuning values.

		if f.Scale.KEDA == nil || len(f.Scale.KEDA.Triggers) == 0 {
			f.Scale.KEDA = &KEDAScaleOptions{
				Triggers: []KEDATrigger{{Type: "http"}},
			}

pkg/functions/function_migrations.go:394

• The migration reads legacy scale from deploy.options.scale, but does not read or move a legacy deploy.kafka field into run.kafka. This contradicts the PR's stated automatic migration and causes existing functions using the old Kafka location to lose their broker/topic/credential configuration; their Kafka trigger will then fail validation or deploy without the intended runtime configuration. Add the legacy field to the migration input and copy it to Run.Kafka, or correct the documented migration guarantee.

	type oldDeploy struct {
		Options  oldOptions `yaml:"options,omitempty"`
		Deployer string     `yaml:"deployer,omitempty"`
		Expose   string     `yaml:"expose,omitempty"`
	}
	var disk struct {
		Deploy oldDeploy     `yaml:"deploy,omitempty"`
		Scale  *ScaleOptions `yaml:"scale,omitempty"`
	}

pkg/functions/function_scale.go:116

• Validation rejects combining HTTP and Kafka triggers, but it still accepts run.kafka with an HTTP-only KEDA trigger. AppendKafkaEnvs then switches the workload to FUNC_TRANSPORT=kafka while this deployer creates only an HTTPScaledObject, so consumer lag is never used for scaling. Reject this configuration or require a Kafka trigger whenever run.kafka is set for KEDA.

	if sawHTTP && sawKafka {
		errors = append(errors, "scale.keda.triggers must not combine type http with type kafka: they cannot scale the same Deployment together, not yet supported")
	}

pkg/keda/deployer.go:233

• On an HTTP-to-Kafka redeploy, this branch only clears the Route and exposure annotations. The bridge Service and HTTPScaledObject created by the previous HTTP deployment are owned by the still-existing Deployment, so they remain and continue to manage an HTTP scaler alongside the new Kafka ScaledObject for the same workload. Reconcile/delete the HTTP scaler and bridge before switching trigger types.

		// A prior deploy may have exposed this function over HTTP. Nothing
		// reconciles that exposure once the HTTP trigger is gone, so clear
		// it the same way deployClusterLocal does -- otherwise the old
		// Route and the Service's exposure annotations stay active,
		// pointing at a function that no longer has anything serving HTTP.

pkg/keda/deployer.go:284

• For a Kafka-to-HTTP redeploy, the HTTP path above has already created and waited for the HTTPScaledObject before this cleanup runs. Deleting the old Kafka ScaledObject afterward is asynchronous and may be held by KEDA finalizers, so the workload can temporarily have two scaler resources and the HTTP setup can race or fail. Remove and wait for the Kafka resources before creating the HTTP scaler.

		// The Kafka trigger was dropped (or never configured): remove any
		// scaler resources a prior deploy left behind, so switching back to
		// http-only doesn't leave a ScaledObject/TriggerAuthentication still
		// acting on stale Kafka lag config. Not fatal to Deploy, same as
		// Remover.Remove's treatment of these: they're owned by the

pkg/keda/deployer.go:115

• The deployer only derives wantHTTP and wantKafka; a direct API caller that invokes Deployer.Deploy without Function.Validate with a cron (or unknown) trigger gets both false, successfully deploys the raw workload, and creates no scaler. Since this method already guards other validation preconditions for direct callers, reject unsupported trigger types before creating the Deployment.

	wantHTTP := hasHTTPTrigger(triggers)
	wantKafka := hasKafkaTrigger(triggers)

pkg/keda/deployer.go:226

• Kafka-only deployments take this branch without creating an HTTPScaledObject, but the existing KEDA lister and describer unconditionally GET one (pkg/keda/lister.go:82 and pkg/keda/describer.go:74). As a result, func list and func describe fail for every Kafka-only function. Add a Kafka-aware status path (or make those components detect the Kafka ScaledObject) before shipping this trigger mode.

	} else {
		// No HTTP trigger — URL is the app service. The Service listens on
		// port 80 (routing to the container's DefaultHTTPPort via
		// targetPort), so the URL, like elsewhere in the codebase (e.g.
		// pkg/k8s/describer.go), has no explicit port.

• Files reviewed: 32/32 changed files
• Comments generated: 7
• Review effort level: Lite

[Copilot]

🟡 Changes recommended

Unresolved migration, KEDA validation/lifecycle, and scaler-status issues block approval.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (8)

pkg/functions/function_migrations.go:390

• This migration claims to move the legacy deploy.kafka configuration, but oldDeploy does not deserialize a Kafka field and no later assignment copies one into f.Run.Kafka. An existing pre-0.37 function therefore loses its Kafka settings during migration, so its Kafka consumer/scaler can no longer be recreated; capture the old field, move it to Run.Kafka, and add a regression test.

	type oldDeploy struct {
		Options  oldOptions `yaml:"options,omitempty"`
		Deployer string     `yaml:"deployer,omitempty"`
		Expose   string     `yaml:"expose,omitempty"`
	}

pkg/functions/function_migrations.go:464

• This migration decides whether to add the compatibility HTTP trigger from f.Deployer only. For an older func.yaml that has only the observed deploy.deployer: keda, the state rename below populates ActiveDeployer but leaves f.Deployer empty, so the trigger is not added; the next flag-less deploy resolves the effective deployer from ActiveDeployer and then fails validation because scale.keda.triggers is missing. Use the effective deployer (including the old observed field) when applying this default.

	// keda deployer without triggers: default to http
	if f.Deployer == "keda" {
		if f.Scale == nil {
			f.Scale = &ScaleOptions{}
		}
		if f.Scale.KEDA == nil || len(f.Scale.KEDA.Triggers) == 0 {
			f.Scale.KEDA = &KEDAScaleOptions{
				Triggers: []KEDATrigger{{Type: "http"}},
			}
		}

pkg/keda/deployer.go:157

• This direct-call guard only checks maxScale < 1; it does not enforce the cross-field bounds that ValidateScale enforces. A caller that bypasses Function.Validate can supply scale.min: 5 and scale.max: 3, causing the raw Deployment to start with 5 replicas while the ScaledObject is created with an invalid min/max pair; the deploy can return success even though KEDA will not scale it. Apply the shared scale validation (or at least reject maxScale < minScale and invalid minScale) before creating the workload.

	minScale, maxScale := replicaBounds(f)
	if maxScale < 1 {
		// deployer: keda's HTTPScaledObject/ScaledObject map scale.max
		// straight into an HPA's maxReplicas, which must be >= 1.
		return fn.DeploymentResult{}, fmt.Errorf("function %q: scale.max must be >= 1 for deployer: keda, got %d", f.Name, maxScale)

pkg/keda/deployer.go:161

• This only rejects a nil Kafka config. Direct Deployer.Deploy callers bypass Function.Validate, so a Kafka config with an empty broker, topic, or consumer group reaches buildScaledObject and creates a ScaledObject with empty Kafka metadata after the raw Deployment/Service already exist. Validate these required fields before the raw deploy (and keep the existing validation for the other Kafka constraints).

	if wantKafka && f.Run.Kafka == nil {
		return fn.DeploymentResult{}, fmt.Errorf("function %q: scale.keda.triggers has a kafka trigger but run.kafka is not configured", f.Name)
	}

pkg/keda/describer.go:84

• A missing HTTPScaledObject is not sufficient evidence that this is a Kafka-only function: an HTTP KEDA scaler can be deleted or fail to be created while its Deployment and Service remain. This fallback then reports the Deployment as healthy with a cluster-local URL, hiding the broken HTTP scaler. Please verify the Kafka ScaledObject/trigger before taking this path, and return an error when neither scaler exists.

	httpScaledObject, err := httpScaledObjectClientset.HttpV1alpha1().HTTPScaledObjects(namespace).Get(ctx, name, metav1.GetOptions{})
	hasHTTPTrigger := true
	if err != nil {
		if !errors.IsNotFound(err) {
			return fn.Instance{}, fmt.Errorf("unable to get HTTPScaledObject: %w", err)
		}
		// A Kafka-only (or otherwise no-http-trigger) function never gets an
		// HTTPScaledObject at all -- that's expected, not a failure.
		hasHTTPTrigger = false
	}

pkg/keda/kafka_scaling.go:283

• KEDA's Kafka scaler expects the SASL/PLAIN metadata value plaintext; plain is not a supported value. Any function using run.kafka.sasl.mechanism: PLAIN will therefore create a ScaledObject that cannot authenticate or scale.

		return "plain"

pkg/keda/kafka_scaling.go:49

• Because the fallback checks only whether scale.keda is nil, a direct KEDA deployment with scale.kpa set is silently converted into an HTTP-trigger deployment instead of being rejected. Function.Validate disallows that combination, but Deployer.Deploy intentionally supports callers that bypass validation; make this helper return no triggers for an incompatible KPA (and for both KPA/KEDA) so the existing empty-trigger error is reached.

func triggers(f fn.Function) []fn.KEDATrigger {
	if f.Scale != nil && f.Scale.KEDA != nil {
		return f.Scale.KEDA.Triggers
	}
	return []fn.KEDATrigger{{Type: "http"}}

pkg/keda/lister.go:93

• The same missing-HTTPScaledObject inference makes list silently classify a broken HTTP KEDA deployment as Kafka-only and return it as ready based only on Deployment status. A Kafka ScaledObject check (and an error when neither scaler exists) is needed here as well; otherwise func list can hide a failed HTTP scaler.

	httpScaledObject, err := httpScaledObjectClientset.HttpV1alpha1().HTTPScaledObjects(namespace).Get(ctx, name, metav1.GetOptions{})
	hasHTTPTrigger := true
	if err != nil {
		if !errors.IsNotFound(err) {
			return fn.ListItem{}, fmt.Errorf("unable to get HTTPScaledObject: %v", err)
		}
		// A Kafka-only (or otherwise no-http-trigger) function never gets an
		// HTTPScaledObject at all -- that's expected, not a failure.
		hasHTTPTrigger = false
	}

• Files reviewed: 34/34 changed files
• Comments generated: 3
• Review effort level: Lite

[knative-prow]

@aliok: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
unit-tests_func_main	15053a9	link	true	/test unit-tests

Your PR dashboard.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.