Install KEDA in tests

control-plane/pkg/autoscaler/autoscaler_api.go

@@ -50,15 +50,15 @@ const (
 	AutoscalingActivationLagThreshold = Autoscaling + "/activation-lag-threshold"
 
 	// DefaultPollingInterval is the default value for AutoscalingPollingIntervalAnnotation.
-	DefaultPollingInterval = 30
+	DefaultPollingInterval = 10
 	// DefaultCooldownPeriod is the default value for AutoscalingCooldownPeriodAnnotation.
 	DefaultCooldownPeriod = 300
 	// DefaultMinReplicaCount is the default value for AutoscalingMinScaleAnnotation
 	DefaultMinReplicaCount = 0
 	// DefaultMaxReplicaCount is the default value for AutoscalingMaxScaleAnnotation.
 	DefaultMaxReplicaCount = 50
 	// DefaultLagThreshold is the default value for AutoscalingLagThreshold.
-	DefaultLagThreshold = 10
+	DefaultLagThreshold = 100
 	// DefaultActivationLagThreshold is the default value for AutoscalingActivationLagThreshold.
 	DefaultActivationLagThreshold = 1
 )

control-plane/pkg/reconciler/consumergroup/consumergroup.go

@@ -44,6 +44,7 @@
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/metrics"
 	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
 
 	sources "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1"
 
@@ -113,6 +114,7 @@
 	ConfigMapLister corelisters.ConfigMapLister
 	PodLister       corelisters.PodLister
 	KubeClient      kubernetes.Interface
+	Resolver        *resolver.URIResolver
 
 	NameGenerator names.NameGenerator
 
@@ -163,7 +165,7 @@
 		return err
 	}
 
-	errCondition, err := r.propagateStatus(cg)
+	errCondition, err := r.propagateStatus(ctx, cg)
 	if err != nil {
 		return cg.MarkReconcileConsumersFailed("PropagateConsumerStatus", err)
 	}
@@ -442,7 +444,7 @@
 	return placementConsumers
 }
 
-func (r *Reconciler) propagateStatus(cg *kafkainternals.ConsumerGroup) (*apis.Condition, error) {
+func (r *Reconciler) propagateStatus(ctx context.Context, cg *kafkainternals.ConsumerGroup) (*apis.Condition, error) {
 	consumers, err := r.ConsumerLister.Consumers(cg.GetNamespace()).List(labels.SelectorFromSet(cg.Spec.Selector))
 	if err != nil {
 		return nil, fmt.Errorf("failed to list consumers for selector %+v: %w", cg.Spec.Selector, err)
@@ -471,6 +473,14 @@
 	}
 	cg.Status.Replicas = pointer.Int32(count)
 
+	if cg.Spec.Replicas != nil && *cg.Spec.Replicas == 0 {
+		subscriber, err := r.Resolver.URIFromDestinationV1(ctx, cg.Spec.Template.Spec.Subscriber, cg)
+		if err != nil {
+			return condition, fmt.Errorf("failed to resolve subscribed URI: %w", err)
+		}
+		cg.Status.SubscriberURI = subscriber
+	}
+
 	return condition, nil
 }
 

control-plane/pkg/reconciler/consumergroup/controller.go

@@ -45,6 +45,7 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
 	"knative.dev/pkg/system"
 
 	statefulsetscheduler "knative.dev/eventing/pkg/scheduler/statefulset"
@@ -151,6 +152,8 @@ func NewController(ctx context.Context, watcher configmap.Watcher) *controller.I
 		}
 	})
 
+	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+
 	configStore := config.NewStore(ctx, func(name string, value *config.KafkaFeatureFlags) {
 		r.KafkaFeatureFlags.Reset(value)
 		impl.GlobalResync(consumerGroupInformer.Informer())

control-plane/pkg/reconciler/consumergroup/controller_test.go

@@ -24,6 +24,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"

hack/update-keda.sh

@@ -0,0 +1,29 @@
+#!/usr/bin/env bash
+
+# Copyright 2023 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# check that we have exactly one argument denoting which keda version to update to
+
+if [[ "$#" -ne 1 ]]; then
+	echo "Usage: $0 <keda_version>"
+	exit 1
+fi
+
+keda_version=$1
+
+echo "Using KEDA version: ${keda_version}"
+
+curl -L "https://github.com/kedacore/keda/releases/download/v${keda_version}/keda-${keda_version}.yaml" \
+	> $(dirname $0)/../third_party/keda/keda.yaml

test/config/100-config-kafka-features.yaml

@@ -6,7 +6,7 @@ metadata:
 data:
   dispatcher.rate-limiter: "enabled"
   dispatcher.ordered-executor-metrics: "enabled"
-  controller.autoscaler: "enabled"
+  controller.autoscaler: "disabled"
   triggers.consumergroup.template: "knative-trigger-{{ .Namespace }}-{{ .Name }}"
   brokers.topic.template: "knative-broker-{{ .Namespace }}-{{ .Name }}"
   channels.topic.template: "knative-messaging-kafka.{{ .Namespace }}.{{ .Name }}"

test/e2e-common.sh

@@ -112,13 +112,19 @@ function knative_eventing() {
   ./test/upload-test-images.sh "test/test_images" e2e || fail_test "Error uploading test images"
 
   kafka_setup
+  keda_setup
 }
 
 function kafka_setup() {
   echo ">> Prepare to deploy Strimzi"
   ./test/kafka/kafka_setup.sh || fail_test "Failed to set up Kafka cluster"
 }
 
+function keda_setup() {
+  echo ">> Prepare to deploy KEDA"
+  ./test/keda/keda_setup.sh || fail_test "Failed to set up KEDA"
+}
+
 function build_components_from_source() {
 
   [ -f "${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}" ] && rm "${EVENTING_KAFKA_CONTROL_PLANE_ARTIFACT}"

test/keda/keda_setup.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+
+# Copyright 2023 The Knative Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -euo pipefail
+
+source $(dirname $0)/../../vendor/knative.dev/hack/e2e-tests.sh
+
+header "Applying KEDA Operator file"
+kubectl apply -f $(dirname $0)/../../third_party/keda/keda.yaml
+
+wait_until_pods_running keda || fail_test "Failed to start up KEDA operator"