Repository: knative-sandbox/eventing-kafka-broker
Pull request #2021 (2022-03-18 16:40:29)

Moving hardcoded values to controller environment variables, and config maps ready (Consumer Scheduling and Scaling)
Signed-off-by: aavarghese <avarghese@us.ibm.com>

Part of #1537 


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/2021#" 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 **rhuss** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @rhuss` 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":["rhuss"]} -->
aavarghese: /cc @pierDipi 
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-unit-tests | 6970757fc5fc4cbbaef3b4e7df346a39625b78c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2021/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1504860305719562240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/2021). [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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2021](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6970757) 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 `14.58%`.
> The diff coverage is `n/a`.

```diff
@@              Coverage Diff              @@
##               main    #2021       +/-   ##
=============================================
+ Coverage     67.26%   81.84%   +14.58%     
- Complexity      640      642        +2     
=============================================
  Files           141       72       -69     
  Lines          6708     2176     -4532     
  Branches        187      187               
=============================================
- Hits           4512     1781     -2731     
+ Misses         1820      294     -1526     
+ Partials        376      101      -275     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.84% <ø> (+0.09%)` | :arrow_up: |

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/2021?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | | |
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | | |
| [control-plane/pkg/core/config/utils.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvdXRpbHMuZ28=) | | |
| [control-plane/pkg/kafka/consumer\_group\_lag.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvY29uc3VtZXJfZ3JvdXBfbGFnLmdv) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| [...ntrol-plane/pkg/reconciler/trigger/v2/triggerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL3RyaWdnZXJ2Mi5nbw==) | | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | | |
| [control-plane/pkg/receiver/path.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvcGF0aC5nbw==) | | |
| [...trol-plane/pkg/prober/probertesting/mock\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2JlcnRlc3RpbmcvbW9ja19wcm9iZXIuZ28=) | | |
| ... and [60 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021?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/2021?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [d525d4b...6970757](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2021?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 tests **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-unit-tests | 6970757fc5fc4cbbaef3b4e7df346a39625b78c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2021/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1504860305719562240) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`
pull-knative-sandbox-eventing-kafka-broker-build-tests | 6970757fc5fc4cbbaef3b4e7df346a39625b78c7 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2021/pull-knative-sandbox-eventing-kafka-broker-build-tests/1504860305681813504) | true | `/test pull-knative-sandbox-eventing-kafka-broker-build-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/2021). [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 -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/consumergroup/controller.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2021/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1504864906178990080/artifacts/line-cov.html#file0) | 62.1% | 62.9% | 0.7

knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2021#" title="Author self-approved">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2021#pullrequestreview-914651713" title="Approved">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->
knative-prow-robot: New changes are detected. LGTM label has been removed.
aavarghese: @pierDipi sorry!! There was a newline lint err. Can you lgtm again please?
knative-prow-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2021#" title="Author self-approved">aavarghese</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2021#pullrequestreview-914660963" title="LGTM">pierDipi</a>*

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).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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":[]} -->

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/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()),
 					),
 				},