Merge branch 'knative-extensions:main' into chore/set-idle-timeout-unit-milliseconds

Author: lamluongbinh

control-plane/config/eventing-kafka-broker/200-controller/100-config-tracing.yaml

@@ -20,4 +20,4 @@ metadata:
   name: knative-kafka-controller-post-install
   labels:
     app.kubernetes.io/version: devel
-rules: []
+rules: null

control-plane/config/post-install/200-controller-cluster-role.yaml

@@ -26,6 +26,7 @@ const (
 	ConditionConsumerGroupConsumers          apis.ConditionType = "Consumers"
 	ConditionConsumerGroupConsumersScheduled apis.ConditionType = "ConsumersScheduled"
 	ConditionAutoscaling                     apis.ConditionType = "Autoscaler"
+	AutoscalerDisabled                                          = "AutoscalerDisabled"
 	// Labels
 	KafkaChannelNameLabel           = "kafkachannel-name"
 	ConsumerLabelSelector           = "kafka.eventing.knative.dev/metadata.uid"
@@ -98,7 +99,7 @@ func (cg *ConsumerGroup) MarkAutoscalerSucceeded() {
 }
 
 func (cg *ConsumerGroup) MarkAutoscalerDisabled() {
-	cg.GetConditionSet().Manage(cg.GetStatus()).MarkTrueWithReason(ConditionAutoscaling, "Autoscaler is disabled", "")
+	cg.GetConditionSet().Manage(cg.GetStatus()).MarkTrueWithReason(ConditionAutoscaling, AutoscalerDisabled, "")
 }
 
 func (cg *ConsumerGroup) MarkAutoscalerFailed(reason string, err error) error {

control-plane/pkg/apis/internalskafkaeventing/v1alpha1/consumer_group_lifecycle.go

@@ -232,6 +232,18 @@ func (cg *ConsumerGroup) IsNotScheduled() bool {
 	return cond.IsFalse() || cond.IsUnknown()
 }
 
+func (cg *ConsumerGroup) IsAutoscalerNotDisabled() bool {
+	condition := cg.Status.GetCondition(ConditionAutoscaling)
+
+	// If condition doesn't exist or is not true, it's not in "disabled" state
+	if condition == nil || condition.Status != corev1.ConditionTrue {
+		return true
+	}
+
+	// If condition is True but reason is not "autoscaler is disabled", then it's actively enabled
+	return condition.Reason != AutoscalerDisabled
+}
+
 func (cg *ConsumerGroup) HasDeadLetterSink() bool {
 	return hasDeadLetterSink(cg.Spec.Template.Spec.Delivery)
 }

control-plane/pkg/apis/internalskafkaeventing/v1alpha1/consumer_group_types.go

@@ -79,6 +79,10 @@ func (c *client) Topics() ([]string, error) {
 	return x, err
 }
 
+func (c *client) PartitionNotReadable(topic string, partition int32) bool {
+	return c.client.PartitionNotReadable(topic, partition)
+}
+
 func (c *client) Partitions(topic string) ([]int32, error) {
 	x, err := c.client.Partitions(topic)
 	if c.isFatalError(err) {

control-plane/pkg/kafka/clientpool/client.go

@@ -35,6 +35,11 @@ type clusterAdmin struct {
 	onFatalError func(err error)
 }
 
+func (a *clusterAdmin) Coordinator(group string) (*sarama.Broker, error) {
+	//TODO implement me
+	panic("implement me")
+}
+
 func clusterAdminFromClient(saramaClient sarama.Client, makeClusterAdmin kafka.NewClusterAdminFromClientFunc) (*clusterAdmin, error) {
 	c, ok := saramaClient.(*client)
 	if !ok {
@@ -64,6 +69,14 @@ func (a *clusterAdmin) CreateTopic(topic string, detail *sarama.TopicDetail, val
 	return err
 }
 
+func (a *clusterAdmin) ElectLeaders(electionType sarama.ElectionType, m map[string][]int32) (map[string]map[int32]*sarama.PartitionResult, error) {
+	x, err := a.clusterAdmin.ElectLeaders(electionType, m)
+	if a.isFatalError(err) {
+		a.onFatalError(err)
+	}
+	return x, err
+}
+
 func (a *clusterAdmin) ListTopics() (map[string]sarama.TopicDetail, error) {
 	x, err := a.clusterAdmin.ListTopics()
 	if a.isFatalError(err) {

control-plane/pkg/kafka/clientpool/clusteradmin.go

@@ -212,6 +212,14 @@ type saramaClientMock struct {
 	closed            bool
 }
 
+func (s *saramaClientMock) ElectLeaders(electionType sarama.ElectionType, m map[string][]int32) (map[string]map[int32]*sarama.PartitionResult, error) {
+	panic("implement me")
+}
+
+func (s *saramaClientMock) PartitionNotReadable(topic string, partition int32) bool {
+	panic("implement me")
+}
+
 func (s *saramaClientMock) CreateACLs(acls []*sarama.ResourceAcls) error {
 	panic("implement me")
 }

control-plane/pkg/kafka/consumer_group_lag_test.go

@@ -63,6 +63,20 @@ type MockKafkaClusterAdmin struct {
 	T *testing.T
 }
 
+func (m *MockKafkaClusterAdmin) ElectLeaders(electionType sarama.ElectionType, m2 map[string][]int32) (map[string]map[int32]*sarama.PartitionResult, error) {
+	if m.ErrorBrokenPipe {
+		return nil, brokenPipeError{}
+	}
+	panic("implement me")
+}
+
+func (m *MockKafkaClusterAdmin) Coordinator(group string) (*sarama.Broker, error) {
+	if m.ErrorBrokenPipe {
+		return nil, brokenPipeError{}
+	}
+	panic("implement me")
+}
+
 func (m *MockKafkaClusterAdmin) CreateACLs(acls []*sarama.ResourceAcls) error {
 	if m.ErrorBrokenPipe {
 		return brokenPipeError{}

control-plane/pkg/kafka/testing/admin_mock.go

@@ -62,6 +62,10 @@ func (m MockKafkaClient) Brokers() []*sarama.Broker {
 	panic("implement me")
 }
 
+func (m MockKafkaClient) PartitionNotReadable(topic string, partition int32) bool {
+	panic("implement me")
+}
+
 func (m MockKafkaClient) Broker(brokerID int32) (*sarama.Broker, error) {
 	if m.ShouldFailBrokenPipe {
 		return nil, brokenPipeError{}

control-plane/pkg/kafka/testing/client_mock.go

@@ -211,11 +211,18 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, cg *kafkainternals.Consu
 		}
 		cg.MarkAutoscalerSucceeded()
 	} else {
-		// If KEDA is not installed or autoscaler feature disabled, do nothing
-		cg.MarkAutoscalerDisabled()
-		if err := r.deleteKedaObjects(ctx, cg); err != nil {
-			return err
+
+		// If KEDA autoscaler feature is disabled, we need to check if it was enabled before
+		// check if the conditoion is not yet AutoscalerDisabled, only than delete KEDA objects
+		if cg.IsAutoscalerNotDisabled() {
+			logger.Debugw("Deleting KEDA objects as autoscaler is being disabled")
+			if err := r.deleteKedaObjects(ctx, cg); err != nil {
+				return err
+			}
 		}
+
+		// Mark as disabled after successful deletion, or if the feature was already disabled
+		cg.MarkAutoscalerDisabled()
 	}
 
 	logger.Debugw("Reconciling consumers")