Repository: knative-sandbox/eventing-kafka-broker
Pull request #1707 (2022-01-07 14:48:14)

Support TLS for the metrics server
For data plane deployments, we need to be able to secure prometheus
endpoints by using TLS certificates.

Now, the receiver and the dispatcher accept the following env
variables:
- `METRICS_PEM_CERT_PATH`: TLS cert path
- `METRICS_PEM_KEY_PATH`: TLS key path
- `METRICS_HOST`: metrics server host

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Support TLS for the metrics server

<!--
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
Support TLS for the metrics server.

Now, the receiver and the dispatcher accept the following env
variables:
- `METRICS_PEM_CERT_PATH`: TLS cert path
- `METRICS_PEM_KEY_PATH`: TLS key path
- `METRICS_HOST`: metrics server host
```


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/1707#" title="Author self-approved">pierDipi</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
pierDipi: FYI @skonto
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1707](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (162dece) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/6899909620d1db2ab2477f7e4de8a6cd909f768a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (6899909) will **decrease** coverage by `6.86%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/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/1707?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    #1707      +/-   ##
============================================
- Coverage     72.69%   65.83%   -6.87%     
============================================
  Files           120       54      -66     
  Lines          4443     2514    -1929     
  Branches        171        0     -171     
============================================
- Hits           3230     1655    -1575     
+ Misses          948      686     -262     
+ Partials        265      173      -92     
```

| 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/1707?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ve/eventing/kafka/broker/core/metrics/Metrics.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvbWV0cmljcy9NZXRyaWNzLmphdmE=) | | |
| [...ver/impl/handler/ControlPlaneProbeRequestUtil.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL2hhbmRsZXIvQ29udHJvbFBsYW5lUHJvYmVSZXF1ZXN0VXRpbC5qYXZh) | | |
| [...fka/broker/core/security/PlaintextCredentials.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvc2VjdXJpdHkvUGxhaW50ZXh0Q3JlZGVudGlhbHMuamF2YQ==) | | |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | | |
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | |
| [...fka/broker/dispatcher/impl/filter/ExactFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvRXhhY3RGaWx0ZXIuamF2YQ==) | | |
| [...re/reconciler/impl/ResourcesReconcilerBuilder.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJCdWlsZGVyLmphdmE=) | | |
| [...tive/eventing/kafka/broker/receiver/main/Main.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL01haW4uamF2YQ==) | | |
| [...nting/kafka/broker/dispatcher/ResponseHandler.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvUmVzcG9uc2VIYW5kbGVyLmphdmE=) | | |
| ... and [58 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707/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/1707?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/1707?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [6899909...162dece](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1707?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: @pierDipi: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1479464921480564736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1479464921480564736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1479464921509924864) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`

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

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-integration-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-integration-tests/1479464921480564736) | true | `/test pull-knative-sandbox-eventing-kafka-broker-integration-tests`
pull-knative-sandbox-eventing-kafka-broker-upgrade-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-upgrade-tests/1479464921509924864) | true | `/test pull-knative-sandbox-eventing-kafka-broker-upgrade-tests`
pull-knative-sandbox-eventing-kafka-broker-reconciler-tests | 162deceb3beec7c35f84eedf5ad22640b1158f2b | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1707/pull-knative-sandbox-eventing-kafka-broker-reconciler-tests/1479464921543479296) | true | `/test pull-knative-sandbox-eventing-kafka-broker-reconciler-tests`

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

