Add Kafka scaler options to scale.keda kafka trigger

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:

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 {

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"}}}},

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
+}

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

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])
+				}
+			}
+		})
+	}
+}

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,