Repository: knative-sandbox/eventing-kafka-broker
Pull request #1419 (2021-11-05 12:16:28)

Better naming and consistency for config.Env and broker.Config
<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Better naming and consistency for `config.Env`
- Got rid of `broker.Config` which only had 1 additional field. That `broker.Config` thing was used in source, channel and sink. Changed those usages to `config.Env`.
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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/1419#" title="Author self-approved">aliok</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)~~ [aliok]

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":[]} -->
google-cla[bot]: All (the pull request submitter and all commit authors) CLAs are signed, **but** one or more commits were authored or co-authored by someone other than the pull request submitter.

We need to confirm that all authors are ok with their commits being contributed to this project.  Please have them confirm that by leaving a comment that contains only `@googlebot I consent.` in this pull request.

*Note to project maintainer:* There may be cases where the author cannot leave a comment, or the comment is not properly detected as consent.  In those cases, you can manually confirm consent of the commit author(s), and set the `cla` label to `yes` (if enabled on your project).

ℹ️ **Googlers: [Go here](https://goto.google.com/prinfo/https%3A%2F%2Fgithub.com%2Fknative-sandbox%2Feventing-kafka-broker%2Fpull%2F1419) for more info**.

<!-- need_author_consent -->
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/kafka/topic.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1419/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1456596279843885056/artifacts/line-cov.html#file5) | 100.0% | 97.5% | -2.5

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1419](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (d809f03) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c108f41e59088e88a366f70eb57df62f47daf7c8?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (c108f41) will **decrease** coverage by `15.08%`.
> The diff coverage is `50.94%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@              Coverage Diff              @@
##               main    #1419       +/-   ##
=============================================
- Coverage     76.24%   61.15%   -15.09%     
=============================================
  Files            98       35       -63     
  Lines          3658     2039     -1619     
  Branches        166        0      -166     
=============================================
- Hits           2789     1247     -1542     
- Misses          653      662        +9     
+ Partials        216      130       -86     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

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/1419?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | `97.67% <0.00%> (-2.33%)` | :arrow_down: |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | `67.83% <50.00%> (ø)` | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | `76.05% <66.66%> (ø)` | |
| [control-plane/pkg/reconciler/channel/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NvbnRyb2xsZXIuZ28=) | `85.71% <77.77%> (ø)` | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `70.44% <80.00%> (ø)` | |
| [control-plane/pkg/reconciler/trigger/trigger.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXIuZ28=) | `70.83% <80.00%> (ø)` | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | `72.34% <100.00%> (ø)` | |
| [control-plane/pkg/reconciler/source/source.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zb3VyY2Uvc291cmNlLmdv) | `57.69% <100.00%> (ø)` | |
| ... and [68 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419/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/1419?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/1419?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [c108f41...d809f03](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1419?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: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1419#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1419#pullrequestreview-799851046" 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)~~ [aliok,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/cmd/kafka-controller/main.go b/control-plane/cmd/kafka-controller/main.go
index 785b07a40..5d1ce3585 100644
--- a/control-plane/cmd/kafka-controller/main.go
+++ b/control-plane/cmd/kafka-controller/main.go
@@ -57,7 +57,7 @@ func main() {
 
 		// Broker controller
 		func(ctx context.Context, watcher configmap.Watcher) *controller.Impl {
-			return broker.NewController(ctx, watcher, &broker.Configs{Env: *brokerEnv})
+			return broker.NewController(ctx, watcher, brokerEnv, "")
 		},
 
 		// Trigger controller
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index a0c796b3e..0def56f54 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -74,7 +74,8 @@ type StatusConditionManager struct {
 
 	SetAddress func(u *apis.URL)
 
-	Configs *config.Env
+	Env              *config.Env
+	BootstrapServers string
 
 	Recorder record.EventRecorder
 }
@@ -100,13 +101,13 @@ func (manager *StatusConditionManager) FailedToGetConfigMap(err error) reconcile
 		ConditionConfigMapUpdated,
 		fmt.Sprintf(
 			"Failed to get ConfigMap: %s",
-			manager.Configs.DataPlaneConfigMapAsString(),
+			manager.Env.DataPlaneConfigMapAsString(),
 		),
 		"%v",
 		err,
 	)
 
-	return fmt.Errorf("failed to get contract config map %s: %w", manager.Configs.DataPlaneConfigMapAsString(), err)
+	return fmt.Errorf("failed to get contract config map %s: %w", manager.Env.DataPlaneConfigMapAsString(), err)
 }
 
 func (manager *StatusConditionManager) FailedToGetDataFromConfigMap(err error) reconciler.Event {
@@ -115,32 +116,32 @@ func (manager *StatusConditionManager) FailedToGetDataFromConfigMap(err error) r
 		ConditionConfigMapUpdated,
 		fmt.Sprintf(
 			"Failed to get contract data from ConfigMap: %s",
-			manager.Configs.DataPlaneConfigMapAsString(),
+			manager.Env.DataPlaneConfigMapAsString(),
 		),
 		"%v",
 		err,
 	)
 
-	return fmt.Errorf("failed to get broker and triggers data from config map %s: %w", manager.Configs.DataPlaneConfigMapAsString(), err)
+	return fmt.Errorf("failed to get broker and triggers data from config map %s: %w", manager.Env.DataPlaneConfigMapAsString(), err)
 }
 
 func (manager *StatusConditionManager) FailedToUpdateConfigMap(err error) reconciler.Event {
 
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
 		ConditionConfigMapUpdated,
-		fmt.Sprintf("Failed to update ConfigMap: %s", manager.Configs.DataPlaneConfigMapAsString()),
+		fmt.Sprintf("Failed to update ConfigMap: %s", manager.Env.DataPlaneConfigMapAsString()),
 		"%s",
 		err,
 	)
 
-	return fmt.Errorf("failed to update contract config map %s: %w", manager.Configs.DataPlaneConfigMapAsString(), err)
+	return fmt.Errorf("failed to update contract config map %s: %w", manager.Env.DataPlaneConfigMapAsString(), err)
 }
 
 func (manager *StatusConditionManager) ConfigMapUpdated() {
 
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrueWithReason(
 		ConditionConfigMapUpdated,
-		fmt.Sprintf("Config map %s updated", manager.Configs.DataPlaneConfigMapAsString()),
+		fmt.Sprintf("Config map %s updated", manager.Env.DataPlaneConfigMapAsString()),
 		"",
 	)
 }
@@ -183,7 +184,7 @@ func (manager *StatusConditionManager) Reconciled() reconciler.Event {
 
 		manager.SetAddress(&apis.URL{
 			Scheme: "http",
-			Host:   network.GetServiceHostname(manager.Configs.IngressName, manager.Configs.SystemNamespace),
+			Host:   network.GetServiceHostname(manager.Env.IngressName, manager.Env.SystemNamespace),
 			Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
 		})
 		object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 7eb605cbc..3cbe2842d 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -48,14 +48,9 @@ const (
 	TopicPrefix = "knative-broker-"
 )
 
-type Configs struct {
-	config.Env
-
-	BootstrapServers string
-}
-
 type Reconciler struct {
 	*base.Reconciler
+	*config.Env
 
 	Resolver *resolver.URIResolver
 
@@ -69,7 +64,7 @@ type Reconciler struct {
 	// mock the function used during the reconciliation loop.
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
 
-	Configs *Configs
+	BootstrapServers string
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, broker *eventing.Broker) reconciler.Event {
@@ -84,7 +79,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	statusConditionManager := base.StatusConditionManager{
 		Object:     broker,
 		SetAddress: broker.Status.SetAddress,
-		Configs:    &r.Configs.Env,
+		Env:        r.Env,
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
@@ -241,7 +236,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get contract config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
+		return fmt.Errorf("failed to get contract config map %s: %w", r.DataPlaneConfigMapAsString(), err)
 	}
 
 	logger.Debug("Got contract config map")
@@ -400,7 +395,7 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 		}
 	}
 
-	egressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, broker, broker.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	egressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, broker, broker.Spec.Delivery, r.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 8b23ebc89..a292d5898 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -53,7 +53,7 @@ const (
 	DefaultReplicationFactor = 1
 )
 
-func NewController(ctx context.Context, watcher configmap.Watcher, configs *Configs) *controller.Impl {
+func NewController(ctx context.Context, watcher configmap.Watcher, env *config.Env, bootstrapServers string) *controller.Impl {
 
 	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
@@ -64,10 +64,10 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			KubeClient:                  kubeclient.Get(ctx),
 			PodLister:                   podinformer.Get(ctx).Lister(),
 			SecretLister:                secretinformer.Get(ctx).Lister(),
-			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-			SystemNamespace:             configs.SystemNamespace,
+			DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+			DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+			DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+			SystemNamespace:             env.SystemNamespace,
 			DispatcherLabel:             base.BrokerDispatcherLabel,
 			ReceiverLabel:               base.BrokerReceiverLabel,
 		},
@@ -77,7 +77,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 			ReplicationFactor: DefaultReplicationFactor,
 		},
 		ConfigMapLister: configmapInformer.Lister(),
-		Configs:         configs,
+		Env:             env,
 	}
 
 	logger := logging.FromContext(ctx)
@@ -85,13 +85,13 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 	_, err := reconciler.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
 		logger.Fatal("Failed to get or create data plane config map",
-			zap.String("configmap", configs.DataPlaneConfigMapAsString()),
+			zap.String("configmap", env.DataPlaneConfigMapAsString()),
 			zap.Error(err),
 		)
 	}
 
-	if configs.BootstrapServers != "" {
-		reconciler.SetBootstrapServers(configs.BootstrapServers)
+	if bootstrapServers != "" {
+		reconciler.SetBootstrapServers(bootstrapServers)
 	}
 
 	impl := brokerreconciler.NewImpl(ctx, reconciler, kafka.BrokerClass)
@@ -110,7 +110,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 	}
 
 	configmapInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
-		FilterFunc: controller.FilterWithNameAndNamespace(configs.DataPlaneConfigMapNamespace, configs.DataPlaneConfigMapName),
+		FilterFunc: controller.FilterWithNameAndNamespace(env.DataPlaneConfigMapNamespace, env.DataPlaneConfigMapName),
 		Handler: cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				globalResync(obj)
@@ -142,14 +142,14 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *Conf
 		},
 	})
 
-	cm, err := reconciler.KubeClient.CoreV1().ConfigMaps(configs.SystemNamespace).Get(ctx, configs.GeneralConfigMapName, metav1.GetOptions{})
+	cm, err := reconciler.KubeClient.CoreV1().ConfigMaps(env.SystemNamespace).Get(ctx, env.GeneralConfigMapName, metav1.GetOptions{})
 	if err != nil {
-		panic(fmt.Errorf("failed to get config map %s/%s: %w", configs.SystemNamespace, configs.GeneralConfigMapName, err))
+		panic(fmt.Errorf("failed to get config map %s/%s: %w", env.SystemNamespace, env.GeneralConfigMapName, err))
 	}
 
 	reconciler.ConfigMapUpdated(ctx)(cm)
 
-	watcher.Watch(configs.GeneralConfigMapName, reconciler.ConfigMapUpdated(ctx))
+	watcher.Watch(env.GeneralConfigMapName, reconciler.ConfigMapUpdated(ctx))
 
 	return impl
 }
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index 7ce51249c..589c3d838 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -57,6 +57,7 @@ const (
 
 type Reconciler struct {
 	*base.Reconciler
+	*config.Env
 
 	Resolver *resolver.URIResolver
 
@@ -69,8 +70,6 @@ type Reconciler struct {
 	NewKafkaClient kafka.NewClientFunc
 
 	ConfigMapLister corelisters.ConfigMapLister
-
-	Configs *config.Env
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, channel *messagingv1beta1.KafkaChannel) reconciler.Event {
@@ -85,7 +84,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, channel *messagingv1beta
 	statusConditionManager := base.StatusConditionManager{
 		Object:     channel,
 		SetAddress: channel.Status.SetAddress,
-		Configs:    r.Configs,
+		Env:        r.Env,
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
@@ -273,7 +272,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, channel *messagingv1beta1
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get contract config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
+		return fmt.Errorf("failed to get contract config map %s: %w", r.DataPlaneConfigMapAsString(), err)
 	}
 	logger.Debug("Got contract config map")
 
@@ -451,11 +450,11 @@ func (r *Reconciler) getSubscriberConfig(ctx context.Context, channel *messaging
 		Uid:           string(subscriber.UID),
 	}
 
-	subscriptionEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, subscriber.Delivery, r.Configs.DefaultBackoffDelayMs)
+	subscriptionEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, subscriber.Delivery, r.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, err
 	}
-	channelEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	channelEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, err
 	}
@@ -527,7 +526,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		}
 	}
 
-	egressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	egressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, channel, channel.Spec.Delivery, r.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, err
 	}
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index 739d5f943..c86cc490a 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -62,7 +62,7 @@ func NewController(ctx context.Context, watcher configmap.Watcher, configs *conf
 		},
 		NewKafkaClient:             sarama.NewClient,
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
-		Configs:                    configs,
+		Env:                        configs,
 		ConfigMapLister:            configmapInformer.Lister(),
 	}
 
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index aa4784957..cd83f8dab 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -60,7 +60,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 		},
 		ConfigMapLister:            configmapInformer.Lister(),
 		NewKafkaClusterAdminClient: sarama.NewClusterAdmin,
-		Configs:                    configs,
+		Env:                        configs,
 	}
 
 	_, err := reconciler.GetOrCreateDataPlaneConfigMap(ctx)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 26e0744e0..6ed5ac6fb 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -45,14 +45,13 @@ const (
 
 type Reconciler struct {
 	*base.Reconciler
+	*config.Env
 
 	ConfigMapLister corelisters.ConfigMapLister
 
 	// NewKafkaClusterAdminClient creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
 	// mock the function used during the reconciliation loop.
 	NewKafkaClusterAdminClient kafka.NewClusterAdminClientFunc
-
-	Configs *config.Env
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, ks *eventing.KafkaSink) reconciler.Event {
@@ -67,7 +66,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	statusConditionManager := base.StatusConditionManager{
 		Object:     ks,
 		SetAddress: ks.Status.SetAddress,
-		Configs:    r.Configs,
+		Env:        r.Env,
 		Recorder:   controller.GetEventRecorder(ctx),
 	}
 
@@ -232,7 +231,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 	// Get contract config map.
 	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get contract config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
+		return fmt.Errorf("failed to get contract config map %s: %w", r.DataPlaneConfigMapAsString(), err)
 	}
 
 	logger.Debug("Got contract config map")
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
index bf5fb73e5..d27d0238a 100644
--- a/control-plane/pkg/reconciler/source/source.go
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -81,7 +81,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *sources.KafkaSource)
 
 	statusConditionManager := base.StatusConditionManager{
 		Object:   ks,
-		Configs:  r.Env,
+		Env:      r.Env,
 		Recorder: controller.GetEventRecorder(ctx),
 	}
 
diff --git a/control-plane/pkg/reconciler/trigger/controller.go b/control-plane/pkg/reconciler/trigger/controller.go
index 425253dee..8f4bffe99 100644
--- a/control-plane/pkg/reconciler/trigger/controller.go
+++ b/control-plane/pkg/reconciler/trigger/controller.go
@@ -73,7 +73,7 @@ func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env
 		},
 		BrokerLister:   brokerInformer.Lister(),
 		EventingClient: eventingclient.Get(ctx),
-		Configs:        configs,
+		Env:            configs,
 	}
 
 	impl := triggerreconciler.NewImpl(ctx, reconciler, func(impl *controller.Impl) controller.Options {
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index d2f91fa2e..27f53f742 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -54,7 +54,7 @@ type Reconciler struct {
 	EventingClient eventingclientset.Interface
 	Resolver       *resolver.URIResolver
 
-	Configs *config.Env
+	Env *config.Env
 }
 
 func (r *Reconciler) ReconcileKind(ctx context.Context, trigger *eventing.Trigger) reconciler.Event {
@@ -68,7 +68,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 
 	statusConditionManager := statusConditionManager{
 		Trigger:  trigger,
-		Configs:  r.Configs,
+		Configs:  r.Env,
 		Recorder: controller.GetEventRecorder(ctx),
 	}
 
@@ -201,7 +201,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, trigger *eventing.Trigger
 	// Get data plane config map.
 	dataPlaneConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get data plane config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
+		return fmt.Errorf("failed to get data plane config map %s: %w", r.Env.DataPlaneConfigMapAsString(), err)
 	}
 
 	logger.Debug("Got data plane config map")
@@ -248,7 +248,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, trigger *eventing.Trigger
 		return err
 	}
 
-	logger.Debug("Updated data plane config map", zap.String("configmap", r.Configs.DataPlaneConfigMapAsString()))
+	logger.Debug("Updated data plane config map", zap.String("configmap", r.Env.DataPlaneConfigMapAsString()))
 
 	// Update volume generation annotation of dispatcher pods
 	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
@@ -285,11 +285,11 @@ func (r *Reconciler) reconcileTriggerEgress(ctx context.Context, broker *eventin
 		}
 	}
 
-	triggerEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, trigger, trigger.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	triggerEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, trigger, trigger.Spec.Delivery, r.Env.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, fmt.Errorf("[trigger] %w", err)
 	}
-	brokerEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, broker, broker.Spec.Delivery, r.Configs.DefaultBackoffDelayMs)
+	brokerEgressConfig, err := coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, broker, broker.Spec.Delivery, r.Env.DefaultBackoffDelayMs)
 	if err != nil {
 		return nil, fmt.Errorf("[broker] %w", err)
 	}


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index b913ddc82..cf19f8182 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
@@ -90,28 +91,28 @@ func TestBrokerReconciler(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		brokerReconciliation(t, f, *DefaultConfigs)
+		brokerReconciliation(t, f, *DefaultEnv)
 	}
 }
 
-func brokerReconciliation(t *testing.T, format string, configs Configs) {
+func brokerReconciliation(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
 			Name: "Reconciled normal - no DLS",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -121,7 +122,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -132,11 +133,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -148,11 +149,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				{
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -168,17 +169,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -190,10 +191,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -205,11 +206,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -226,17 +227,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(WithServiceNamespace(BrokerNamespace)),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -248,10 +249,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -263,11 +264,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						WithDelivery(WithNoDeadLetterSinkNamespace),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURLFrom(BrokerNamespace, ServiceName)),
 					),
 				},
@@ -280,8 +281,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			Name: "Failed to create topic",
 			Objects: []runtime.Object{
 				NewBroker(),
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -319,12 +320,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					WithDelivery(),
 				),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 				&corev1.ConfigMap{
 					ObjectMeta: metav1.ObjectMeta{
-						Namespace: configs.DataPlaneConfigMapNamespace,
-						Name:      configs.DataPlaneConfigMapName + "a", // Use a different name
+						Namespace: env.DataPlaneConfigMapNamespace,
+						Name:      env.DataPlaneConfigMapName + "a", // Use a different name
 					},
 				},
 			},
@@ -334,10 +335,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -349,10 +350,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -366,9 +367,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerConfigParsed,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -381,10 +382,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 			Name: "Reconciled normal - config map not readable",
 			Objects: []runtime.Object{
 				NewBroker(),
-				NewConfigMap(&configs, []byte(`{"hello": "world"}`)),
+				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -393,7 +394,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -404,10 +405,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -419,10 +420,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -448,12 +449,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -462,7 +463,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -484,10 +485,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -499,10 +500,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -539,12 +540,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
 			},
@@ -553,7 +554,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -570,10 +571,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -595,10 +596,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved("http://www.my-sink.com/api"),
 					),
 				},
@@ -630,12 +631,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
 			},
@@ -644,7 +645,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -661,10 +662,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -679,10 +680,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -709,12 +710,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
 			},
@@ -723,7 +724,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -740,10 +741,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -755,10 +756,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -792,11 +793,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
 			},
@@ -840,9 +841,9 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMapFromContract(&contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 1,
-				}, &configs),
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -876,13 +877,13 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					),
 				),
 				BrokerConfig(bootstrapServers, 20, 5),
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "3",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -892,7 +893,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -903,11 +904,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -922,11 +923,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							KReference(BrokerConfig(bootstrapServers, 20, 5)),
 						),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -948,12 +949,12 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewSSLSecret(ConfigMapNamespace, "secret-1"),
 				BrokerConfig(bootstrapServers, 20, 5, BrokerAuthConfig("secret-1")),
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
@@ -962,7 +963,7 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -981,11 +982,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -1000,11 +1001,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 							BrokerAuthConfig("secret-1"),
 						))),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 					),
 				},
 			},
@@ -1024,8 +1025,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						KReference(BrokerConfig(bootstrapServers, 20, 5)),
 					),
 				),
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -1077,8 +1078,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						Name:      BrokerName,
 					},
 				},
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -1151,17 +1152,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, nil),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				BrokerReceiverPod(env.SystemNamespace, nil),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -1196,10 +1197,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -1212,10 +1213,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
 						BrokerDataPlaneAvailable,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerConfigParsed,
 						BrokerTopicReady,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1260,17 +1261,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1287,10 +1288,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -1303,11 +1304,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						WithRetry(pointer.Int32Ptr(10), &exponential, pointer.StringPtr("PT2S")),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1325,17 +1326,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1352,10 +1353,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -1368,11 +1369,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						WithRetry(pointer.Int32Ptr(10), &linear, pointer.StringPtr("PT2S")),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1390,17 +1391,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1414,10 +1415,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -1430,11 +1431,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						WithRetry(nil, &linear, pointer.StringPtr("PT2S")),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1452,17 +1453,17 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              BrokerUUID,
@@ -1473,16 +1474,16 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 								DeadLetter:    ServiceURL,
 								Retry:         10,
 								BackoffPolicy: contract.BackoffPolicy_Linear,
-								BackoffDelay:  configs.DefaultBackoffDelayMs,
+								BackoffDelay:  env.DefaultBackoffDelayMs,
 							},
 						},
 					},
 					Generation: 2,
 				}),
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -1495,11 +1496,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						WithDelivery(),
 						WithRetry(pointer.Int32Ptr(10), &linear, nil),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1525,10 +1526,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1543,11 +1544,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1573,18 +1574,18 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+				BrokerReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+				BrokerDispatcherPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerReceiverPodUpdate(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1594,11 +1595,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Object: NewBroker(
 						WithDelivery(),
 						reconcilertesting.WithInitBrokerConditions,
-						BrokerConfigMapUpdatedReady(&configs),
+						BrokerConfigMapUpdatedReady(&env),
 						BrokerDataPlaneAvailable,
 						BrokerTopicReady,
 						BrokerConfigParsed,
-						BrokerAddressable(&configs),
+						BrokerAddressable(&env),
 						BrokerDLSResolved(ServiceURL),
 					),
 				},
@@ -1613,22 +1614,22 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 		table[i].Name = table[i].Name + " - " + format
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
 func TestBrokerFinalizer(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		brokerFinalization(t, f, *DefaultConfigs)
+		brokerFinalization(t, f, *DefaultEnv)
 	}
 }
 
-func brokerFinalization(t *testing.T, format string, configs Configs) {
+func brokerFinalization(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", BrokerNamespace, BrokerName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
@@ -1644,11 +1645,11 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources:  []*contract.Resource{},
 					Generation: 2,
 				}),
@@ -1673,11 +1674,11 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 2,
 				}),
 			},
@@ -1698,7 +1699,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -1711,7 +1712,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 2,
 				}),
 			},
@@ -1730,7 +1731,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			OtherTestData: map[string]interface{}{
@@ -1755,11 +1756,11 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 5,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -1792,11 +1793,11 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 5,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -1825,7 +1826,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 						},
 					},
 					Generation: 5,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			OtherTestData: map[string]interface{}{
@@ -1838,14 +1839,14 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 		table[i].Name = table[i].Name + " - " + format
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
-func useTable(t *testing.T, table TableTest, configs *Configs) {
+func useTable(t *testing.T, table TableTest, env *config.Env) {
 
 	testCtx, cancel := context.WithCancel(context.Background())
 
-	table.Test(t, NewFactory(configs, func(ctx context.Context, listers *Listers, configs *Configs, row *TableRow) controller.Reconciler {
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		defaultTopicDetail := sarama.TopicDetail{
 			NumPartitions:     DefaultNumPartitions,
@@ -1877,10 +1878,10 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 				KubeClient:                  kubeclient.Get(ctx),
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
-				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-				SystemNamespace:             configs.SystemNamespace,
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
 			},
@@ -1896,7 +1897,7 @@ func useTable(t *testing.T, table TableTest, configs *Configs) {
 					T:                   t,
 				}, nil
 			},
-			Configs: configs,
+			Env: env,
 		}
 		reconciler.SetBootstrapServers(bootstrapServers)
 
diff --git a/control-plane/pkg/reconciler/broker/controller_test.go b/control-plane/pkg/reconciler/broker/controller_test.go
index ae0caa44e..5a259625c 100644
--- a/control-plane/pkg/reconciler/broker/controller_test.go
+++ b/control-plane/pkg/reconciler/broker/controller_test.go
@@ -40,19 +40,17 @@ import (
 func TestNewController(t *testing.T) {
 	ctx, _ := reconcilertesting.SetupFakeContext(t)
 
-	configs := &Configs{
-		Env: config.Env{
-			SystemNamespace:      "cm",
-			GeneralConfigMapName: "cm",
-		},
+	env := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
 	}
 
 	ctx, _ = fakekubeclient.With(
 		ctx,
 		&corev1.ConfigMap{
 			ObjectMeta: metav1.ObjectMeta{
-				Name:      configs.Env.GeneralConfigMapName,
-				Namespace: configs.Env.SystemNamespace,
+				Name:      env.GeneralConfigMapName,
+				Namespace: env.SystemNamespace,
 			},
 		},
 	)
@@ -68,7 +66,8 @@ func TestNewController(t *testing.T) {
 				Name: "cm",
 			},
 		}),
-		configs,
+		env,
+		"",
 	)
 	if controller == nil {
 		t.Error("failed to create controller: <nil>")
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 787d352df..8f8422cd9 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -22,6 +22,7 @@ import (
 	"io"
 	"testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/Shopify/sarama"
@@ -41,7 +42,6 @@ import (
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
@@ -83,15 +83,15 @@ func TestSinkReconciler(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		sinkReconciliation(t, f, *DefaultConfigs)
+		sinkReconciliation(t, f, *DefaultEnv)
 	}
 }
 
-func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
+func sinkReconciliation(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", SinkNamespace, SinkName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
@@ -100,8 +100,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				NewSink(
 					SinkControllerOwnsTopic,
 				),
-				NewConfigMap(&configs, nil),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				NewConfigMap(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
@@ -110,7 +110,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -121,7 +121,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -137,10 +137,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -153,8 +153,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					SinkAuthSecretRef("secret-1"),
 				),
 				NewSSLSecret(SinkNamespace, "secret-1"),
-				NewConfigMap(&configs, nil),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				NewConfigMap(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -164,7 +164,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -183,7 +183,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -200,10 +200,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -218,8 +218,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						sink.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&configs, nil),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				NewConfigMap(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -229,7 +229,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -245,7 +245,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -265,10 +265,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ExternalTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -283,8 +283,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						sink.Spec.NumPartitions = nil
 					},
 				),
-				NewConfigMap(&configs, nil),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				NewConfigMap(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -331,8 +331,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						sink.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 					},
 				),
-				NewConfigMap(&configs, nil),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				NewConfigMap(&env, nil),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -342,7 +342,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -353,7 +353,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
@@ -374,8 +374,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkConfigParsed,
 						BootstrapServers([]string{"kafka-broker:10000"}),
 						SinkTopicReadyWithOwner("my-topic-1", sink.ControllerTopicOwner),
-						SinkConfigMapUpdatedReady(&configs.Env),
-						SinkAddressable(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -390,7 +390,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
-				SinkReceiverPod(configs.SystemNamespace, nil),
+				SinkReceiverPod(env.SystemNamespace, nil),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -429,11 +429,11 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					BootstrapServers(bootstrapServersArr),
 				),
 				NewService(),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 				&corev1.ConfigMap{
 					ObjectMeta: metav1.ObjectMeta{
-						Namespace: configs.DataPlaneConfigMapNamespace,
-						Name:      configs.DataPlaneConfigMapName + "a", // Use a different name
+						Namespace: env.DataPlaneConfigMapNamespace,
+						Name:      env.DataPlaneConfigMapName + "a", // Use a different name
 					},
 				},
 			},
@@ -443,10 +443,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -457,7 +457,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -472,10 +472,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -487,15 +487,15 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
-				NewConfigMap(&configs, []byte(`{"hello": "world"}`)),
-				SinkReceiverPod(configs.SystemNamespace, nil),
+				NewConfigMap(&env, []byte(`{"hello": "world"}`)),
+				SinkReceiverPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:              SinkUUID,
@@ -506,7 +506,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -521,10 +521,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -548,8 +548,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 							Topics: []string{"my-existing-topic-b"},
 						},
 					},
-				}, &configs),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
 			},
@@ -558,7 +558,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -578,7 +578,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -593,10 +593,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -622,8 +622,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED},
 						},
 					},
-				}, &configs),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "5",
 				}),
 			},
@@ -632,7 +632,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
@@ -648,7 +648,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -663,9 +663,9 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -723,8 +723,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -743,10 +743,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -777,8 +777,8 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
-				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				SinkReceiverPod(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "0",
 				}),
 			},
@@ -787,7 +787,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+				SinkReceiverPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -802,10 +802,10 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 						SinkDataPlaneAvailable,
 						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
-						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkConfigMapUpdatedReady(&env),
 						SinkTopicReady,
 						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
-						SinkAddressable(&configs.Env),
+						SinkAddressable(&env),
 					),
 				},
 			},
@@ -816,7 +816,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 		table[i].Name = table[i].Name + " - " + format
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
 func TestSinkFinalizer(t *testing.T) {
@@ -826,15 +826,15 @@ func TestSinkFinalizer(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		sinkFinalization(t, f, *DefaultConfigs)
+		sinkFinalization(t, f, *DefaultEnv)
 	}
 }
 
-func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
+func sinkFinalization(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", SinkNamespace, SinkName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
@@ -870,11 +870,11 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -924,11 +924,11 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -981,7 +981,7 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 1,
-				}, &configs),
+				}, &env),
 			},
 			Key:     testKey,
 			WantErr: true,
@@ -994,7 +994,7 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 				),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    SinkUUID + "a",
@@ -1022,12 +1022,12 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 		table[i].Name = table[i].Name + " - " + format
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
-func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
+func useTable(t *testing.T, table TableTest, env *config.Env) {
 
-	table.Test(t, NewFactory(configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		expectedTopicName := fmt.Sprintf("%s%s-%s", TopicPrefix, SinkNamespace, SinkName)
 		if want, ok := row.OtherTestData[wantTopicName]; ok {
@@ -1084,10 +1084,10 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 				KubeClient:                  kubeclient.Get(ctx),
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
-				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-				SystemNamespace:             configs.SystemNamespace,
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
 				ReceiverLabel:               base.SinkReceiverLabel,
 			},
 			ConfigMapLister: listers.GetConfigMapLister(),
@@ -1103,7 +1103,7 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 					T:                                      t,
 				}, nil
 			},
-			Configs: &configs.Env,
+			Env: env,
 		}
 
 		reconciler.SecretTracker = &FakeTracker{}
diff --git a/control-plane/pkg/reconciler/source/source_test.go b/control-plane/pkg/reconciler/source/source_test.go
index 93c19d2cf..aa628247e 100644
--- a/control-plane/pkg/reconciler/source/source_test.go
+++ b/control-plane/pkg/reconciler/source/source_test.go
@@ -27,6 +27,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
@@ -42,7 +43,6 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/source"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
@@ -64,7 +64,7 @@ func TestReconcileKind(t *testing.T) {
 
 	sources.RegisterAlternateKafkaConditionSet(base.EgressConditionSet)
 
-	configs := *DefaultConfigs
+	env := *DefaultEnv
 	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
 
 	table := TableTest{
@@ -73,13 +73,13 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSourceSinkObject(),
 				NewSource(),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -101,19 +101,19 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -134,13 +134,13 @@ func TestReconcileKind(t *testing.T) {
 				NewSource(WithCloudEventOverrides(&duckv1.CloudEventOverrides{
 					Extensions: map[string]string{"a": "foo", "b": "foo"},
 				})),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -165,14 +165,14 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
@@ -180,7 +180,7 @@ func TestReconcileKind(t *testing.T) {
 						WithCloudEventOverrides(&duckv1.CloudEventOverrides{
 							Extensions: map[string]string{"a": "foo", "b": "foo"},
 						}),
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -199,13 +199,13 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSourceSinkObject(),
 				NewSource(WithKeyType("string")),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -228,20 +228,20 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
 						WithKeyType("string"),
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -260,13 +260,13 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSourceSinkObject(),
 				NewSource(WithKeyType("int")),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -289,20 +289,20 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
 						WithKeyType("int"),
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -321,13 +321,13 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSourceSinkObject(),
 				NewSource(WithKeyType("byte-array")),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -350,20 +350,20 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
 						WithKeyType("byte-array"),
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -382,13 +382,13 @@ func TestReconcileKind(t *testing.T) {
 			Objects: []runtime.Object{
 				NewSourceSinkObject(),
 				NewSource(WithKeyType("float")),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 1,
 					Resources: []*contract.Resource{
 						{
@@ -411,20 +411,20 @@ func TestReconcileKind(t *testing.T) {
 						},
 					},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSource(
 						WithKeyType("float"),
-						SourceConfigMapUpdatedReady(&configs),
+						SourceConfigMapUpdatedReady(&env),
 						SourceTopicsReady,
 						SourceDataPlaneAvailable,
 						InitialOffsetsCommitted,
@@ -440,11 +440,11 @@ func TestReconcileKind(t *testing.T) {
 		},
 	}
 
-	useTable(t, table, configs)
+	useTable(t, table, env)
 }
 
-func useTable(t *testing.T, table TableTest, configs broker.Configs) {
-	table.Test(t, NewFactory(&configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
+func useTable(t *testing.T, table TableTest, env config.Env) {
+	table.Test(t, NewFactory(&env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		var topicMetadata []*sarama.TopicMetadata
 		for _, t := range SourceTopics {
@@ -456,13 +456,13 @@ func useTable(t *testing.T, table TableTest, configs broker.Configs) {
 				KubeClient:                  kubeclient.Get(ctx),
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
-				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-				SystemNamespace:             configs.SystemNamespace,
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
 				DispatcherLabel:             base.SourceDispatcherLabel,
 			},
-			Env: &configs.Env,
+			Env: env,
 			InitOffsetsFunc: func(ctx context.Context, kafkaClient sarama.Client, kafkaAdminClient sarama.ClusterAdmin, topics []string, consumerGroup string) (int32, error) {
 				return 1, nil
 			},
@@ -506,7 +506,7 @@ func TestFinalizeKind(t *testing.T) {
 
 	sources.RegisterAlternateKafkaConditionSet(base.EgressConditionSet)
 
-	configs := *DefaultConfigs
+	env := *DefaultEnv
 	testKey := fmt.Sprintf("%s/%s", SourceNamespace, SourceName)
 
 	table := TableTest{
@@ -523,18 +523,18 @@ func TestFinalizeKind(t *testing.T) {
 							BootstrapServers: SourceBootstrapServers,
 						},
 					},
-				}, &configs),
-				SourceDispatcherPod(configs.SystemNamespace, map[string]string{
+				}, &env),
+				SourceDispatcherPod(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Generation: 2,
 					Resources:  []*contract.Resource{},
 				}),
-				SourceDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				SourceDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					"annotation_to_preserve":           "value_to_preserve",
 					base.VolumeGenerationAnnotationKey: "2",
 				}),
@@ -543,7 +543,7 @@ func TestFinalizeKind(t *testing.T) {
 		},
 	}
 
-	useTable(t, table, configs)
+	useTable(t, table, env)
 }
 
 func SourceDispatcherPodUpdate(namespace string, annotations map[string]string) clientgotesting.UpdateActionImpl {
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 4339757b6..426a78c00 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -40,7 +40,6 @@ import (
 	fakeeventingkafkabrokerclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
 
 const (
@@ -49,24 +48,19 @@ const (
 	recorderBufferSize = 20
 )
 
-var DefaultConfigs = &broker.Configs{
-
-	Env: config.Env{
-		DataPlaneConfigMapNamespace: "knative-eventing",
-		DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
-		IngressName:                 "kafka-broker-receiver",
-		SystemNamespace:             "knative-eventing",
-		DataPlaneConfigFormat:       base.Json,
-		DefaultBackoffDelayMs:       1000,
-	},
-
-	BootstrapServers: "",
+var DefaultEnv = &config.Env{
+	DataPlaneConfigMapNamespace: "knative-eventing",
+	DataPlaneConfigMapName:      "kafka-broker-brokers-triggers",
+	IngressName:                 "kafka-broker-receiver",
+	SystemNamespace:             "knative-eventing",
+	DataPlaneConfigFormat:       base.Json,
+	DefaultBackoffDelayMs:       1000,
 }
 
 // Ctor functions create a k8s controller with given params.
-type Ctor func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) pkgcontroller.Reconciler
+type Ctor func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) pkgcontroller.Reconciler
 
-func NewFactory(configs *broker.Configs, ctor Ctor) Factory {
+func NewFactory(env *config.Env, ctor Ctor) Factory {
 	return func(t *testing.T, row *TableRow) (pkgcontroller.Reconciler, ActionRecorderList, EventList) {
 
 		listers := newListers(row.Objects)
@@ -99,7 +93,7 @@ func NewFactory(configs *broker.Configs, ctor Ctor) Factory {
 
 		ctx = addressable.WithDuck(ctx)
 
-		controller := ctor(ctx, listers, configs, row)
+		controller := ctor(ctx, listers, env, row)
 
 		if la, ok := controller.(reconciler.LeaderAware); ok {
 			_ = la.Promote(reconciler.UniversalBucket(), func(reconciler.Bucket, types.NamespacedName) {})
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index e7624bc4d..a8293289c 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgotesting "k8s.io/client-go/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
@@ -168,11 +169,11 @@ func BrokerReady(broker *eventing.Broker) {
 	}
 }
 
-func BrokerConfigMapUpdatedReady(configs *Configs) func(broker *eventing.Broker) {
+func BrokerConfigMapUpdatedReady(env *config.Env) func(broker *eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.GetConditionSet().Manage(broker.GetStatus()).MarkTrueWithReason(
 			base.ConditionConfigMapUpdated,
-			fmt.Sprintf("Config map %s updated", configs.DataPlaneConfigMapAsString()),
+			fmt.Sprintf("Config map %s updated", env.DataPlaneConfigMapAsString()),
 			"",
 		)
 	}
@@ -208,13 +209,13 @@ func BrokerConfigNotParsed(reason string) func(broker *eventing.Broker) {
 	}
 }
 
-func BrokerAddressable(configs *Configs) func(broker *eventing.Broker) {
+func BrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
 
 	return func(broker *eventing.Broker) {
 
 		broker.Status.Address.URL = &apis.URL{
 			Scheme: "http",
-			Host:   network.GetServiceHostname(configs.IngressName, configs.SystemNamespace),
+			Host:   network.GetServiceHostname(env.IngressName, env.SystemNamespace),
 			Path:   fmt.Sprintf("/%s/%s", broker.Namespace, broker.Name),
 		}
 
@@ -239,7 +240,7 @@ func BrokerFailedToCreateTopic(broker *eventing.Broker) {
 
 }
 
-func BrokerFailedToGetConfigMap(configs *Configs) func(broker *eventing.Broker) {
+func BrokerFailedToGetConfigMap(env *config.Env) func(broker *eventing.Broker) {
 
 	return func(broker *eventing.Broker) {
 
@@ -247,7 +248,7 @@ func BrokerFailedToGetConfigMap(configs *Configs) func(broker *eventing.Broker)
 			base.ConditionConfigMapUpdated,
 			fmt.Sprintf(
 				"Failed to get ConfigMap: %s",
-				configs.DataPlaneConfigMapAsString(),
+				env.DataPlaneConfigMapAsString(),
 			),
 			`configmaps "knative-eventing" not found`,
 		)
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 5268ca1dd..c29e30f2e 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io/ioutil"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
 
@@ -33,7 +34,6 @@ import (
 	reconcilertesting "knative.dev/eventing/pkg/reconciler/testing/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
 
 const (
@@ -82,10 +82,10 @@ func ServiceURLFrom(ns, name string) string {
 	return fmt.Sprintf("http://%s.%s.svc.cluster.local", name, ns)
 }
 
-func NewConfigMap(configs *Configs, data []byte) runtime.Object {
+func NewConfigMap(env *config.Env, data []byte) runtime.Object {
 	return reconcilertesting.NewConfigMap(
-		configs.DataPlaneConfigMapName,
-		configs.DataPlaneConfigMapNamespace,
+		env.DataPlaneConfigMapName,
+		env.DataPlaneConfigMapNamespace,
 		func(configMap *corev1.ConfigMap) {
 			if configMap.BinaryData == nil {
 				configMap.BinaryData = make(map[string][]byte, 1)
@@ -98,10 +98,10 @@ func NewConfigMap(configs *Configs, data []byte) runtime.Object {
 	)
 }
 
-func NewConfigMapFromContract(contract *contract.Contract, configs *Configs) runtime.Object {
+func NewConfigMapFromContract(contract *contract.Contract, env *config.Env) runtime.Object {
 	var data []byte
 	var err error
-	if configs.DataPlaneConfigFormat == base.Protobuf {
+	if env.DataPlaneConfigFormat == base.Protobuf {
 		data, err = proto.Marshal(contract)
 	} else {
 		data, err = protojson.Marshal(contract)
@@ -110,18 +110,18 @@ func NewConfigMapFromContract(contract *contract.Contract, configs *Configs) run
 		panic(err)
 	}
 
-	return NewConfigMap(configs, data)
+	return NewConfigMap(env, data)
 }
 
-func ConfigMapUpdate(configs *Configs, contract *contract.Contract) clientgotesting.UpdateActionImpl {
+func ConfigMapUpdate(env *config.Env, contract *contract.Contract) clientgotesting.UpdateActionImpl {
 	return clientgotesting.NewUpdateAction(
 		schema.GroupVersionResource{
 			Group:    "*",
 			Version:  "v1",
 			Resource: "ConfigMap",
 		},
-		configs.DataPlaneConfigMapNamespace,
-		NewConfigMapFromContract(contract, configs),
+		env.DataPlaneConfigMapNamespace,
+		NewConfigMapFromContract(contract, env),
 	)
 }
 
diff --git a/control-plane/pkg/reconciler/testing/objects_source.go b/control-plane/pkg/reconciler/testing/objects_source.go
index 41a3e0b33..d80a00a4d 100644
--- a/control-plane/pkg/reconciler/testing/objects_source.go
+++ b/control-plane/pkg/reconciler/testing/objects_source.go
@@ -24,12 +24,12 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	"knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 )
 
 const (
@@ -113,11 +113,11 @@ func NewSourceSinkReference() duckv1.Destination {
 	}
 }
 
-func SourceConfigMapUpdatedReady(configs *Configs) func(source *sources.KafkaSource) {
+func SourceConfigMapUpdatedReady(env *config.Env) func(source *sources.KafkaSource) {
 	return func(source *sources.KafkaSource) {
 		source.GetConditionSet().Manage(source.GetStatus()).MarkTrueWithReason(
 			base.ConditionConfigMapUpdated,
-			fmt.Sprintf("Config map %s updated", configs.DataPlaneConfigMapAsString()),
+			fmt.Sprintf("Config map %s updated", env.DataPlaneConfigMapAsString()),
 			"",
 		)
 	}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 2669617b1..5301d2d74 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -24,6 +24,7 @@ import (
 
 	"google.golang.org/protobuf/testing/protocmp"
 	"k8s.io/utils/pointer"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/tracker"
@@ -49,7 +50,6 @@ import (
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
@@ -82,15 +82,15 @@ func TestTriggerReconciler(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		triggerReconciliation(t, f, *DefaultConfigs)
+		triggerReconciliation(t, f, *DefaultEnv)
 	}
 }
 
-func triggerReconciliation(t *testing.T, format string, configs broker.Configs) {
+func triggerReconciliation(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
@@ -109,8 +109,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -120,7 +120,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -137,7 +137,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -172,8 +172,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -183,7 +183,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -201,7 +201,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -235,8 +235,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -246,7 +246,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -270,7 +270,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -305,8 +305,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -316,7 +316,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -334,7 +334,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -369,8 +369,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -380,7 +380,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -398,7 +398,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -433,8 +433,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -444,7 +444,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -468,7 +468,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -535,8 +535,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 						},
 					},
 					Generation: 2,
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -546,7 +546,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "z",
@@ -583,7 +583,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 3,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "3",
 				}),
 			},
@@ -617,7 +617,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 			WantErr:                 true,
 			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
 			WantCreates: []runtime.Object{
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantEvents: []string{
 				finalizerUpdatedEvent,
@@ -626,7 +626,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					"InternalError",
 					fmt.Sprintf(
 						"broker not found in data plane config map %s",
-						configs.DataPlaneConfigMapAsString(),
+						env.DataPlaneConfigMapAsString(),
 					),
 				),
 			},
@@ -636,7 +636,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerFailed(
 							"Broker not found in data plane map",
-							fmt.Sprintf("config map: %s", configs.DataPlaneConfigMapAsString()),
+							fmt.Sprintf("config map: %s", env.DataPlaneConfigMapAsString()),
 						),
 					),
 				},
@@ -648,7 +648,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				NewBroker(BrokerReady),
 				newTrigger(),
 				NewService(),
-				NewConfigMap(&configs, nil),
+				NewConfigMap(&env, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -662,7 +662,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					"InternalError",
 					fmt.Sprintf(
 						"broker not found in data plane config map %s",
-						configs.DataPlaneConfigMapAsString(),
+						env.DataPlaneConfigMapAsString(),
 					),
 				),
 			},
@@ -672,7 +672,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 						reconcilertesting.WithInitTriggerConditions,
 						reconcilertesting.WithTriggerBrokerFailed(
 							"Broker not found in data plane map",
-							fmt.Sprintf("config map: %s", configs.DataPlaneConfigMapAsString()),
+							fmt.Sprintf("config map: %s", env.DataPlaneConfigMapAsString()),
 						),
 					),
 				},
@@ -730,7 +730,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				NewDeletedBroker(),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 8,
-				}, &configs),
+				}, &env),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -760,7 +760,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 						},
 					},
 					Generation: 8,
-				}, &configs),
+				}, &env),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -802,8 +802,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 						},
 					},
 					Generation: 8,
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -813,7 +813,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -829,7 +829,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 9,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "9",
 				}),
 			},
@@ -871,8 +871,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -980,8 +980,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -991,7 +991,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -1065,7 +1065,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -1162,8 +1162,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1173,7 +1173,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -1247,7 +1247,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -1381,8 +1381,8 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1392,7 +1392,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -1496,7 +1496,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -1545,7 +1545,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 		table[i].Name = table[i].Name + " - " + format
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
 func withDelivery(trigger *eventing.Trigger) {
@@ -1563,16 +1563,16 @@ func TestTriggerFinalizer(t *testing.T) {
 	t.Parallel()
 
 	for _, f := range Formats {
-		triggerFinalizer(t, f, *DefaultConfigs)
+		triggerFinalizer(t, f, *DefaultEnv)
 	}
 
 }
 
-func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
+func triggerFinalizer(t *testing.T, format string, env config.Env) {
 
 	testKey := fmt.Sprintf("%s/%s", triggerNamespace, triggerName)
 
-	configs.DataPlaneConfigFormat = format
+	env.DataPlaneConfigFormat = format
 
 	table := TableTest{
 		{
@@ -1600,8 +1600,8 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 8,
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1611,7 +1611,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 				finalizerUpdatedEvent,
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:    BrokerUUID,
@@ -1627,7 +1627,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 9,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "9",
 				}),
 			},
@@ -1652,7 +1652,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 						},
 					},
 					Generation: 8,
-				}, &configs),
+				}, &env),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1676,7 +1676,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 				NewDeletedBroker(),
 				NewConfigMapFromContract(&contract.Contract{
 					Generation: 8,
-				}, &configs),
+				}, &env),
 			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
@@ -1770,8 +1770,8 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1781,7 +1781,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID,
@@ -1847,7 +1847,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -1936,8 +1936,8 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1947,7 +1947,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -2013,7 +2013,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -2135,8 +2135,8 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 							},
 						},
 					},
-				}, &configs),
-				BrokerDispatcherPod(configs.SystemNamespace, nil),
+				}, &env),
+				BrokerDispatcherPod(env.SystemNamespace, nil),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -2146,7 +2146,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 				patchFinalizers(),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
+				ConfigMapUpdate(&env, &contract.Contract{
 					Resources: []*contract.Resource{
 						{
 							Uid:     BrokerUUID + "a",
@@ -2245,7 +2245,7 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 					},
 					Generation: 1,
 				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
+				BrokerDispatcherPodUpdate(env.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
@@ -2259,12 +2259,12 @@ func triggerFinalizer(t *testing.T, format string, configs broker.Configs) {
 		},
 	}
 
-	useTable(t, table, &configs)
+	useTable(t, table, &env)
 }
 
-func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
+func useTable(t *testing.T, table TableTest, env *config.Env) {
 
-	table.Test(t, NewFactory(configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
+	table.Test(t, NewFactory(env, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
 
 		logger := logging.FromContext(ctx)
 
@@ -2273,17 +2273,17 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 				KubeClient:                  kubeclient.Get(ctx),
 				PodLister:                   listers.GetPodLister(),
 				SecretLister:                listers.GetSecretLister(),
-				DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
-				DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
-				DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
-				SystemNamespace:             configs.SystemNamespace,
+				DataPlaneConfigMapNamespace: env.DataPlaneConfigMapNamespace,
+				DataPlaneConfigMapName:      env.DataPlaneConfigMapName,
+				DataPlaneConfigFormat:       env.DataPlaneConfigFormat,
+				SystemNamespace:             env.SystemNamespace,
 				DispatcherLabel:             base.BrokerDispatcherLabel,
 				ReceiverLabel:               base.BrokerReceiverLabel,
 			},
 			BrokerLister:   listers.GetBrokerLister(),
 			EventingClient: eventingclient.Get(ctx),
 			Resolver:       nil,
-			Configs:        &configs.Env,
+			Env:            env,
 		}
 
 		reconciler.Resolver = resolver.NewURIResolverFromTracker(ctx, tracker.New(func(name types.NamespacedName) {}, 0))