Repository: knative-sandbox/eventing-kafka-broker
Pull request #1701 (2022-01-07 08:04:34)

Set resource reference in Broker, Channel, and Sink contracts
To align receiver metrics to Eventing core #1695, we need
to have the resource reference in the data plane.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

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/1701#" 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":[]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1701](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (84dc1ec) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/e32734a003082847ed1d98a27f07f6a362e26248?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e32734a) will **increase** coverage by `0.16%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/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/1701?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    #1701      +/-   ##
============================================
+ Coverage     72.60%   72.76%   +0.16%     
+ Complexity      597      595       -2     
============================================
  Files           120      120              
  Lines          4428     4443      +15     
  Branches        171      171              
============================================
+ Hits           3215     3233      +18     
+ Misses          948      947       -1     
+ Partials        265      263       -2     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `81.80% <ø> (-0.11%)` | :arrow_down: |

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/1701?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `67.14% <100.00%> (+0.81%)` | :arrow_up: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `64.33% <100.00%> (+0.66%)` | :arrow_up: |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | `71.06% <100.00%> (+0.93%)` | :arrow_up: |
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `87.50% <0.00%> (-6.25%)` | :arrow_down: |
| [control-plane/pkg/prober/cache.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2NhY2hlLmdv) | `100.00% <0.00%> (+4.76%)` | :arrow_up: |
| [control-plane/pkg/prober/prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL3Byb2Jlci5nbw==) | `75.00% <0.00%> (+15.00%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701?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/1701?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [e32734a...84dc1ec](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1701?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/1701#pullrequestreview-846380629" title="Approved">matzew</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1701#" 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)~~ [matzew,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/.gitignore b/.gitignore
index 88cbbf329..2c2295fc7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -31,5 +31,6 @@ data-plane/**/build/
 eventing-kafka-controller.yaml
 eventing-kafka-source.yaml
 eventing-kafka-broker.yaml
+eventing-kafka-channel.yaml
 eventing-kafka-sink.yaml
 
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index d5596fc8a..7fcee74bb 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -380,6 +380,11 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 			},
 		},
 		BootstrapServers: config.GetBootstrapServers(),
+		Reference: &contract.Reference{
+			Uuid:      string(broker.GetUID()),
+			Namespace: broker.GetNamespace(),
+			Name:      broker.GetName(),
+		},
 	}
 
 	if secret != nil {
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index c7068a1eb..428d284f5 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -535,6 +535,11 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 			},
 		},
 		BootstrapServers: config.GetBootstrapServers(),
+		Reference: &contract.Reference{
+			Uuid:      string(channel.GetUID()),
+			Namespace: channel.GetNamespace(),
+			Name:      channel.GetName(),
+		},
 	}
 
 	if secret != nil {
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 3609b1506..c4f870ae7 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -175,6 +175,11 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 			IngressType: &contract.Ingress_Path{Path: receiver.PathFromObject(ks)},
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
+		Reference: &contract.Reference{
+			Uuid:      string(ks.GetUID()),
+			Namespace: ks.GetNamespace(),
+			Name:      ks.GetName(),
+		},
 	}
 	if ks.Spec.HasAuthConfig() {
 		sinkConfig.Auth = &contract.Resource_AuthSecret{
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index cad289f55..c7da64c13 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -20,8 +20,8 @@ Lists of 159 third-party dependencies.
      (Apache License, Version 2.0) Guava: Google Core Libraries for Java (com.google.guava:guava:30.1.1-android - https://github.com/google/guava/guava)
      (The Apache Software License, Version 2.0) Guava ListenableFuture only (com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava - https://github.com/google/guava/listenablefuture)
      (The Apache Software License, Version 2.0) J2ObjC Annotations (com.google.j2objc:j2objc-annotations:1.3 - https://github.com/google/j2objc/)
-     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java/)
-     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.1 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
+     (3-Clause BSD License) Protocol Buffers [Core] (com.google.protobuf:protobuf-java:3.19.2 - https://developers.google.com/protocol-buffers/protobuf-java/)
+     (3-Clause BSD License) Protocol Buffers [Util] (com.google.protobuf:protobuf-java-util:3.19.2 - https://developers.google.com/protocol-buffers/protobuf-java-util/)
      (Apache 2.0) OkHttp Logging Interceptor (com.squareup.okhttp3:logging-interceptor:3.12.12 - https://github.com/square/okhttp/logging-interceptor)
      (Apache 2.0) MockWebServer (com.squareup.okhttp3:mockwebserver:3.12.12 - https://github.com/square/okhttp/mockwebserver)
      (Apache 2.0) OkHttp (com.squareup.okhttp3:okhttp:3.14.9 - https://github.com/square/okhttp/okhttp)
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 468c4d300..8463b365f 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -46,7 +46,7 @@
     <micrometer.version>1.8.1</micrometer.version>
     <opentelemetry.version>1.9.1</opentelemetry.version>
     <jackson.version>2.13.1</jackson.version>
-    <protobuf.version>3.19.1</protobuf.version>
+    <protobuf.version>3.19.2</protobuf.version>
     <slf4j.version>1.7.32</slf4j.version>
     <ch.qos.logback.version>1.2.10</ch.qos.logback.version>
     <net.logstash.logback.encoder.version>7.0.1</net.logstash.logback.encoder.version>


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 7f4f3f2c3..757685137 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -144,6 +144,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -203,6 +204,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -264,6 +266,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -324,6 +327,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 2,
@@ -382,6 +386,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURLFrom(BrokerNamespace, ServiceName)},
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 2,
@@ -486,6 +491,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -541,6 +547,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -621,6 +628,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -707,6 +715,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: "http://www.my-sink.com/api"},
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -798,6 +807,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -877,6 +887,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 2,
@@ -929,6 +940,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -1046,6 +1058,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 						},
 					},
 					Generation: 1,
@@ -1117,6 +1130,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							Auth: &contract.Resource_AuthSecret{
 								AuthSecret: &contract.Reference{
 									Uuid:      SecretUUID,
@@ -1335,6 +1349,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							},
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 						},
 					},
@@ -1422,6 +1437,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter:    ServiceURL,
 								Retry:         10,
@@ -1487,6 +1503,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter:    ServiceURL,
 								Retry:         10,
@@ -1552,6 +1569,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter: ServiceURL,
 							},
@@ -1614,6 +1632,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
 								DeadLetter:    ServiceURL,
 								Retry:         10,
@@ -1665,6 +1684,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 						},
 					},
@@ -1714,6 +1734,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{BrokerTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        BrokerReference(),
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 438e1d885..eecb34858 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -22,9 +22,10 @@ import (
 	"testing"
 
 	"github.com/Shopify/sarama"
+	"knative.dev/pkg/network"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
-	"knative.dev/pkg/network"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -147,6 +148,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -215,6 +217,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -285,6 +288,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -356,6 +360,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -429,6 +434,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -502,6 +508,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -578,6 +585,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -855,6 +863,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -929,6 +938,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Auth: &contract.Resource_AuthSecret{
 								AuthSecret: &contract.Reference{
 									Uuid:      SecretUUID,
@@ -1009,6 +1019,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
@@ -1078,6 +1089,7 @@ func TestReconcileKind(t *testing.T) {
 							Uid:              ChannelUUID,
 							Topics:           []string{ChannelTopic()},
 							BootstrapServers: ChannelBootstrapServers,
+							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
 								IngressType: &contract.Ingress_Path{
 									Path: receiver.Path(ChannelNamespace, ChannelName),
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index c62a9001c..524d54a8f 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -133,6 +133,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -186,6 +187,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							Auth: &contract.Resource_AuthSecret{
 								AuthSecret: &contract.Reference{
@@ -252,6 +254,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 							Ingress: &contract.Ingress{
 								ContentMode: contract.ContentMode_STRUCTURED,
 								IngressType: &contract.Ingress_Path{
@@ -370,6 +373,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{"my-topic-1"},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: "kafka-broker:10000",
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -476,6 +480,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -525,6 +530,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -597,6 +603,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -642,6 +649,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED},
 						},
 					},
@@ -667,6 +675,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -744,6 +753,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -798,6 +808,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -857,6 +868,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
@@ -910,6 +922,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
 							BootstrapServers: bootstrapServers,
+							Reference:        SinkReference(),
 						},
 					},
 					Generation: 1,
diff --git a/control-plane/pkg/reconciler/testing/objects_broker.go b/control-plane/pkg/reconciler/testing/objects_broker.go
index 4af25d83f..5e1e368bf 100644
--- a/control-plane/pkg/reconciler/testing/objects_broker.go
+++ b/control-plane/pkg/reconciler/testing/objects_broker.go
@@ -33,6 +33,7 @@ import (
 	"knative.dev/pkg/network"
 
 	"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/prober"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
@@ -231,6 +232,14 @@ func BrokerAddressable(env *config.Env) func(broker *eventing.Broker) {
 	}
 }
 
+func BrokerReference() *contract.Reference {
+	return &contract.Reference{
+		Uuid:      BrokerUUID,
+		Namespace: BrokerNamespace,
+		Name:      BrokerName,
+	}
+}
+
 func BrokerDLSResolved(uri string) func(broker *eventing.Broker) {
 	return func(broker *eventing.Broker) {
 		broker.Status.DeadLetterSinkURI, _ = apis.ParseURL(uri)
diff --git a/control-plane/pkg/reconciler/testing/objects_channel.go b/control-plane/pkg/reconciler/testing/objects_channel.go
index 675b518da..336fc2ece 100644
--- a/control-plane/pkg/reconciler/testing/objects_channel.go
+++ b/control-plane/pkg/reconciler/testing/objects_channel.go
@@ -25,15 +25,17 @@ 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"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
-	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 	"knative.dev/pkg/network"
+
+	"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/reconciler/base"
+	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/channel"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
 const (
@@ -174,6 +176,14 @@ func ChannelAddressable(env *config.Env) func(obj duckv1.KRShaped) {
 	}
 }
 
+func ChannelReference() *contract.Reference {
+	return &contract.Reference{
+		Uuid:      ChannelUUID,
+		Namespace: ChannelNamespace,
+		Name:      ChannelName,
+	}
+}
+
 func WithInitKafkaChannelConditions(obj duckv1.KRShaped) {
 	channel := obj.(*messagingv1beta1.KafkaChannel)
 	channel.Status.InitializeConditions()
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index bc41426c2..7aaea0294 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -32,6 +32,7 @@ import (
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	"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/reconciler/base"
 )
 
@@ -126,6 +127,14 @@ func SinkAddressable(configs *config.Env) func(obj duckv1.KRShaped) {
 	}
 }
 
+func SinkReference() *contract.Reference {
+	return &contract.Reference{
+		Uuid:      SinkUUID,
+		Namespace: SinkNamespace,
+		Name:      SinkName,
+	}
+}
+
 func SinkReceiverPod(namespace string, annotations map[string]string) runtime.Object {
 	return &corev1.Pod{
 		TypeMeta: metav1.TypeMeta{