Repository: knative-sandbox/eventing-kafka-broker
Pull request #2022 (2022-03-18 16:43:42)

Consumer cleanup of unused field
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537 

aavarghese: /cc @pierDipi 
knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2022#" title="Author self-approved">aavarghese</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **pierdipi** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @pierdipi` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["pierdipi"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2022](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2d80afb) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/d525d4bd0c2a1708197c63278f5aaf632331f034?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d525d4b) will **increase** coverage by `0.03%`.
> The diff coverage is `0.00%`.

```diff
@@             Coverage Diff              @@
##               main    #2022      +/-   ##
============================================
+ Coverage     67.26%   67.29%   +0.03%     
  Complexity      640      640              
============================================
  Files           141      141              
  Lines          6708     6715       +7     
  Branches        187      187              
============================================
+ Hits           4512     4519       +7     
  Misses         1820     1820              
  Partials        376      376              
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.75% <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | `50.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/v2/sourcev2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2UvdjIvc291cmNldjIuZ28=) | `89.88% <0.00%> (+0.86%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [d525d4b...2d80afb](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2022?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).

knative-prow-robot: @aavarghese: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl | 2d80afbd9c5ddbc5f2b0717214d9fe9a1c27ee65 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2022/pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl/1504861118697312256) | true | `/test pull-knative-sandbox-eventing-kafka-broker-channel-integration-tests-ssl`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/2022). [Your PR dashboard](https://gubernator.knative.dev/pr/aavarghese).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml
index 00b90633a..4ff410dd4 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-descheduler.yaml
@@ -20,35 +20,9 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 data:
-  _example: |
-    ################################
-    #                              #
-    #    EXAMPLE CONFIGURATION     #
-    #                              #
-    ################################
-
-    # This block is not actually functional configuration,
-    # but serves to illustrate the available configuration
-    # options and document them in a way that is accessible
-    # to users that `kubectl edit` this config map.
-    #
-    # These sample configuration options may be copied out of
-    # this example block and unindented to be in the data block
-    # to actually change the configuration.
-    #
-    # These predicate plugins are used to filter out pods that
-    # a vreplica cannot be removed from due to not satisfying the
-    # predicate condition. For each pod, the descheduler
-    # will call filter plugins in their configured order.
-    # predicates: |+
+    predicates: |+
                     []
-    #
-    # These priority plugins are used to score each pod that
-    # has passed the filtering phase. The scheduler will then
-    # select the pod with the highest weighted score sum to
-    # remove vreplica from. If two pods have an equal score,
-    # a pod is chosen randomly.
-    # priorities: |+
+    priorities: |+
                   [
                     {"Name": "RemoveWithEvenPodSpreadPriority",
                     "Weight": 10,
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
index a419b8471..cfc3310f1 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/100-config-kafka-scheduler.yaml
@@ -20,27 +20,7 @@ metadata:
   labels:
     kafka.eventing.knative.dev/release: devel
 data:
-  _example: |
-    ################################
-    #                              #
-    #    EXAMPLE CONFIGURATION     #
-    #                              #
-    ################################
-
-    # This block is not actually functional configuration,
-    # but serves to illustrate the available configuration
-    # options and document them in a way that is accessible
-    # to users that `kubectl edit` this config map.
-    #
-    # These sample configuration options may be copied out of
-    # this example block and unindented to be in the data block
-    # to actually change the configuration.
-    #
-    # These predicate plugins are used to filter out pods that
-    # a vreplica cannot be placed on due to not satisfying the
-    # predicate condition. For each pod, the scheduler
-    # will call filter plugins in their configured order.
-    # predicates: |+
+    predicates: |+
                   [
                     {"Name": "PodFitsResources"},
                     {"Name": "NoMaxResourceCount",
@@ -48,12 +28,7 @@ data:
                     {"Name": "EvenPodSpread",
                     "Args": "{\"MaxSkew\": 2}"}
                   ]
-    #
-    # These priority plugins are used to score each pod that
-    # has passed the filtering phase. The scheduler will then
-    # select the pod with the highest weighted score sum. If
-    # two pods have an equal score, a pod is chosen randomly.
-    # priorities: |+
+    priorities: |+
                   [
                     {"Name": "AvailabilityZonePriority",
                     "Weight": 10,
diff --git a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
index ea50753e8..8b0a78e62 100644
--- a/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
+++ b/control-plane/config/eventing-kafka-broker/200-controller/500-controller.yaml
@@ -129,6 +129,20 @@ spec:
                 fieldRef:
                   fieldPath: metadata.namespace
 
+            # How often (in seconds) the autoscaler tries to scale down the statefulset.
+            - name: AUTOSCALER_REFRESH_PERIOD
+              value: '100'
+
+            # The number of virtual replicas each adapter pod can handle.
+            - name: POD_CAPACITY
+              value: '20'
+
+            - name: SCHEDULER_CONFIG
+              value: 'config-kafka-scheduler'
+
+            - name: DESCHEDULER_CONFIG
+              value: 'config-kafka-descheduler'
+
             - name: CONFIG_LEADERELECTION_NAME
               value: config-kafka-leader-election
             - name: CONFIG_LOGGING_NAME
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index f5ac60934..6a449fc59 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -44,10 +44,7 @@ import (
 )
 
 type Reconciler struct {
-	SerDe contract.FormatSerDe
-
-	Enqueue func(c *kafkainternals.Consumer)
-
+	SerDe               contract.FormatSerDe
 	Resolver            *resolver.URIResolver
 	Tracker             tracker.Interface
 	ConsumerGroupLister kafkainternalslisters.ConsumerGroupLister
@@ -136,7 +133,7 @@ func (r Reconciler) reconcileContractResource(ctx context.Context, c *kafkainter
 func (r Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainternals.Consumer) (*contract.Egress, error) {
 	destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Subscriber, c)
 	if err != nil {
-		return nil, fmt.Errorf("failed to resolve subcriber: %w", err)
+		return nil, fmt.Errorf("failed to resolve subscriber: %w", err)
 	}
 	c.Status.SubscriberURI = destination
 
diff --git a/control-plane/pkg/reconciler/consumergroup/controller.go b/control-plane/pkg/reconciler/consumergroup/controller.go
index 23c9b86c6..d423a7654 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller.go
@@ -23,6 +23,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/kelseyhightower/envconfig"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 	appsv1 "k8s.io/api/apps/v1"
@@ -58,31 +59,36 @@ const (
 	// KafkaChannelScheduler is the key for the scheduler map for any KafkaChannel.
 	// Keep this constant lowercase.
 	KafkaChannelScheduler = "kafkachannel"
-
-	// ConfigKafkaSchedulerName is the name of the ConfigMap to configure the scheduler.
-	ConfigKafkaSchedulerName = "config-kafka-scheduler"
-	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
-	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
 )
 
+type envConfig struct {
+	SchedulerRefreshPeriod     int64  `envconfig:"AUTOSCALER_REFRESH_PERIOD" required:"true"`
+	PodCapacity                int32  `envconfig:"POD_CAPACITY" required:"true"`
+	SchedulerPolicyConfigMap   string `envconfig:"SCHEDULER_CONFIG" required:"true"`
+	DeSchedulerPolicyConfigMap string `envconfig:"DESCHEDULER_CONFIG" required:"true"`
+}
+
 type SchedulerConfig struct {
-	StatefulSetName     string
-	RefreshPeriod       time.Duration
-	Capacity            int32
-	SchedulerPolicyType scheduler.SchedulerPolicyType
-	SchedulerPolicy     *scheduler.SchedulerPolicy
-	DeSchedulerPolicy   *scheduler.SchedulerPolicy
+	StatefulSetName   string
+	RefreshPeriod     time.Duration
+	Capacity          int32
+	SchedulerPolicy   *scheduler.SchedulerPolicy
+	DeSchedulerPolicy *scheduler.SchedulerPolicy
 }
 
 func NewController(ctx context.Context) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	env := &envConfig{}
+	if err := envconfig.Process("", env); err != nil {
+		logger.Panicf("unable to process required environment variables: %v", err)
+	}
 
-	// TODO(pierDipi) use env variables to configure these.
 	c := SchedulerConfig{
-		RefreshPeriod:       100 * time.Second,
-		Capacity:            20,
-		SchedulerPolicyType: "", // old non-HA scheduler
-		SchedulerPolicy:     schedulerPolicyFromConfigMapOrFail(ctx, ConfigKafkaSchedulerName),
-		DeSchedulerPolicy:   schedulerPolicyFromConfigMapOrFail(ctx, ConfigKafkaDeSchedulerName),
+		RefreshPeriod:     time.Duration(env.SchedulerRefreshPeriod) * time.Second,
+		Capacity:          env.PodCapacity,
+		SchedulerPolicy:   schedulerPolicyFromConfigMapOrFail(ctx, env.SchedulerPolicyConfigMap),
+		DeSchedulerPolicy: schedulerPolicyFromConfigMapOrFail(ctx, env.DeSchedulerPolicyConfigMap),
 	}
 
 	schedulers := map[string]scheduler.Scheduler{
@@ -143,12 +149,11 @@ func createKafkaScheduler(ctx context.Context, c SchedulerConfig, ssName string)
 	return createStatefulSetScheduler(
 		ctx,
 		SchedulerConfig{
-			StatefulSetName:     ssName,
-			RefreshPeriod:       c.RefreshPeriod,
-			Capacity:            c.Capacity,
-			SchedulerPolicyType: c.SchedulerPolicyType,
-			SchedulerPolicy:     c.SchedulerPolicy,
-			DeSchedulerPolicy:   c.DeSchedulerPolicy,
+			StatefulSetName:   ssName,
+			RefreshPeriod:     c.RefreshPeriod,
+			Capacity:          c.Capacity,
+			SchedulerPolicy:   c.SchedulerPolicy,
+			DeSchedulerPolicy: c.DeSchedulerPolicy,
 		},
 		func() ([]scheduler.VPod, error) {
 			consumerGroups, err := lister.List(labels.SelectorFromSet(getSelectorLabel(ssName)))
@@ -192,7 +197,7 @@ func createStatefulSetScheduler(ctx context.Context, c SchedulerConfig, lister s
 		lister,
 		c.RefreshPeriod,
 		c.Capacity,
-		c.SchedulerPolicyType,
+		"", //  scheduler.SchedulerPolicyType field only applicable for old scheduler policy
 		nodeinformer.Get(ctx).Lister(),
 		newEvictor(ctx, zap.String("kafka.eventing.knative.dev/component", "evictor")).evict,
 		c.SchedulerPolicy,
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index 90ab36c6b..c2a95da3c 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingclientset "knative.dev/eventing/pkg/client/clientset/versioned"
 	eventinglisters "knative.dev/eventing/pkg/client/listers/eventing/v1"
@@ -83,31 +84,17 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigge
 		return nil
 	}
 
-	cg, err := r.reconcileConsumerGroup(ctx, trigger)
+	cg, err := r.reconcileConsumerGroup(ctx, broker, trigger)
 	if err != nil {
 		trigger.Status.MarkDependencyFailed("failed to reconcile consumer group", err.Error())
 		return err
 	}
-	if cg.IsReady() {
-		trigger.Status.MarkDependencySucceeded()
-	} else {
-		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
-		if topLevelCondition == nil {
-			trigger.Status.MarkDependencyUnknown("failed to reconcile consumer group", "consumer group not ready")
-		} else {
-			trigger.Status.MarkDependencyFailed(topLevelCondition.Reason, topLevelCondition.Message)
-		}
-	}
-	trigger.Status.SubscriberURI = cg.Status.SubscriberURI
-	trigger.Status.MarkSubscriberResolvedSucceeded()
-
-	trigger.Status.DeadLetterSinkURI = cg.Status.DeadLetterSinkURI
-	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+	propagateConsumerGroupStatus(cg, trigger)
 
 	return nil
 }
 
-func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
+func (r Reconciler) reconcileConsumerGroup(ctx context.Context, broker *eventing.Broker, trigger *eventing.Trigger) (*internalscg.ConsumerGroup, error) {
 
 	var deliveryOrdering = internals.Unordered
 	var err error
@@ -141,8 +128,9 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 					Topics:  []string{},                                                                               //todo get topics from broker resource
 					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{"group.id": string(trigger.UID)}}, //todo get bootstrap.servers from broker resource
 					Delivery: &internalscg.DeliverySpec{
-						DeliverySpec: trigger.Spec.Delivery,
-						Ordering:     deliveryOrdering},
+						DeliverySpec: deliverySpec(broker, trigger),
+						Ordering:     deliveryOrdering,
+					},
 					Filters: &internalscg.Filters{
 						Filter:  trigger.Spec.Filter,
 						Filters: trigger.Spec.Filters,
@@ -187,6 +175,32 @@ func (r Reconciler) reconcileConsumerGroup(ctx context.Context, trigger *eventin
 	return cg, nil
 }
 
+func deliverySpec(broker *eventing.Broker, trigger *eventing.Trigger) *eventingduck.DeliverySpec {
+	// TOOD(pierDipi) use `Merge` in https://github.com/knative/eventing/pull/6277/files
+	if trigger.Spec.Delivery != nil {
+		return trigger.Spec.Delivery
+	}
+	return broker.Spec.Delivery
+}
+
+func propagateConsumerGroupStatus(cg *internalscg.ConsumerGroup, trigger *eventing.Trigger) {
+	if cg.IsReady() {
+		trigger.Status.MarkDependencySucceeded()
+	} else {
+		topLevelCondition := cg.GetConditionSet().Manage(cg.GetStatus()).GetTopLevelCondition()
+		if topLevelCondition == nil {
+			trigger.Status.MarkDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready")
+		} else {
+			trigger.Status.MarkDependencyFailed(topLevelCondition.Reason, topLevelCondition.Message)
+		}
+	}
+	trigger.Status.SubscriberURI = cg.Status.SubscriberURI
+	trigger.Status.MarkSubscriberResolvedSucceeded()
+
+	trigger.Status.DeadLetterSinkURI = cg.Status.DeadLetterSinkURI
+	trigger.Status.MarkDeadLetterSinkResolvedSucceeded()
+}
+
 func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/controller_test.go b/control-plane/pkg/reconciler/consumergroup/controller_test.go
index 3ee7d7a21..a56639ac8 100644
--- a/control-plane/pkg/reconciler/consumergroup/controller_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/controller_test.go
@@ -17,6 +17,7 @@
 package consumergroup
 
 import (
+	"os"
 	"testing"
 
 	"github.com/stretchr/testify/require"
@@ -35,6 +36,15 @@ import (
 	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
 )
 
+const (
+	RefreshPeriod = "100"
+	PodCapacity   = "20"
+	// ConfigKafkaSchedulerName is the name of the ConfigMap to configure the scheduler.
+	ConfigKafkaSchedulerName = "config-kafka-scheduler"
+	// ConfigKafkaDeSchedulerName is the name of the ConfigMap to configure the descheduler.
+	ConfigKafkaDeSchedulerName = "config-kafka-descheduler"
+)
+
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
@@ -77,6 +87,10 @@ func TestNewController(t *testing.T) {
 		},
 	)
 
+	os.Setenv("AUTOSCALER_REFRESH_PERIOD", RefreshPeriod)
+	os.Setenv("POD_CAPACITY", PodCapacity)
+	os.Setenv("SCHEDULER_CONFIG", ConfigKafkaSchedulerName)
+	os.Setenv("DESCHEDULER_CONFIG", ConfigKafkaDeSchedulerName)
 	controller := NewController(ctx)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
index 19e7102e5..417d586a8 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2_test.go
@@ -21,6 +21,9 @@ import (
 	"fmt"
 	"testing"
 
+	"k8s.io/utils/pointer"
+	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
+
 	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 
@@ -62,6 +65,8 @@ var (
 		Host:   "localhost",
 		Path:   "/path",
 	}
+
+	exponential = eventingduck.BackoffPolicyExponential
 )
 
 func TestReconcileKind(t *testing.T) {
@@ -102,7 +107,52 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
+						withDeadLetterSinkURI(""),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - fallback to broker delivery",
+			Objects: []runtime.Object{
+				NewBroker(
+					BrokerReady,
+					WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
+				),
+				newTrigger(),
+			},
+			Key: testKey,
+			WantCreates: []runtime.Object{
+				NewConsumerGroup(
+					WithConsumerGroupName(TriggerUUID),
+					WithConsumerGroupNamespace(triggerNamespace),
+					WithConsumerGroupOwnerRef(kmeta.NewControllerRef(newTrigger())),
+					WithConsumerGroupMetaLabels(OwnerAsTriggerLabel),
+					WithConsumerGroupLabels(ConsumerTriggerLabel),
+					ConsumerGroupConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(),
+						ConsumerConfigs(
+							ConsumerGroupIdConfig(TriggerUUID),
+						),
+						ConsumerDelivery(NewConsumerSpecDelivery(
+							internals.Unordered,
+							NewConsumerRetry(10),
+							NewConsumerBackoffDelay("PT2S"),
+							NewConsumerBackoffPolicy(eventingduck.BackoffPolicyExponential),
+						)),
+						ConsumerFilters(NewConsumerSpecFilters()),
+						ConsumerReply(ConsumerTopicReply()),
+					)),
+				),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: newTrigger(
+						reconcilertesting.WithInitTriggerConditions,
+						reconcilertesting.WithTriggerBrokerReady(),
+						withTriggerSubscriberResolvedSucceeded(),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -141,7 +191,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Ordered)),
 					),
@@ -181,7 +231,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 						reconcilertesting.WithAnnotation(deliveryOrderAnnotation, string(internals.Unordered)),
 					),
@@ -313,7 +363,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -388,7 +438,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(""),
 					),
 				},
@@ -464,7 +514,7 @@ func TestReconcileKind(t *testing.T) {
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerReady(),
 						withTriggerSubscriberResolvedSucceeded(),
-						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group not ready"),
+						reconcilertesting.WithTriggerDependencyUnknown("failed to reconcile consumer group", "consumer group is not ready"),
 						withDeadLetterSinkURI(url.String()),
 					),
 				},