<details>

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

Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
index 8c9f8b626..c08ad5e87 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_types.go
@@ -93,6 +93,12 @@ type ConsumerGroupStatus struct {
 	// DeliveryStatus contains a resolved URL to the dead letter sink address, and any other
 	// resolved delivery options.
 	eventingduckv1.DeliveryStatus `json:",inline"`
+
+	// Replicas is the latest observed number of replicas of the given Template.
+	// These are replicas in the sense that they are instantiations of the
+	// same Template, but individual replicas also have a consistent identity.
+	// +optional
+	Replicas *int32 `json:"replicas,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 93284aba9..cc293e0de 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -19,10 +19,12 @@ package v1alpha1
 import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	bindings "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
 	eventingduck "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
 )
 
@@ -59,7 +61,9 @@ type ConsumerSpec struct {
 	// More info: https://kafka.apache.org/documentation/#consumerconfigs
 	Configs ConsumerConfigs `json:"configs,omitempty"`
 
-	// TODO Add auth
+	// Auth is the auth configuration for the Consumer.
+	// +optional
+	Auth *Auth `json:"auth,omitempty"`
 
 	// DeliverySpec contains the delivery options for event senders.
 	// +optional
@@ -71,6 +75,16 @@ type ConsumerSpec struct {
 
 	// Subscriber is the addressable that receives events that pass the Filters.
 	Subscriber duckv1.Destination `json:"subscriber"`
+
+	// CloudEventOverrides defines overrides to control the output format and
+	// modifications of the event sent to the subscriber.
+	// +optional
+	CloudEventOverrides *duckv1.CloudEventOverrides `json:"ceOverrides,omitempty"`
+}
+
+type Auth struct {
+	NetSpec  *bindings.KafkaNetSpec
+	AuthSpec *eventingv1alpha1.Auth
 }
 
 type DeliverySpec struct {
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index d2d8c9a78..60024489f 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -23,11 +23,40 @@ package v1alpha1
 
 import (
 	runtime "k8s.io/apimachinery/pkg/runtime"
-	v1 "knative.dev/eventing/pkg/apis/duck/v1"
+	eventingv1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	duckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	apis "knative.dev/pkg/apis"
+	v1 "knative.dev/pkg/apis/duck/v1"
 )
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Auth) DeepCopyInto(out *Auth) {
+	*out = *in
+	if in.NetSpec != nil {
+		in, out := &in.NetSpec, &out.NetSpec
+		*out = new(v1beta1.KafkaNetSpec)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.AuthSpec != nil {
+		in, out := &in.AuthSpec, &out.AuthSpec
+		*out = new(eventingv1alpha1.Auth)
+		(*in).DeepCopyInto(*out)
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Auth.
+func (in *Auth) DeepCopy() *Auth {
+	if in == nil {
+		return nil
+	}
+	out := new(Auth)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Consumer) DeepCopyInto(out *Consumer) {
 	*out = *in
@@ -180,6 +209,11 @@ func (in *ConsumerGroupStatus) DeepCopyInto(out *ConsumerGroupStatus) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.DeliveryStatus.DeepCopyInto(&out.DeliveryStatus)
+	if in.Replicas != nil {
+		in, out := &in.Replicas, &out.Replicas
+		*out = new(int32)
+		**out = **in
+	}
 	return
 }
 
@@ -235,6 +269,11 @@ func (in *ConsumerSpec) DeepCopyInto(out *ConsumerSpec) {
 		copy(*out, *in)
 	}
 	in.Configs.DeepCopyInto(&out.Configs)
+	if in.Auth != nil {
+		in, out := &in.Auth, &out.Auth
+		*out = new(Auth)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Delivery != nil {
 		in, out := &in.Delivery, &out.Delivery
 		*out = new(DeliverySpec)
@@ -246,6 +285,11 @@ func (in *ConsumerSpec) DeepCopyInto(out *ConsumerSpec) {
 		(*in).DeepCopyInto(*out)
 	}
 	in.Subscriber.DeepCopyInto(&out.Subscriber)
+	if in.CloudEventOverrides != nil {
+		in, out := &in.CloudEventOverrides, &out.CloudEventOverrides
+		*out = new(v1.CloudEventOverrides)
+		(*in).DeepCopyInto(*out)
+	}
 	return
 }
 
@@ -305,7 +349,7 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 	*out = *in
 	if in.DeliverySpec != nil {
 		in, out := &in.DeliverySpec, &out.DeliverySpec
-		*out = new(v1.DeliverySpec)
+		*out = new(duckv1.DeliverySpec)
 		(*in).DeepCopyInto(*out)
 	}
 	return
diff --git a/control-plane/pkg/reconciler/consumergroup/filter.go b/control-plane/pkg/reconciler/consumergroup/filter.go
index 1ce2b2a2f..0ff1b0927 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter.go
@@ -19,6 +19,8 @@ package consumergroup
 import (
 	"strings"
 
+	"k8s.io/apimachinery/pkg/types"
+
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
 
@@ -41,3 +43,20 @@ func Filter(userFacingResource string) func(obj interface{}) bool {
 		return false
 	}
 }
+
+// Enqueue enqueues using the provided enqueue function the resource associated with a ConsumerGroup
+func Enqueue(userFacingResource string, enqueue func(key types.NamespacedName)) func(obj interface{}) {
+	userFacingResource = strings.ToLower(userFacingResource)
+	return func(obj interface{}) {
+		cg, ok := obj.(*kafkainternals.ConsumerGroup)
+		if !ok {
+			return
+		}
+
+		for _, or := range cg.OwnerReferences {
+			if strings.ToLower(or.Kind) == userFacingResource {
+				enqueue(types.NamespacedName{Namespace: cg.GetNamespace(), Name: or.Name})
+			}
+		}
+	}
+}
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2.go b/control-plane/pkg/reconciler/source/v2/controllerv2.go
new file mode 100644
index 000000000..dbd161ca1
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2.go
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"context"
+
+	"k8s.io/client-go/tools/cache"
+	consumergroupclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/client"
+	consumergroupinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup"
+	"knative.dev/pkg/controller"
+
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/consumergroup"
+)
+
+func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl {
+
+	kafkaInformer := kafkainformer.Get(ctx)
+	consumerGroupInformer := consumergroupinformer.Get(ctx)
+
+	sources.RegisterAlternateKafkaConditionSet(conditionSet)
+
+	r := &ReconcilerV2{
+		ConsumerGroupLister: consumerGroupInformer.Lister(),
+		InternalsClient:     consumergroupclient.Get(ctx),
+	}
+
+	impl := kafkasource.NewImpl(ctx, r)
+
+	kafkaInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		Handler: controller.HandleAll(impl.Enqueue),
+	})
+
+	// ConsumerGroup changes and enqueue associated KafkaSource
+	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: consumergroup.Filter("kafkasource"),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("kafkasource", impl.EnqueueKey)),
+	})
+	return impl
+}
diff --git a/control-plane/pkg/reconciler/source/v2/sourcev2.go b/control-plane/pkg/reconciler/source/v2/sourcev2.go
new file mode 100644
index 000000000..9886d206f
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/sourcev2.go
@@ -0,0 +1,137 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/api/equality"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/reconciler"
+
+	internals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	internalscg "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	internalsclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/clientset/versioned"
+	internalslst "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
+)
+
+const (
+	DefaultDeliveryOrder = internals.Ordered
+
+	KafkaConditionConsumerGroup apis.ConditionType = "ConsumerGroup" //condition is registered by controller
+)
+
+var (
+	conditionSet = apis.NewLivingConditionSet(
+		KafkaConditionConsumerGroup,
+	)
+)
+
+type ReconcilerV2 struct {
+	ConsumerGroupLister internalslst.ConsumerGroupLister
+	InternalsClient     internalsclient.Interface
+}
+
+func (r *ReconcilerV2) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+
+	consgroup, err := r.reconcileConsumerGroup(ctx, ks)
+	if err != nil {
+		ks.GetConditionSet().Manage(&ks.Status).MarkFalse(KafkaConditionConsumerGroup, "failed to reconcile consumer group", err.Error())
+		return err
+	}
+	ks.GetConditionSet().Manage(&ks.Status).MarkTrue(KafkaConditionConsumerGroup)
+
+	ks.Status.MarkSink(consgroup.Status.SubscriberURI)
+	ks.Status.Placeable = consgroup.Status.Placeable
+	ks.Status.Consumers = *consgroup.Status.Replicas
+
+	return nil
+}
+
+func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, ks *sources.KafkaSource) (*internalscg.ConsumerGroup, error) {
+
+	newcg := &internalscg.ConsumerGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      string(ks.UID),
+			Namespace: ks.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				*kmeta.NewControllerRef(ks),
+			},
+		},
+		Spec: internalscg.ConsumerGroupSpec{
+			Template: internalscg.ConsumerTemplateSpec{
+				Spec: internalscg.ConsumerSpec{
+					Topics: ks.Spec.Topics,
+					Configs: internalscg.ConsumerConfigs{Configs: map[string]string{
+						"group.id":          ks.Spec.ConsumerGroup,
+						"bootstrap.servers": strings.Join(ks.Spec.BootstrapServers, ","),
+					}},
+					Auth: &internalscg.Auth{
+						NetSpec: &ks.Spec.KafkaAuthSpec.Net,
+					},
+					Delivery: &internalscg.DeliverySpec{
+						Ordering: DefaultDeliveryOrder,
+					},
+					Subscriber: ks.Spec.Sink,
+					CloudEventOverrides: &duckv1.CloudEventOverrides{
+						Extensions: ks.Spec.CloudEventOverrides.Extensions,
+					},
+				},
+			},
+			Replicas: ks.Spec.Consumers,
+		},
+	}
+
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(ks.GetNamespace()).Get(string(ks.UID)) //Get by consumer group id
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, err
+	}
+
+	if apierrors.IsNotFound(err) {
+		cg, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(newcg.GetNamespace()).Create(ctx, newcg, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			return nil, fmt.Errorf("failed to create consumer group %s/%s: %w", newcg.GetNamespace(), newcg.GetName(), err)
+		}
+		if apierrors.IsAlreadyExists(err) {
+			return newcg, nil
+		}
+		return cg, nil
+	}
+
+	if equality.Semantic.DeepDerivative(newcg.Spec, cg.Spec) {
+		return cg, nil
+	}
+
+	newCg := &internalscg.ConsumerGroup{
+		TypeMeta:   cg.TypeMeta,
+		ObjectMeta: cg.ObjectMeta,
+		Spec:       newcg.Spec,
+		Status:     cg.Status,
+	}
+	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
+	}
+
+	return cg, nil
+}
diff --git a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
index 8f0ec27df..22229f0b9 100644
--- a/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/controllerv2.go
@@ -88,7 +88,7 @@ func NewControllerV2(ctx context.Context, configs *config.Env) *controller.Impl
 	// ConsumerGroup changes and enqueue associated Trigger
 	consumerGroupInformer.Informer().AddEventHandler(cache.FilteringResourceEventHandler{
 		FilterFunc: consumergroup.Filter("trigger"),
-		Handler:    controller.HandleAll(impl.Enqueue),
+		Handler:    controller.HandleAll(consumergroup.Enqueue("trigger", impl.EnqueueKey)),
 	})
 
 	return impl
diff --git a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
index c1ef22033..8a79670b9 100644
--- a/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
+++ b/control-plane/pkg/reconciler/trigger/v2/triggerv2.go
@@ -111,7 +111,6 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		}
 	}
 
-	//todo additionally handle trigger.Spec.Filter
 	newcg := &internalscg.ConsumerGroup{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      string(trigger.UID),
@@ -138,8 +137,6 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		},
 	}
 
-	//todo validate newcg
-
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(trigger.GetNamespace()).Get(string(trigger.UID)) //Get by consumer group name
 	if err != nil && !apierrors.IsNotFound(err) {
 		return nil, err
@@ -166,7 +163,7 @@ func (r ReconcilerV2) reconcileConsumerGroup(ctx context.Context, trigger *event
 		Spec:       newcg.Spec,
 		Status:     cg.Status,
 	}
-	if _, err := r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
+	if cg, err = r.InternalsClient.InternalV1alpha1().ConsumerGroups(cg.GetNamespace()).Update(ctx, newCg, metav1.UpdateOptions{}); err != nil {
 		return nil, fmt.Errorf("failed to update consumer group %s/%s: %w", newCg.GetNamespace(), newCg.GetName(), err)
 	}
 
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index c7da64c13..cd4f96acb 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 159 third-party dependencies.
+Lists of 158 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.10 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.10 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.13.1 - http://github.com/FasterXML/jackson)
@@ -85,19 +85,18 @@ Lists of 159 third-party dependencies.
      (Apache License, Version 2.0) Netty/Transport/Classes/Epoll (io.netty:netty-transport-classes-epoll:4.1.72.Final - https://netty.io/netty-transport-classes-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.72.Final - https://netty.io/netty-transport-native-epoll/)
      (Apache License, Version 2.0) Netty/Transport/Native/Unix/Common (io.netty:netty-transport-native-unix-common:4.1.72.Final - https://netty.io/netty-transport-native-unix-common/)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-metrics:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.9.1 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.9.1-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-extension-trace-propagators:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-logs:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-testing:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:1.10.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.10.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.12.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (The Apache Software License, Version 2.0) Prometheus Java Span Context Supplier - Common (io.prometheus:simpleclient_tracer_common:0.12.0 - http://github.com/prometheus/client_java/simpleclient_tracer/simpleclient_tracer_common)
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
index ca8e8fe21..cb89d7d0c 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
@@ -20,6 +20,7 @@ import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.metrics.MetricsOptions;
+import io.vertx.core.net.PemKeyCertOptions;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.micrometer.MetricsDomain;
 import io.vertx.micrometer.MetricsNaming;
@@ -28,9 +29,16 @@ import io.vertx.micrometer.VertxPrometheusOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.producer.Producer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class Metrics {
 
+  private static final Logger logger = LoggerFactory.getLogger(Metrics.class);
+
+  private static final PemKeyCertOptions pemKeyCertOptions = permKeyCertOptions();
+  private static final String host = getHost();
+
   public static final String METRICS_REGISTRY_NAME = "metrics";
 
   // Micrometer employs a naming convention that separates lowercase words with a '.' (dot) character.
@@ -77,6 +85,9 @@ public class Metrics {
         .setEmbeddedServerOptions(new HttpServerOptions()
           .setPort(metricsConfigs.getMetricsPort())
           .setTracingPolicy(TracingPolicy.IGNORE)
+          .setSsl(pemKeyCertOptions != null)
+          .setPemKeyCertOptions(pemKeyCertOptions)
+          .setHost(host)
         )
         .setEmbeddedServerEndpoint(metricsConfigs.getMetricsPath())
         .setPublishQuantiles(metricsConfigs.isPublishQuantilesEnabled())
@@ -119,4 +130,23 @@ public class Metrics {
     clientMetrics.bindTo(getRegistry());
     return clientMetrics;
   }
+
+  public static PemKeyCertOptions permKeyCertOptions() {
+    final var certPath = System.getenv().get("METRICS_PEM_CERT_PATH");
+    final var keyPath = System.getenv().get("METRICS_PEM_KEY_PATH");
+    if (certPath == null || keyPath == null) {
+      logger.info("Metrics cert paths weren't provided, server will start without TLS");
+      return null;
+    }
+    return new PemKeyCertOptions().setCertPath(certPath).setKeyPath(keyPath);
+  }
+
+  private static String getHost() {
+    final var host = System.getenv().get("METRICS_HOST");
+    if (host == null) {
+      logger.info("Metrics server host wasn't provided, using default value " + HttpServerOptions.DEFAULT_HOST);
+      return HttpServerOptions.DEFAULT_HOST;
+    }
+    return host;
+  }
 }
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 8463b365f..8fb9aa685 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -44,7 +44,7 @@
     <vertx.version>4.2.3</vertx.version>
     <cloudevents.sdk.version>2.3.0</cloudevents.sdk.version>
     <micrometer.version>1.8.1</micrometer.version>
-    <opentelemetry.version>1.9.1</opentelemetry.version>
+    <opentelemetry.version>1.10.0</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
     <protobuf.version>3.19.2</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
diff --git a/go.mod b/go.mod
index 195e5aefa..b0f2f368d 100644
--- a/go.mod
+++ b/go.mod
@@ -23,9 +23,9 @@ require (
 	k8s.io/apiserver v0.22.5
 	k8s.io/client-go v0.22.5
 	k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a
-	knative.dev/eventing v0.28.1-0.20220104175430-5e3183b552ca
-	knative.dev/eventing-kafka v0.28.1-0.20220104160131-50694802d1ac
+	knative.dev/eventing v0.28.1-0.20220107104023-eb62db93c6c6
+	knative.dev/eventing-kafka v0.28.1-0.20220106163101-15c2aac56047
 	knative.dev/hack v0.0.0-20211222071919-abd085fc43de
-	knative.dev/pkg v0.0.0-20220104185830-52e42b760b54
+	knative.dev/pkg v0.0.0-20220105211333-96f18522d78d
 	knative.dev/reconciler-test v0.0.0-20211222120418-816f2192fec9
 )
diff --git a/go.sum b/go.sum
index 315f8aec1..60195c330 100644
--- a/go.sum
+++ b/go.sum
@@ -786,7 +786,6 @@ go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E=
 go.opentelemetry.io/contrib v0.20.0/go.mod h1:G/EtFaa6qaN7+LxqfIAT3GiZa7Wv5DTBUzl5H4LY0Kc=
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.20.0/go.mod h1:oVGt1LRbBOBq1A5BQLlUg9UaU/54aiHw8cgjV3aWZ/E=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.20.0/go.mod h1:2AboqHi0CiIZU0qwhtUfCYD1GeUzvvIXWNkhDt7ZMG4=
-go.opentelemetry.io/otel v0.16.0/go.mod h1:e4GKElweB8W2gWUqbghw0B8t5MCTccc9212eNHnOHwA=
 go.opentelemetry.io/otel v0.20.0 h1:eaP0Fqu7SXHwvjiqDq83zImeehOHX8doTvU9AwXON8g=
 go.opentelemetry.io/otel v0.20.0/go.mod h1:Y3ugLH2oa81t5QO+Lty+zXf8zC9L26ax4Nzoxm/dooo=
 go.opentelemetry.io/otel/exporters/otlp v0.20.0/go.mod h1:YIieizyaN77rtLJra0buKiNBOm9XQfkPEKBeuhoMwAM=
@@ -1395,21 +1394,20 @@ k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a h1:8dYfu/Fc9Gz2rNJKB9IQRGgQOh2clmRzNIPPY1xLY5g=
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20211208065539-a3e65c058156/go.mod h1:LR9JdrBn9RoCLboRN9a0OSiq0/PIoi8WlQQMKR+SXG8=
-knative.dev/eventing v0.28.0/go.mod h1:zxoB37kQxEkfCvKA7wZuHOPYInz6SFJzrhmG9IAd/iY=
-knative.dev/eventing v0.28.1-0.20220104175430-5e3183b552ca h1:MkWAmlLOfIYbtuVPzwsuuoqWCcoFz3oD/RZJ5Ea3OKs=
-knative.dev/eventing v0.28.1-0.20220104175430-5e3183b552ca/go.mod h1:Oucdv3+KRpZunYs26FSp98t0mLm78AAwZQok5tKeDTA=
-knative.dev/eventing-kafka v0.28.1-0.20220104160131-50694802d1ac h1:cjkmGWbEmNNgWYvhnWCaz9R25EuCBDXLT6ROst1QE4I=
-knative.dev/eventing-kafka v0.28.1-0.20220104160131-50694802d1ac/go.mod h1:/AYX5c6EhXGCQrTVDazkbGnsEPYdKfkW1JNE7D/cT9I=
+knative.dev/eventing v0.28.1-0.20220105073332-4348c292392a/go.mod h1:iZaGo6qiM8mPac2OCTZxeI/4gJIlFIBZA2oVZ8pUwh4=
+knative.dev/eventing v0.28.1-0.20220107104023-eb62db93c6c6 h1:Zs0L0HNpgxJwIjHKCNGDPn5zv4VEp+faXgsWd75xOco=
+knative.dev/eventing v0.28.1-0.20220107104023-eb62db93c6c6/go.mod h1:iZaGo6qiM8mPac2OCTZxeI/4gJIlFIBZA2oVZ8pUwh4=
+knative.dev/eventing-kafka v0.28.1-0.20220106163101-15c2aac56047 h1:dq/VCinmHmv6xOelvzsoqtymaClIqJyN6g5t6nL6108=
+knative.dev/eventing-kafka v0.28.1-0.20220106163101-15c2aac56047/go.mod h1:HNMGhMYFDKEm2b2QgV5+oaUng6JfZKQZJOhQWGaeUOo=
 knative.dev/hack v0.0.0-20211122162614-813559cefdda/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211203062838-e11ac125e707/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20211222071919-abd085fc43de h1:K7UeyvIfdTjznffAZg2L4fDkOuFWEDiaxgEa+B33nP8=
 knative.dev/hack v0.0.0-20211222071919-abd085fc43de/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
-knative.dev/hack/schema v0.0.0-20211203062838-e11ac125e707/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/hack/schema v0.0.0-20211222071919-abd085fc43de/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20211206113427-18589ac7627e/go.mod h1:E6B4RTjZyxe55a0kxOlnEHEl71zuG7gghnqYvNBKwBw=
-knative.dev/pkg v0.0.0-20211216142117-79271798f696/go.mod h1:hrD91/shO1o4KMZa4oWhnbRPmVJhvq86TLy/STF/qf8=
-knative.dev/pkg v0.0.0-20220104185830-52e42b760b54 h1:rSu9UPVufkvSTKhPoF5vTJTPNmvzwW4Dza1pD9GgO2w=
 knative.dev/pkg v0.0.0-20220104185830-52e42b760b54/go.mod h1:189cvGP0mwpqwZGFrLk5WuERIsNI/J6HuQ1CIX7SXxY=
+knative.dev/pkg v0.0.0-20220105211333-96f18522d78d h1:KqTqUP+w382CaI7NdIGaFLSI0qq2vo4QT93zzbsLnYY=
+knative.dev/pkg v0.0.0-20220105211333-96f18522d78d/go.mod h1:189cvGP0mwpqwZGFrLk5WuERIsNI/J6HuQ1CIX7SXxY=
 knative.dev/reconciler-test v0.0.0-20211207070557-0d138a88867b/go.mod h1:dCq1Fuu+eUISdnxABMvoDhefF91DYwE6O3rTYTraXbw=
 knative.dev/reconciler-test v0.0.0-20211222120418-816f2192fec9 h1:TPGEgrp5qF18YAc6hd4CvI4ykx10qWNv5KZJMJWr1Qc=
 knative.dev/reconciler-test v0.0.0-20211222120418-816f2192fec9/go.mod h1:dCq1Fuu+eUISdnxABMvoDhefF91DYwE6O3rTYTraXbw=
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
index d78c7a786..388071e88 100644
--- a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -80,7 +80,10 @@ func (c *Clientset) Tracker() testing.ObjectTracker {
 	return c.tracker
 }
 
-var _ clientset.Interface = &Clientset{}
+var (
+	_ clientset.Interface = &Clientset{}
+	_ testing.FakeClient  = &Clientset{}
+)
 
 // BindingsV1beta1 retrieves the BindingsV1beta1Client
 func (c *Clientset) BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface {
diff --git a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/fake/clientset_generated.go b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/fake/clientset_generated.go
index 72ba5e0dc..e601c247c 100644
--- a/vendor/knative.dev/eventing/pkg/client/clientset/versioned/fake/clientset_generated.go
+++ b/vendor/knative.dev/eventing/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -84,7 +84,10 @@ func (c *Clientset) Tracker() testing.ObjectTracker {
 	return c.tracker
 }
 
-var _ clientset.Interface = &Clientset{}
+var (
+	_ clientset.Interface = &Clientset{}
+	_ testing.FakeClient  = &Clientset{}
+)
 
 // EventingV1beta1 retrieves the EventingV1beta1Client
 func (c *Clientset) EventingV1beta1() eventingv1beta1.EventingV1beta1Interface {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index bda0e2589..dbf401ac5 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1114,7 +1114,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.28.1-0.20220104175430-5e3183b552ca
+# knative.dev/eventing v0.28.1-0.20220107104023-eb62db93c6c6
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1239,7 +1239,7 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/eventing-kafka v0.28.1-0.20220104160131-50694802d1ac
+# knative.dev/eventing-kafka v0.28.1-0.20220106163101-15c2aac56047
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1303,7 +1303,7 @@ knative.dev/eventing-kafka/test/upgrade/continual
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/pkg v0.0.0-20220104185830-52e42b760b54
+# knative.dev/pkg v0.0.0-20220105211333-96f18522d78d
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/reconciler/consumergroup/filter_test.go b/control-plane/pkg/reconciler/consumergroup/filter_test.go
index e56dc59b3..5f00b23c7 100644
--- a/control-plane/pkg/reconciler/consumergroup/filter_test.go
+++ b/control-plane/pkg/reconciler/consumergroup/filter_test.go
@@ -20,6 +20,7 @@ import (
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 
 	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
 )
@@ -76,3 +77,61 @@ func TestFilter(t *testing.T) {
 		})
 	}
 }
+
+func TestEnqueue(t *testing.T) {
+	tests := []struct {
+		name               string
+		resource           interface{}
+		userFacingResource string
+		want               int
+	}{
+		{
+			name:               "unknown type",
+			resource:           &kafkainternals.Consumer{},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+		{
+			name: "trigger pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "trigger",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               1,
+		},
+		{
+			name: "no pass",
+			resource: &kafkainternals.ConsumerGroup{
+				TypeMeta: metav1.TypeMeta{},
+				ObjectMeta: metav1.ObjectMeta{
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							Kind: "kafkasource",
+						},
+					},
+				},
+			},
+			userFacingResource: "trigger",
+			want:               0,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			called := 0
+			f := func(key types.NamespacedName) {
+				called++
+			}
+			Enqueue(tt.userFacingResource, f)(tt.resource)
+			if called != tt.want {
+				t.Errorf("Enqueue called %v, want %v", called, tt.want)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/source/v2/controllerv2_test.go b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
new file mode 100644
index 000000000..e378c42b1
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/v2/controllerv2_test.go
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package v2
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	_ "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/informers/eventing/v1alpha1/consumergroup/fake"
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewControllerV2(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	configs := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
+	}
+
+	ctx, _ = fakekubeclient.With(
+		ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configs.GeneralConfigMapName,
+				Namespace: configs.SystemNamespace,
+			},
+		},
+	)
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	controller := NewControllerV2(ctx, configs)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
diff --git a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
index 40907d43e..da6e5182c 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/broker/source_to_sink.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/google/uuid"
 	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/delivery"
 	"knative.dev/eventing/test/rekt/resources/eventlibrary"
 	"knative.dev/eventing/test/rekt/resources/flaker"
 	"knative.dev/eventing/test/rekt/resources/trigger"
@@ -70,18 +71,21 @@ func SourceToSink(brokerName string) *feature.Feature {
 	return f
 }
 
-// SourceToSinkWithDLQ tests to see if a Ready Broker acts as middleware.
+// SourceToSinkWithDLQ tests to see if a Trigger with no DLQ and a bad Sink, subscribed
+// to a Ready Broker, forward unsent events to it's Broker's DLQ
 //
 // source ---> broker<Via> --[trigger]--> bad uri
 //                |
-//                +--[DLQ]--> sink
+//                +--[DLQ]--> dlq
 //
-func SourceToSinkWithDLQ(brokerName, sinkName string, prober *eventshub.EventProber) *feature.Feature {
-	prober.SetTargetResource(broker.GVR(), brokerName)
+func SourceToSinkWithDLQ() *feature.Feature {
+	f := feature.NewFeature()
 
+	prober := eventshub.NewProber()
+	brokerName := feature.MakeRandomK8sName("broker")
+	dlq := feature.MakeRandomK8sName("dlq")
 	via := feature.MakeRandomK8sName("via")
-
-	f := feature.NewFeature()
+	source := feature.MakeRandomK8sName("source")
 
 	lib := feature.MakeRandomK8sName("lib")
 	f.Setup("install events", eventlibrary.Install(lib))
@@ -91,73 +95,29 @@ func SourceToSinkWithDLQ(brokerName, sinkName string, prober *eventshub.EventPro
 		panic(fmt.Errorf("can not find event files: %s", err))
 	}
 
-	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
-
-	// Resources ready.
-	f.Setup("trigger goes ready", trigger.IsReady(via))
-
-	// Install events after data plane is ready.
-	f.Setup("install source", prober.SenderInstall("source"))
-
-	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
-
-	// Assert events ended up where we expected.
-	f.Stable("broker with DLQ").
-		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ", prober.AssertReceivedAll("source", sinkName))
-
-	return f
-}
+	f.Setup("install dlq", prober.ReceiverInstall(dlq))
 
-// SourceToTwoSinksWithDLQ tests to see if a Ready Broker acts as middleware.
-//
-// source ---> broker +--[trigger<via1>]--> bad uri
-//                |   |
-//                |   +--[trigger<vai2>]--> sink2
-//                |
-//                +--[DLQ]--> sink1
-//
-func SourceToTwoSinksWithDLQ(brokerName, sink1, sink2 string, prober *eventshub.EventProber) *feature.Feature {
+	brokerConfig := append(broker.WithEnvConfig(), delivery.WithDeadLetterSink(prober.AsKReference(dlq), ""))
+	f.Setup("install broker", broker.Install(brokerName, brokerConfig...))
+	// Block till broker is ready
+	f.Setup("Broker is ready", broker.IsReady(brokerName))
 	prober.SetTargetResource(broker.GVR(), brokerName)
 
-	via1 := feature.MakeRandomK8sName("via")
-	via2 := feature.MakeRandomK8sName("via")
-
-	f := feature.NewFeature()
-
-	lib := feature.MakeRandomK8sName("lib")
-	f.Setup("install events", eventlibrary.Install(lib))
-	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
-	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
-	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
-		panic(fmt.Errorf("can not find event files: %s", err))
-	}
-
-	// Setup Probes
-	f.Setup("install recorder2", prober.ReceiverInstall(sink2))
-
-	// Setup data plane
-	f.Setup("install trigger via1", trigger.Install(via1, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
-	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference(sink2), "")))
-
-	// Resources ready.
-	f.Setup("trigger1 goes ready", trigger.IsReady(via1))
-	f.Setup("trigger2 goes ready", trigger.IsReady(via2))
+	f.Setup("install trigger", trigger.Install(via, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
+	// Trigger is ready.
+	f.Setup("trigger goes ready", trigger.IsReady(via))
 
 	// Install events after data plane is ready.
-	f.Setup("install source", prober.SenderInstall("source"))
+	f.Setup("install source", prober.SenderInstall(source))
 
 	// After we have finished sending.
-	f.Requirement("sender is finished", prober.SenderDone("source"))
-	f.Requirement("receiver 1 is finished", prober.ReceiverDone("source", sink1))
-	f.Requirement("receiver 2 is finished", prober.ReceiverDone("source", sink2))
+	f.Requirement("sender is finished", prober.SenderDone(source))
+	f.Requirement("receiver is finished", prober.ReceiverDone(source, dlq))
 
 	// Assert events ended up where we expected.
 	f.Stable("broker with DLQ").
-		Must("accepted all events", prober.AssertSentAll("source")).
-		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll("source", sink1)).
-		Must("deliver event to sink (via2)", prober.AssertReceivedAll("source", sink2))
+		Must("accepted all events", prober.AssertSentAll(source)).
+		Must("deliver event to DLQ", prober.AssertReceivedAll(source, dlq))
 
 	return f
 }
diff --git a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
index 6a5e3677d..b14f9b5a6 100644
--- a/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
+++ b/vendor/knative.dev/eventing/test/rekt/features/trigger/trigger_sink_resolution.go
@@ -132,6 +132,66 @@ func SourceToTriggerSinkWithDLSDontUseBrokers(triggerName, brokerName, brokerSin
 	return f
 }
 
+// source ---> broker +--[trigger<via1>]--> bad uri
+//                |   |
+//                |   +--[trigger<vai2>]--> sink
+//                |
+//                +--[DLQ]--> dlq
+//
+func BadTriggerDoesNotAffectOkTrigger() *feature.Feature {
+	f := feature.NewFeatureNamed("Bad Trigger does not affect good Trigger")
+
+	prober := eventshub.NewProber()
+	brokerName := feature.MakeRandomK8sName("broker")
+	via1 := feature.MakeRandomK8sName("via")
+	via2 := feature.MakeRandomK8sName("via")
+	dlq := feature.MakeRandomK8sName("dlq")
+	sink := feature.MakeRandomK8sName("sink")
+	source := feature.MakeRandomK8sName("source")
+
+	lib := feature.MakeRandomK8sName("lib")
+	f.Setup("install events", eventlibrary.Install(lib))
+	f.Setup("event cache is ready", eventlibrary.IsReady(lib))
+	f.Setup("use events cache", prober.SenderEventsFromSVC(lib, "events/three.ce"))
+	if err := prober.ExpectYAMLEvents(eventlibrary.PathFor("events/three.ce")); err != nil {
+		panic(fmt.Errorf("can not find event files: %s", err))
+	}
+
+	// Setup Probes
+	f.Setup("install dlq", prober.ReceiverInstall(dlq))
+	f.Setup("install sink2", prober.ReceiverInstall(sink))
+
+	// Setup data plane
+	brokerConfig := append(broker.WithEnvConfig(), delivery.WithDeadLetterSink(prober.AsKReference(dlq), ""))
+	f.Setup("install broker", broker.Install(brokerName, brokerConfig...))
+	// Block till broker is ready
+	f.Setup("Broker is ready", broker.IsReady(brokerName))
+	prober.SetTargetResource(broker.GVR(), brokerName)
+
+	f.Setup("install trigger via1", trigger.Install(via1, brokerName, trigger.WithSubscriber(nil, "bad://uri")))
+	f.Setup("install trigger via2", trigger.Install(via2, brokerName, trigger.WithSubscriber(prober.AsKReference(sink), "")))
+
+	// Resources ready.
+	f.Setup("trigger1 goes ready", trigger.IsReady(via1))
+	f.Setup("trigger2 goes ready", trigger.IsReady(via2))
+
+	// Install events after data plane is ready.
+	f.Requirement("install source", prober.SenderInstall(source))
+
+	// After we have finished sending.
+	f.Requirement("sender is finished", prober.SenderDone(source))
+	f.Requirement("receiver 1 is finished", prober.ReceiverDone(source, dlq))
+	f.Requirement("receiver 2 is finished", prober.ReceiverDone(source, sink))
+
+	// Assert events ended up where we expected.
+	f.Stable("broker with DLQ").
+		Must("accepted all events", prober.AssertSentAll(source)).
+		Must("deliver event to DLQ (via1)", prober.AssertReceivedAll(source, dlq)).
+		Must("deliver event to sink (via2)", prober.AssertReceivedAll(source, sink))
+
+	return f
+}
+
 func noEventsToDLS(prober *eventshub.EventProber, sinkName string) feature.StepFn {
 	return func(ctx context.Context, t feature.T) {
 		if len(prober.ReceivedBy(ctx, sinkName)) == 0 {