From 0ccb73796f875d215a918ba6109de53ac4789ea4 Mon Sep 17 00:00:00 2001 From: Akshat Lakhera Date: Fri, 23 Oct 2026 00:12:42 +0530 Subject: [PATCH] Add Kafka scaler options to scale.keda kafka trigger Signed-off-by: Akshat Lakhera --- docs/reference/func_yaml.md | 5 ++ pkg/functions/function_options.go | 10 +++ pkg/functions/function_options_unit_test.go | 31 +++++++++ pkg/functions/function_scale.go | 31 +++++++++ pkg/keda/kafka_scaling.go | 17 +++++ pkg/keda/kafka_scaling_test.go | 76 +++++++++++++++++++++ schema/func_yaml-schema.json | 19 ++++++ 7 files changed, 189 insertions(+) diff --git a/docs/reference/func_yaml.md b/docs/reference/func_yaml.md index 682b8323b6..272611d633 100644 --- a/docs/reference/func_yaml.md +++ b/docs/reference/func_yaml.md @@ -172,6 +172,11 @@ The scaler sub-keys are deployer-aware: `kpa` holds Knative Pod Autoscaler setti - `kafka`: scales based on consumer group lag. Requires [`run.kafka`](#runkafka) to be configured. - `lagThreshold`: average consumer lag per partition that triggers scaling up. Default is 10. - `activationLagThreshold`: lag below which KEDA keeps replicas at 0 when `scale.min` is 0. Default is 0. + - `offsetResetPolicy`: where the scaler starts when the consumer group has no committed offset: `earliest` or `latest`. KEDA defaults to `latest`. + - `allowIdleConsumers`: boolean. Allow more replicas than the topic has partitions. KEDA defaults to false. Cannot be combined with `limitToPartitionsWithLag`. + - `excludePersistentLag`: boolean. Ignore partitions whose lag has not changed since the last poll, so stuck consumers do not keep the function scaled up. KEDA defaults to false. + - `scaleToZeroOnInvalidOffset`: boolean. Scale to zero when a partition has an invalid offset and the reset policy is `latest`. KEDA defaults to false. + - `limitToPartitionsWithLag`: boolean. Cap replicas at the number of partitions that have lag. KEDA defaults to false. Cannot be combined with `allowIdleConsumers`. ```yaml scale: diff --git a/pkg/functions/function_options.go b/pkg/functions/function_options.go index 8f974a6676..4d6478a30f 100644 --- a/pkg/functions/function_options.go +++ b/pkg/functions/function_options.go @@ -47,6 +47,16 @@ type KEDATrigger struct { TargetValue *int64 `yaml:"targetValue,omitempty" jsonschema_extras:"minimum=1"` LagThreshold *int64 `yaml:"lagThreshold,omitempty" jsonschema_extras:"minimum=1"` ActivationLagThreshold *int64 `yaml:"activationLagThreshold,omitempty" jsonschema_extras:"minimum=0"` + + // Kafka scaler options, valid only on a kafka trigger. Each maps to the KEDA + // kafka scaler metadata key of the same name and is sent only when set. + // (Blank line below keeps this note out of the generated schema description.) + + OffsetResetPolicy *string `yaml:"offsetResetPolicy,omitempty" jsonschema:"enum=earliest,enum=latest"` + AllowIdleConsumers *bool `yaml:"allowIdleConsumers,omitempty"` + ExcludePersistentLag *bool `yaml:"excludePersistentLag,omitempty"` + ScaleToZeroOnInvalidOffset *bool `yaml:"scaleToZeroOnInvalidOffset,omitempty"` + LimitToPartitionsWithLag *bool `yaml:"limitToPartitionsWithLag,omitempty"` } type KPAScaleOptions struct { diff --git a/pkg/functions/function_options_unit_test.go b/pkg/functions/function_options_unit_test.go index d797d8120d..3c2b7eb1d1 100644 --- a/pkg/functions/function_options_unit_test.go +++ b/pkg/functions/function_options_unit_test.go @@ -379,6 +379,37 @@ func Test_ValidateScale_KEDA(t *testing.T) { &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "kafka", ActivationLagThreshold: ptr.Int64(-1)}}}}, "keda", kafkaRun, 1, }, + { + "kafka scaler options all valid", + &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{ + Type: "kafka", + OffsetResetPolicy: ptr.String("earliest"), + AllowIdleConsumers: ptr.Bool(true), + ExcludePersistentLag: ptr.Bool(true), + ScaleToZeroOnInvalidOffset: ptr.Bool(true), + }}}}, + "keda", kafkaRun, 0, + }, + { + "kafka invalid offsetResetPolicy", + &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "kafka", OffsetResetPolicy: ptr.String("oldest")}}}}, + "keda", kafkaRun, 1, + }, + { + "kafka allowIdleConsumers with limitToPartitionsWithLag", + &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "kafka", AllowIdleConsumers: ptr.Bool(true), LimitToPartitionsWithLag: ptr.Bool(true)}}}}, + "keda", kafkaRun, 1, + }, + { + "kafka allowIdleConsumers false with limitToPartitionsWithLag", + &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "kafka", AllowIdleConsumers: ptr.Bool(false), LimitToPartitionsWithLag: ptr.Bool(true)}}}}, + "keda", kafkaRun, 0, + }, + { + "kafka scaler options rejected on http trigger", + &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "http", OffsetResetPolicy: ptr.String("latest"), ExcludePersistentLag: ptr.Bool(true)}}}}, + "keda", nil, 2, + }, { "repeated http trigger", &ScaleOptions{KEDA: &KEDAScaleOptions{Triggers: []KEDATrigger{{Type: "http"}, {Type: "http"}}}}, diff --git a/pkg/functions/function_scale.go b/pkg/functions/function_scale.go index bf504a3c4c..34b01d80f8 100644 --- a/pkg/functions/function_scale.go +++ b/pkg/functions/function_scale.go @@ -154,6 +154,9 @@ func validateKEDAScale(keda *KEDAScaleOptions, kafka *KafkaConfig) (errors []str if t.ActivationLagThreshold != nil { errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].activationLagThreshold is only valid for a kafka trigger", i)) } + for _, name := range t.kafkaScalerFields() { + errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].%s is only valid for a kafka trigger", i, name)) + } case "kafka": sawKafka = true if kafka == nil { @@ -170,6 +173,13 @@ func validateKEDAScale(keda *KEDAScaleOptions, kafka *KafkaConfig) (errors []str if t.ActivationLagThreshold != nil && *t.ActivationLagThreshold < 0 { errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].activationLagThreshold must not be negative", i)) } + if t.OffsetResetPolicy != nil && *t.OffsetResetPolicy != "earliest" && *t.OffsetResetPolicy != "latest" { + errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].offsetResetPolicy has invalid value %q, allowed: earliest, latest", i, *t.OffsetResetPolicy)) + } + // KEDA documents these two as mutually exclusive. + if t.AllowIdleConsumers != nil && *t.AllowIdleConsumers && t.LimitToPartitionsWithLag != nil && *t.LimitToPartitionsWithLag { + errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].limitToPartitionsWithLag cannot be combined with allowIdleConsumers", i)) + } default: errors = append(errors, fmt.Sprintf("scale.keda.triggers[%d].type has invalid value %q, allowed: http, kafka", i, t.Type)) } @@ -192,3 +202,24 @@ func validateKPAScale(kpa *KPAScaleOptions) (errors []string) { } return } + +// kafkaScalerFields returns the yaml names of the kafka-scaler options set on t +// (not lagThreshold/activationLagThreshold, which are checked separately). +func (t KEDATrigger) kafkaScalerFields() (set []string) { + if t.OffsetResetPolicy != nil { + set = append(set, "offsetResetPolicy") + } + if t.AllowIdleConsumers != nil { + set = append(set, "allowIdleConsumers") + } + if t.ExcludePersistentLag != nil { + set = append(set, "excludePersistentLag") + } + if t.ScaleToZeroOnInvalidOffset != nil { + set = append(set, "scaleToZeroOnInvalidOffset") + } + if t.LimitToPartitionsWithLag != nil { + set = append(set, "limitToPartitionsWithLag") + } + return +} diff --git a/pkg/keda/kafka_scaling.go b/pkg/keda/kafka_scaling.go index 59a53e0115..97bf623d9f 100644 --- a/pkg/keda/kafka_scaling.go +++ b/pkg/keda/kafka_scaling.go @@ -3,6 +3,7 @@ package keda import ( "context" "fmt" + "strconv" "strings" "github.com/kedacore/http-add-on/operator/generated/clientset/versioned" @@ -425,6 +426,22 @@ func buildScaledObject(f fn.Function, trigger fn.KEDATrigger, deployment *v1.Dep triggerMeta["activationLagThreshold"] = fmt.Sprintf("%d", *trigger.ActivationLagThreshold) } + // Optional KEDA kafka scaler settings: sent only when set so KEDA's own + // defaults apply otherwise. + if trigger.OffsetResetPolicy != nil { + triggerMeta["offsetResetPolicy"] = *trigger.OffsetResetPolicy + } + for key, v := range map[string]*bool{ + "allowIdleConsumers": trigger.AllowIdleConsumers, + "excludePersistentLag": trigger.ExcludePersistentLag, + "scaleToZeroOnInvalidOffset": trigger.ScaleToZeroOnInvalidOffset, + "limitToPartitionsWithLag": trigger.LimitToPartitionsWithLag, + } { + if v != nil { + triggerMeta[key] = strconv.FormatBool(*v) + } + } + if kafka.SecurityProtocol == "SSL" || kafka.SecurityProtocol == "SASL_SSL" { // Enable KEDA's TLS handshake based on securityProtocol, not the // presence of run.kafka.tls: a valid config can set SSL/SASL_SSL and diff --git a/pkg/keda/kafka_scaling_test.go b/pkg/keda/kafka_scaling_test.go index f075345d8d..34b26a53cc 100644 --- a/pkg/keda/kafka_scaling_test.go +++ b/pkg/keda/kafka_scaling_test.go @@ -1050,3 +1050,79 @@ func TestKedaSASLType(t *testing.T) { } func strPtr(s string) *string { return &s } + +// TestBuildScaledObject_KafkaScalerOptions verifies the optional kafka scaler +// options reach the trigger metadata under KEDA's key names, and that nothing +// is emitted for options that are unset. +func TestBuildScaledObject_KafkaScalerOptions(t *testing.T) { + b := func(v bool) *bool { return &v } + tests := []struct { + name string + trigger fn.KEDATrigger + want map[string]string + absent []string + }{ + { + name: "none set emits none", + trigger: fn.KEDATrigger{Type: "kafka"}, + absent: []string{"offsetResetPolicy", "allowIdleConsumers", "excludePersistentLag", "scaleToZeroOnInvalidOffset", "limitToPartitionsWithLag"}, + }, + { + name: "offsetResetPolicy earliest", + trigger: fn.KEDATrigger{Type: "kafka", OffsetResetPolicy: strPtr("earliest")}, + want: map[string]string{"offsetResetPolicy": "earliest"}, + absent: []string{"allowIdleConsumers", "limitToPartitionsWithLag"}, + }, + { + name: "all set true", + trigger: fn.KEDATrigger{ + Type: "kafka", + OffsetResetPolicy: strPtr("latest"), + AllowIdleConsumers: b(true), + ExcludePersistentLag: b(true), + ScaleToZeroOnInvalidOffset: b(true), + LimitToPartitionsWithLag: b(true), + }, + want: map[string]string{ + "offsetResetPolicy": "latest", + "allowIdleConsumers": "true", + "excludePersistentLag": "true", + "scaleToZeroOnInvalidOffset": "true", + "limitToPartitionsWithLag": "true", + }, + }, + { + name: "explicit false is emitted", + trigger: fn.KEDATrigger{Type: "kafka", ExcludePersistentLag: b(false)}, + want: map[string]string{"excludePersistentLag": "false"}, + absent: []string{"allowIdleConsumers"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := fn.Function{ + Name: "test-func", + Run: fn.RunSpec{Kafka: &fn.KafkaConfig{ + Brokers: "broker:9092", Topic: "t", ConsumerGroup: "g", + }}, + } + so := buildScaledObject(f, tt.trigger, testDeployment(), "default", 0, 5) + if so == nil { + t.Fatal("expected ScaledObject, got nil") + } + spec := so.Object["spec"].(map[string]interface{}) + trig := spec["triggers"].([]interface{})[0].(map[string]interface{}) + meta := trig["metadata"].(map[string]interface{}) + for k, v := range tt.want { + if meta[k] != v { + t.Errorf("metadata[%s] = %v, want %q", k, meta[k], v) + } + } + for _, k := range tt.absent { + if _, ok := meta[k]; ok { + t.Errorf("metadata[%s] should be absent, got %v", k, meta[k]) + } + } + }) + } +} diff --git a/schema/func_yaml-schema.json b/schema/func_yaml-schema.json index 3a70a2878d..2b8b196e6d 100644 --- a/schema/func_yaml-schema.json +++ b/schema/func_yaml-schema.json @@ -338,6 +338,25 @@ "activationLagThreshold": { "type": "integer", "minimum": 0 + }, + "offsetResetPolicy": { + "enum": [ + "earliest", + "latest" + ], + "type": "string" + }, + "allowIdleConsumers": { + "type": "boolean" + }, + "excludePersistentLag": { + "type": "boolean" + }, + "scaleToZeroOnInvalidOffset": { + "type": "boolean" + }, + "limitToPartitionsWithLag": { + "type": "boolean" } }, "additionalProperties": false,