Repository: knative-sandbox/eventing-kafka-broker
Pull request #511 (2020-12-28 08:39:07)

Skip ConfigMap update when there are no changes
During E2E tests there are a lot of no-op reconciliation
since chaosduck kills the leader pod and every time
a new replica starts we trigger a global resync that 
most of the time doesn't actually change the ConfigMap.

For example, the ConfigMap diff looks like this:

```
  	"""
  	{
- 	  "generation": "1",
+ 	  "generation": "2",
  	  "resources": [
  	    {
  	... // 15 identical lines
  	"""
  )
```

This leads to an unnecessary reconciliation loop on the data plane
and increase update conflicts of the ConfigMap.

## Proposed Changes

- Skip ConfigMap update when there are no changes

<!--
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
:broom: Update or clean up current behavior
Skip ConfigMap update when there are no changes
```

/kind cleanup

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/511#" 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/master/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: /assign @slinkydeveloper 
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511?src=pr&el=h1) Report
> Merging [#511](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511?src=pr&el=desc) (db80eb7) into [master](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/c2552e2bc6471c161b006d9976375a25fc0957d1?el=desc) (c2552e2) will **increase** coverage by `1.15%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511?src=pr&el=tree)

```diff
@@             Coverage Diff              @@
##             master     #511      +/-   ##
============================================
+ Coverage     76.79%   77.94%   +1.15%     
  Complexity      269      269              
============================================
  Files            59       37      -22     
  Lines          1969     1179     -790     
  Branches         82       82              
============================================
- Hits           1512      919     -593     
+ Misses          342      200     -142     
+ Partials        115       60      -55     
```

| Flag | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| java-unittests | `77.94% <ø> (ø)` | `0.00 <ø> (ø)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3ZhbGlkYXRpb24uZ28=) | | | |
| [control-plane/pkg/reconciler/kafka/admin.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9hZG1pbi5nbw==) | | | |
| [control-plane/pkg/reconciler/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS9icm9rZXJfY2xhc3MuZ28=) | | | |
| [...ntrol-plane/pkg/reconciler/broker/broker\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyX2NvbmZpZy5nbw==) | | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | | |
| [.../pkg/apis/eventing/v1alpha1/kafka\_sink\_defaults.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2RlZmF1bHRzLmdv) | | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | | |
| [control-plane/pkg/core/config/utils.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvdXRpbHMuZ28=) | | | |
| ... and [7 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/511/diff?src=pr&el=tree-more) | |

------

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

knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-broker-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[control-plane/pkg/reconciler/broker/broker.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/511/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1343476572501315584/artifacts/line-cov.html#file2) | 89.2% | 89.4% | 0.2
[control-plane/pkg/reconciler/sink/kafka_sink.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/511/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1343476572501315584/artifacts/line-cov.html#file3) | 84.9% | 85.3% | 0.4
[control-plane/pkg/reconciler/trigger/trigger.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/511/pull-knative-sandbox-eventing-kafka-broker-go-coverage/1343476572501315584/artifacts/line-cov.html#file4) | 85.9% | 85.7% | -0.1


Patch:
diff --git a/control-plane/pkg/core/config/egress.go b/control-plane/pkg/core/config/egress.go
index 628dce6a6..938e78b22 100644
--- a/control-plane/pkg/core/config/egress.go
+++ b/control-plane/pkg/core/config/egress.go
@@ -17,7 +17,9 @@
 package config
 
 import (
+	"google.golang.org/protobuf/proto"
 	"k8s.io/apimachinery/pkg/types"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 )
 
@@ -36,3 +38,29 @@ func FindEgress(egresses []*contract.Egress, egress types.UID) int {
 
 	return NoEgress
 }
+
+const (
+	EgressChanged = iota
+	EgressUnchanged
+)
+
+// AddOrUpdateEgressConfig adds or updates the given egress to the given contract at the specified indexes.
+func AddOrUpdateEgressConfig(ct *contract.Contract, brokerIndex int, egress *contract.Egress, egressIndex int) int {
+
+	if egressIndex != NoEgress {
+		prev := ct.Resources[brokerIndex].Egresses[egressIndex]
+		ct.Resources[brokerIndex].Egresses[egressIndex] = egress
+
+		if proto.Equal(prev, egress) {
+			return EgressUnchanged
+		}
+		return EgressChanged
+	}
+
+	ct.Resources[brokerIndex].Egresses = append(
+		ct.Resources[brokerIndex].Egresses,
+		egress,
+	)
+
+	return EgressChanged
+}
diff --git a/control-plane/pkg/core/config/resource.go b/control-plane/pkg/core/config/resource.go
index a4b5ca8a6..306c2b514 100644
--- a/control-plane/pkg/core/config/resource.go
+++ b/control-plane/pkg/core/config/resource.go
@@ -18,7 +18,9 @@ package config
 
 import (
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/proto"
 	"k8s.io/apimachinery/pkg/types"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 )
 
@@ -39,20 +41,32 @@ func FindResource(contract *contract.Contract, resource types.UID) int {
 	return resourceIndex
 }
 
+const (
+	ResourceChanged = iota
+	ResourceUnchanged
+)
+
 // AddOrUpdateResourceConfig adds or updates the given resourceConfig to the given resources at the specified index.
-func AddOrUpdateResourceConfig(contract *contract.Contract, resource *contract.Resource, index int, logger *zap.Logger) {
+func AddOrUpdateResourceConfig(contract *contract.Contract, resource *contract.Resource, index int, logger *zap.Logger) int {
 
 	if index != NoResource {
+		logger.Debug("Resource exists", zap.Int("index", index))
+
+		prev := contract.Resources[index]
 		resource.Egresses = contract.Resources[index].Egresses
 		contract.Resources[index] = resource
 
-		logger.Debug("Resource exists", zap.Int("index", index))
+		if proto.Equal(prev, resource) {
+			return ResourceUnchanged
+		}
+		return ResourceChanged
+	}
 
-	} else {
-		contract.Resources = append(contract.Resources, resource)
+	logger.Debug("Resource doesn't exist")
 
-		logger.Debug("Resource doesn't exist")
-	}
+	contract.Resources = append(contract.Resources, resource)
+
+	return ResourceChanged
 }
 
 // DeleteResource deletes the resource at the given index from Resources.
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index c00559fbe..97297ee63 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -138,51 +138,57 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	brokerIndex := coreconfig.FindResource(ct, broker.UID)
 	// Update contract data with the new contract configuration
-	coreconfig.AddOrUpdateResourceConfig(ct, brokerResource, brokerIndex, logger)
+	changed := coreconfig.AddOrUpdateResourceConfig(ct, brokerResource, brokerIndex, logger)
+
+	logger.Debug("Change detector", zap.Int("changed", changed))
 
 	// Increment volumeGeneration
 	ct.Generation = incrementContractGeneration(ct.Generation)
 
-	// Update the configuration map with the new contract data.
-	if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
-		logger.Error("failed to update data plane config map", zap.Error(
-			statusConditionManager.FailedToUpdateConfigMap(err),
-		))
-		return err
+	if changed == coreconfig.ResourceChanged {
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			logger.Error("failed to update data plane config map", zap.Error(
+				statusConditionManager.FailedToUpdateConfigMap(err),
+			))
+			return err
+		}
+		logger.Debug("Contract config map updated")
 	}
 	statusConditionManager.ConfigMapUpdated()
 
-	logger.Debug("Contract config map updated")
+	if changed == coreconfig.ResourceChanged {
 
-	// After #37 we reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
-	// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
-	// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
-	// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
-	// the update even if here eventually means seconds or minutes after the actual update.
+		// After #37 we reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
+		// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
+		// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
+		// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
+		// the update even if here eventually means seconds or minutes after the actual update.
 
-	// Update volume generation annotation of receiver pods
-	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-		logger.Error("Failed to update receiver pod annotation", zap.Error(
-			statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
-		))
-		return err
-	}
-
-	logger.Debug("Updated receiver pod annotation")
-
-	// Update volume generation annotation of dispatcher pods
-	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
-		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
-		// ready. So, log out the error and move on to the next step.
-		logger.Warn(
-			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
-			zap.Error(err),
-		)
+		// Update volume generation annotation of receiver pods
+		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+			logger.Error("Failed to update receiver pod annotation", zap.Error(
+				statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
+			))
+			return err
+		}
 
-		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
-	} else {
-		logger.Debug("Updated dispatcher pod annotation")
+		logger.Debug("Updated receiver pod annotation")
+
+		// Update volume generation annotation of dispatcher pods
+		if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+			// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+			// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
+			// ready. So, log out the error and move on to the next step.
+			logger.Warn(
+				"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+				zap.Error(err),
+			)
+
+			statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+		} else {
+			logger.Debug("Updated dispatcher pod annotation")
+		}
 	}
 
 	return statusConditionManager.Reconciled()
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index d9f2ee59b..a6b1f6886 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -147,31 +147,35 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 	sinkIndex := coreconfig.FindResource(ct, ks.UID)
 	// Update contract data with the new sink configuration.
-	coreconfig.AddOrUpdateResourceConfig(ct, sinkConfig, sinkIndex, logger)
+	changed := coreconfig.AddOrUpdateResourceConfig(ct, sinkConfig, sinkIndex, logger)
 
 	// Increment volumeGeneration
 	ct.Generation = incrementGeneration(ct.Generation)
 
-	// Update the configuration map with the new contract data.
-	if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
-		logger.Error("failed to update data plane config map", zap.Error(
-			statusConditionManager.FailedToUpdateConfigMap(err),
-		))
-		return err
+	if changed == coreconfig.ResourceChanged {
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			logger.Error("failed to update data plane config map", zap.Error(
+				statusConditionManager.FailedToUpdateConfigMap(err),
+			))
+			return err
+		}
 	}
 	statusConditionManager.ConfigMapUpdated()
 
 	logger.Debug("Config map updated")
 
-	// After #37 we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
-	// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
+	if changed == coreconfig.ResourceChanged {
+		// After #37 we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
+		// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
 
-	// Update volume generation annotation of receiver pods
-	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-		return err
-	}
+		// Update volume generation annotation of receiver pods
+		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+			return err
+		}
 
-	logger.Debug("Updated receiver pod annotation")
+		logger.Debug("Updated receiver pod annotation")
+	}
 
 	return statusConditionManager.Reconciled()
 }
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index d9ba4d2ca..eecf88a7d 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -252,40 +252,34 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	if err != nil {
 		return statusConditionManager.failedToResolveTriggerConfig(err)
 	}
-
 	statusConditionManager.subscriberResolved()
 
-	if triggerIndex == coreconfig.NoEgress {
-		ct.Resources[brokerIndex].Egresses = append(
-			ct.Resources[brokerIndex].Egresses,
-			triggerConfig,
-		)
-	} else {
-		ct.Resources[brokerIndex].Egresses[triggerIndex] = triggerConfig
-	}
+	changed := coreconfig.AddOrUpdateEgressConfig(ct, brokerIndex, triggerConfig, triggerIndex)
 
 	// Increment volumeGeneration
 	ct.Generation = incrementGeneration(ct.Generation)
 
-	// Update the configuration map with the new dataPlaneConfig data.
-	if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
-		trigger.Status.MarkDependencyFailed(string(base.ConditionConfigMapUpdated), err.Error())
-		return err
-	}
-
-	// Update volume generation annotation of dispatcher pods
-	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
-		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Trigger
-		// ready. So, log out the error and move on to the next step.
-		logger.Warn(
-			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
-			zap.Error(err),
-		)
+	if changed == coreconfig.EgressChanged {
+		// Update the configuration map with the new dataPlaneConfig data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			trigger.Status.MarkDependencyFailed(string(base.ConditionConfigMapUpdated), err.Error())
+			return err
+		}
 
-		statusConditionManager.failedToUpdateDispatcherPodsAnnotation(err)
-	} else {
-		logger.Debug("Updated dispatcher pod annotation")
+		// Update volume generation annotation of dispatcher pods
+		if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+			// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+			// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Trigger
+			// ready. So, log out the error and move on to the next step.
+			logger.Warn(
+				"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+				zap.Error(err),
+			)
+
+			statusConditionManager.failedToUpdateDispatcherPodsAnnotation(err)
+		} else {
+			logger.Debug("Updated dispatcher pod annotation")
+		}
 	}
 
 	logger.Debug("Contract config map updated")
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index f27df292c..1009db6f1 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 109 third-party dependencies.
+Lists of 111 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.3 - 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.3 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.11.3 - http://github.com/FasterXML/jackson)
@@ -28,15 +28,15 @@ Lists of 109 third-party dependencies.
      (Unknown license) core (dev.knative.eventing.kafka.broker:core:1.0-SNAPSHOT - no url defined)
      (Unknown license) dispatcher (dev.knative.eventing.kafka.broker:dispatcher:1.0-SNAPSHOT - no url defined)
      (Unknown license) receiver (dev.knative.eventing.kafka.broker:receiver:1.0-SNAPSHOT - no url defined)
-     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.0.0.RC1 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
-     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.0.0.RC1 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
-     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.0.0.RC1 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
-     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.0.0.RC1 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
-     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.0.0.RC1 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
+     (The Apache Software License, Version 2.0) CloudEvents - API (io.cloudevents:cloudevents-api:2.0.0.RC2 - https://cloudevents.github.io/sdk-java/cloudevents-api/)
+     (The Apache Software License, Version 2.0) CloudEvents - Core (io.cloudevents:cloudevents-core:2.0.0.RC2 - https://cloudevents.github.io/sdk-java/cloudevents-core/)
+     (The Apache Software License, Version 2.0) CloudEvents - Vert.x Http Binding (io.cloudevents:cloudevents-http-vertx:2.0.0.RC2 - https://cloudevents.github.io/sdk-java/cloudevents-http-vertx/)
+     (The Apache Software License, Version 2.0) CloudEvents - JSON Jackson (io.cloudevents:cloudevents-json-jackson:2.0.0.RC2 - https://cloudevents.github.io/sdk-java/cloudevents-json-jackson/)
+     (The Apache Software License, Version 2.0) CloudEvents - Kafka Binding (io.cloudevents:cloudevents-kafka:2.0.0.RC2 - https://cloudevents.github.io/sdk-java/cloudevents-kafka/)
      (Apache Software License 2.0) Debezium API (io.debezium:debezium-api:1.3.1.Final - https://debezium.io/debezium-api)
      (Apache Software License 2.0) Debezium Core (io.debezium:debezium-core:1.3.1.Final - https://debezium.io/debezium-core)
-     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.6.1 - https://github.com/micrometer-metrics/micrometer)
-     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.6.1 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-core (io.micrometer:micrometer-core:1.6.2 - https://github.com/micrometer-metrics/micrometer)
+     (The Apache Software License, Version 2.0) micrometer-registry-prometheus (io.micrometer:micrometer-registry-prometheus:1.6.2 - https://github.com/micrometer-metrics/micrometer)
      (Apache License, Version 2.0) Netty/Buffer (io.netty:netty-buffer:4.1.49.Final - https://netty.io/netty-buffer/)
      (Apache License, Version 2.0) Netty/Codec (io.netty:netty-codec:4.1.49.Final - https://netty.io/netty-codec/)
      (Apache License, Version 2.0) Netty/Codec/DNS (io.netty:netty-codec-dns:4.1.49.Final - https://netty.io/netty-codec-dns/)
@@ -51,14 +51,18 @@ Lists of 109 third-party dependencies.
      (Apache License, Version 2.0) Netty/Transport (io.netty:netty-transport:4.1.49.Final - https://netty.io/netty-transport/)
      (Apache License, Version 2.0) Netty/Transport/Native/Epoll (io.netty:netty-transport-native-epoll:4.1.49.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.49.Final - https://netty.io/netty-transport-native-unix-common/)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
-     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:0.12.0 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-baggage:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-common:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-metrics:0.13.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-api-trace:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-context:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-logging:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-exporter-zipkin:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-common:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-metrics:0.13.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
+     (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-sdk-trace:0.13.1 - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.9.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.9.0 - http://github.com/prometheus/client_java/simpleclient_common)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.0.0 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
@@ -83,10 +87,9 @@ Lists of 109 third-party dependencies.
      (Apache License, Version 2.0) Apache ZooKeeper - Jute (org.apache.zookeeper:zookeeper-jute:3.5.8 - http://zookeeper.apache.org/zookeeper-jute)
      (The Apache License, Version 2.0) org.apiguardian:apiguardian-api (org.apiguardian:apiguardian-api:1.1.0 - https://github.com/apiguardian-team/apiguardian)
      (Apache License, Version 2.0) AssertJ fluent assertions (org.assertj:assertj-core:3.18.1 - https://assertj.github.io/doc/assertj-core/)
-     (Apache 2.0) Awaitility (org.awaitility:awaitility:3.0.0 - http://awaitility.org)
+     (Apache 2.0) Awaitility (org.awaitility:awaitility:4.0.3 - http://awaitility.org)
      (GNU General Public License, version 2 (GPL2), with the classpath exception) (The MIT License) Checker Qual (org.checkerframework:checker-compat-qual:2.5.5 - https://checkerframework.org)
-     (New BSD License) Hamcrest Core (org.hamcrest:hamcrest-core:1.3 - https://github.com/hamcrest/JavaHamcrest/hamcrest-core)
-     (New BSD License) Hamcrest library (org.hamcrest:hamcrest-library:1.3 - https://github.com/hamcrest/JavaHamcrest/hamcrest-library)
+     (BSD Licence 3) Hamcrest (org.hamcrest:hamcrest:2.1 - http://hamcrest.org/JavaHamcrest/)
      (BSD-2-Clause) (Public Domain, per Creative Commons CC0) HdrHistogram (org.hdrhistogram:HdrHistogram:2.1.12 - http://hdrhistogram.github.io/HdrHistogram/)
      (Apache License 2.0) (LGPL 2.1) (MPL 1.1) Javassist (org.javassist:javassist:3.26.0-GA - http://www.javassist.org/)
      (Eclipse Public License v2.0) JUnit Jupiter (Aggregator) (org.junit.jupiter:junit-jupiter:5.7.0 - https://junit.org/junit5/)
@@ -99,7 +102,6 @@ Lists of 109 third-party dependencies.
      (The Apache Software License, Version 2.0) LZ4 and xxHash (org.lz4:lz4-java:1.7.1 - https://github.com/lz4/lz4-java)
      (The MIT License) mockito-core (org.mockito:mockito-core:3.6.28 - https://github.com/mockito/mockito)
      (The MIT License) mockito-junit-jupiter (org.mockito:mockito-junit-jupiter:3.6.28 - https://github.com/mockito/mockito)
-     (Apache 2) Objenesis (org.objenesis:objenesis:2.5.1 - http://objenesis.org)
      (Apache License, Version 2.0) Objenesis (org.objenesis:objenesis:3.1 - http://objenesis.org)
      (The Apache License, Version 2.0) org.opentest4j:opentest4j (org.opentest4j:opentest4j:1.2.0 - https://github.com/ota4j-team/opentest4j)
      (The New BSD License) (WTFPL) Reflections (org.reflections:reflections:0.9.12 - http://github.com/ronmamo/reflections)
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/Tracing.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/Tracing.java
index 14233ab9f..73dd8f433 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/Tracing.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/tracing/Tracing.java
@@ -15,13 +15,11 @@
  */
 package dev.knative.eventing.kafka.broker.core.tracing;
 
-import static net.logstash.logback.argument.StructuredArguments.keyValue;
-
 import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig.Backend;
 import io.opentelemetry.exporter.logging.LoggingSpanExporter;
 import io.opentelemetry.exporter.zipkin.ZipkinSpanExporter;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
-import io.opentelemetry.sdk.trace.TracerSdkManagement;
+import io.opentelemetry.sdk.trace.SdkTracerManagement;
 import io.opentelemetry.sdk.trace.config.TraceConfig;
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
 import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
@@ -30,6 +28,8 @@ import io.opentelemetry.sdk.trace.samplers.Sampler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static net.logstash.logback.argument.StructuredArguments.keyValue;
+
 public class Tracing {
 
   public final static String SERVICE_NAME;
@@ -45,7 +45,7 @@ public class Tracing {
 
   private static final Logger logger = LoggerFactory.getLogger(Tracing.class);
 
-  private static final TracerSdkManagement tracerManagement = OpenTelemetrySdk.getGlobalTracerManagement();
+  private static final SdkTracerManagement tracerManagement = OpenTelemetrySdk.getGlobalTracerManagement();
 
   public static void setup(final TracingConfig tracingConfig) {
 
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 2b834f49b..fdf3ffeab 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -45,7 +45,7 @@
     <vertx.version>4.0.0</vertx.version>
     <cloudevents.sdk.version>2.0.0.RC2</cloudevents.sdk.version>
     <micrometer.version>1.6.2</micrometer.version>
-    <opentelemetry.version>0.12.0</opentelemetry.version>
+    <opentelemetry.version>0.13.1</opentelemetry.version>
     <jackson.version>2.12.0</jackson.version>
     <protobuf.version>3.14.0</protobuf.version>
     <slf4j.version>1.7.30</slf4j.version>


Test patch:
diff --git a/control-plane/pkg/core/config/egress_test.go b/control-plane/pkg/core/config/egress_test.go
index 892c30a9e..192b84398 100644
--- a/control-plane/pkg/core/config/egress_test.go
+++ b/control-plane/pkg/core/config/egress_test.go
@@ -19,6 +19,9 @@ package config
 import (
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+	"google.golang.org/protobuf/testing/protocmp"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"k8s.io/apimachinery/pkg/types"
@@ -91,3 +94,127 @@ func TestFindEgress(t *testing.T) {
 		})
 	}
 }
+
+func TestAddOrUpdateEgressConfig(t *testing.T) {
+	tests := []struct {
+		name        string
+		givenCt     *contract.Contract
+		brokerIndex int
+		egress      *contract.Egress
+		egressIndex int
+		changed     int
+		wantCt      *contract.Contract
+	}{
+		{
+			name: "Egress found - changed",
+			givenCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "xyz",
+							},
+						},
+					},
+				},
+			},
+			brokerIndex: 0,
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "abc",
+							},
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "Egress found - unchanged",
+			givenCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "abc",
+							},
+						},
+					},
+				},
+			},
+			brokerIndex: 0,
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: 0,
+			wantCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "abc",
+							},
+						},
+					},
+				},
+			},
+			changed: EgressUnchanged,
+		},
+		{
+			name: "Egress not found",
+			givenCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "abc",
+							},
+						},
+					},
+				},
+			},
+			brokerIndex: 0,
+			egress: &contract.Egress{
+				Uid: "abc",
+			},
+			egressIndex: NoEgress,
+			wantCt: &contract.Contract{
+				Generation: 0,
+				Resources: []*contract.Resource{
+					{
+						Egresses: []*contract.Egress{
+							{
+								Uid: "abc",
+							},
+							{
+								Uid: "abc",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := AddOrUpdateEgressConfig(tt.givenCt, tt.brokerIndex, tt.egress, tt.egressIndex); got != tt.changed {
+				t.Errorf("AddOrUpdateEgressConfig() = %v, want %v", got, tt.changed)
+			}
+
+			if diff := cmp.Diff(tt.wantCt, tt.givenCt, protocmp.Transform()); diff != "" {
+				t.Errorf("(-want, +got) %s", diff)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/core/config/resource_test.go b/control-plane/pkg/core/config/resource_test.go
index 0f593bd83..adc5b2a69 100644
--- a/control-plane/pkg/core/config/resource_test.go
+++ b/control-plane/pkg/core/config/resource_test.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/testing/protocmp"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	"github.com/google/go-cmp/cmp"
@@ -85,6 +86,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 		newResource  *contract.Resource
 		index        int
 		wantContract *contract.Contract
+		changed      int
 	}{
 		{
 			name: "resource not found - add resource",
@@ -277,14 +279,101 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 				Generation: 1,
 			},
 		},
+		{
+			name: "resource found - update resource - unchanged",
+			haveContract: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Uid:    "1",
+						Topics: []string{"topic-name-1"},
+						Egresses: []*contract.Egress{
+							{
+								Filter: &contract.Filter{
+									Attributes: map[string]string{
+										"source": "source1",
+									},
+								},
+								Destination:   "http://localhost:8080",
+								ConsumerGroup: "egress-1",
+								Uid:           "egress-1",
+							},
+						},
+						BootstrapServers: "broker:9092",
+						Ingress: &contract.Ingress{
+							IngressType: &contract.Ingress_Path{
+								Path: "/broker-ns/broker-name",
+							},
+							ContentMode: contract.ContentMode_STRUCTURED,
+						},
+					},
+				},
+				Generation: 1,
+			},
+			newResource: &contract.Resource{
+				Uid:    "1",
+				Topics: []string{"topic-name-1"},
+				Egresses: []*contract.Egress{
+					{
+						Filter: &contract.Filter{
+							Attributes: map[string]string{
+								"source": "source1",
+							},
+						},
+						Destination:   "http://localhost:8080",
+						ConsumerGroup: "egress-1",
+						Uid:           "egress-1",
+					},
+				},
+				BootstrapServers: "broker:9092",
+				Ingress: &contract.Ingress{
+					IngressType: &contract.Ingress_Path{
+						Path: "/broker-ns/broker-name",
+					},
+					ContentMode: contract.ContentMode_STRUCTURED,
+				},
+			},
+			index:   0,
+			changed: ResourceUnchanged,
+			wantContract: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Uid:    "1",
+						Topics: []string{"topic-name-1"},
+						Egresses: []*contract.Egress{
+							{
+								Filter: &contract.Filter{
+									Attributes: map[string]string{
+										"source": "source1",
+									},
+								},
+								Destination:   "http://localhost:8080",
+								ConsumerGroup: "egress-1",
+								Uid:           "egress-1",
+							},
+						},
+						BootstrapServers: "broker:9092",
+						Ingress: &contract.Ingress{
+							IngressType: &contract.Ingress_Path{
+								Path: "/broker-ns/broker-name",
+							},
+							ContentMode: contract.ContentMode_STRUCTURED,
+						},
+					},
+				},
+				Generation: 1,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			AddOrUpdateResourceConfig(tt.haveContract, tt.newResource, tt.index, zap.NewNop())
+			changed := AddOrUpdateResourceConfig(tt.haveContract, tt.newResource, tt.index, zap.NewNop())
 
 			if diff := cmp.Diff(tt.wantContract, tt.haveContract, protocmp.Transform()); diff != "" {
 				t.Errorf("(-want, +got) %s", diff)
 			}
+			if changed != tt.changed {
+				t.Errorf("Changed want %d got %d", tt.changed, changed)
+			}
 		})
 	}
 }
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 2c54a1899..19314b41e 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1363,6 +1363,53 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				BootstrapServersConfigMapKey: bootstrapServers,
 			},
 		},
+		{
+			Name: "Reconciled normal - unchanged",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig:     &contract.EgressConfig{DeadLetter: "http://test-service.test-service-namespace.svc.cluster.local/"},
+						},
+					},
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 841c6f382..d03276798 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -635,6 +635,60 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - unchanged",
+			Objects: []runtime.Object{
+				NewSink(
+					SinkControllerOwnsTopic,
+					BootstrapServers(bootstrapServersArr),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
+							Topics:  []string{"my-existing-topic-a"},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+						},
+						{
+							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
+							Topics: []string{"my-existing-topic-b"},
+						},
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+				}, &configs),
+				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "2",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						SinkControllerOwnsTopic,
+						InitSinkConditions,
+						SinkDataPlaneAvailable,
+						SinkConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						SinkAddressable(&configs.Env),
+					),
+				},
+			},
+		},
 	}
 
 	for i := range table {
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 1654ede64..3413e0fd7 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -507,31 +507,7 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
-			WantUpdates: []clientgotesting.UpdateActionImpl{
-				ConfigMapUpdate(&configs, &contract.Contract{
-					Resources: []*contract.Resource{
-						{
-							Uid:     BrokerUUID,
-							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
-							Egresses: []*contract.Egress{
-								{
-									Destination:   ServiceURL,
-									ConsumerGroup: TriggerUUID,
-									Uid:           TriggerUUID,
-									Filter: &contract.Filter{Attributes: map[string]string{
-										"type": "type1",
-									}},
-								},
-							},
-						},
-					},
-					Generation: 1,
-				}),
-				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
-				}),
-			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{},
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: newTrigger(
@@ -980,6 +956,9 @@ func triggerReconciliation(t *testing.T, format string, configs broker.Configs)
 									Destination:   ServiceURL,
 									ConsumerGroup: TriggerUUID,
 									Uid:           TriggerUUID,
+									Filter: &contract.Filter{Attributes: map[string]string{
+										"source": "source2",
+									}},
 								},
 								{
 									Destination:   "http://example.com/3",