Repository: knative-sandbox/eventing-kafka-broker (go)

Issue:
Contract generation not incremented on Broker and KafkaSink deletion
**Describe the bug**

[Grep `1fac903b-0f2a-4505-acbd-c1dac5e7d417` in logs](https://00f74ba44b97630c0136e935b18fb892569dbad482-apidata.googleusercontent.com/download/storage/v1/b/knative-prow/o/logs%2Fci-knative-sandbox-eventing-kafka-broker-continuous%2F1357312941266833408%2Fartifacts%2Fe200735b-66e8-11eb-bbf2-6679e3cb451f%2Fknative-eventing-logs%2Fknative-eventing%2Ftest-kafka-broker-cm-watcher-9c847d755-9ppcq-test-kafka-broker-cm-watcher.log?jk=AFshE3WI0uzZaHVJLr260z8Tky4FHjO_TTNFjPAqQ0iFNx868g8UgQ8TEHsmppdNtLaPGigcoySyWXyxcmgrhe-Pe7Du2ZKHlw11ACr-SO8gmTmjAXfxLIx5Vj1PqNWISwDr6Ei4ErfJupeS4DmGtcwA54-vvIwsrRhY2McXoRd74lgj8U4ANEIdMlxl9vndWblKQ1sIY3UO0ALkjpB24nPttW874wJoAus5R_uC8tHUKxrGUcCPvnJdh2fk32Imho2wy8ieawAaAubQdXkc1YnJ2MufgjWKv3hCJ5T52VKae28WnLFfoq0cfwOWjBuy7SlS6pcABgxOX54KgC2OYDKD4wjAU00RJOSH5RM7LRLA8jcSVaOgv6OthDr3ZVNP-qK2z6q6QB4ZBbpKJAkZxRhxtxmYG9RQ-LyJyak_mu2v2_EDlvJPguefmOOm_0HMlWi2p3M2J58rDRxDRsgLJ5M2u-Cp5RPcgtw7An2fQAjJ1nckvXe5IywbEnZmWJoo4kw5kPvKMeDfUxKyu2NEroOOlvt2cB4d8CxkT-Miz9HcR-DEsxUhemJyWdz4XkNxPAQBNhXZo88lGhcBz29a7U-InqVsnck6_dDV5IK5qADqG8B9LUvnR_Kx_-nwFOa32M1S1V37tIrm-7zfFVwNVoNqYayHA8E_qPDhx8cgWzenVjeML-wQQekxvlcuP2e2alUYGE5I2tpnm5x9Jx1ZWsNyfz9l1wx-JHHOjY7KFIk4-7QAF824-TcJgqQgG1XqT7G0nxvQLCVrXxD0hFMQfcyrUkVxmVNorJz-oX8_tYewXzjDCVXGPqohbMFWKovygB7dIG7ojMxgkE_Q2DU3dvbXfKp_qIweOBEsna-cTQ0dDphligrH9gxsTTSkm7v53otKCy5Hti26LppbKmW0-qsiIPc4ENiZIuFcNCzgn4NmjMDmCjpWEEHbmUwZom45YWkhYEQl4JS2bny-yCq7295RvPiMUqSHhCcSUYBHMeIiQ_K6if6nYGof0NrAnQzM3wpH_v6eoOSx0CM1c0WxGotrvAWfD0JLxP1Qh6k3_Kg4hjZ77ewgTsF3bTPjRy7pFH4mwBZJpMAXF-nopHWjiPCa0ty68bOrODvJsuBelUGekk_s-Oixw2Zd5_VVE--PtQZ0DK0JxGXl-wkTcg16yhWpT4OAo2c4PYaNuWV4M1w232F6ynvkkP6mULMYbNkTLECPXbxbY--km93gPJ0Ev-uFQAeG&isca=1) 

 **Expected behavior**

Contract incremented on broker and KafkaSink deletion.

 **To Reproduce**

Delete a Broker.

 **Knative release version**
`<= devel`

 **Additional context**




Pull request description:
Increment contract generation on Broker and KafkaSink finalize
Fixes #594

## Proposed Changes

- Increment contract generation on Broker and KafkaSink finalize
- Use a single function to increment generation

Patch:
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index f20e3e491..c2529b0cf 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -80,3 +80,8 @@ func BackoffDelayFromISO8601String(backoffDelay *string, defaultDelay uint64) (u
 
 	return uint64(math.Abs(float64(ms.Milliseconds()))), nil
 }
+
+// Increment contract.Contract.Generation.
+func IncrementContractGeneration(ct *contract.Contract) {
+	ct.Generation = (ct.Generation + 1) % (math.MaxUint64 - 1)
+}
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index abe9acacd..71779b8e9 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -19,7 +19,6 @@ package broker
 import (
 	"context"
 	"fmt"
-	"math"
 	"strings"
 	"sync"
 
@@ -164,8 +163,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	logger.Debug("Change detector", zap.Int("changed", changed))
 
-	// Increment volumeGeneration
-	ct.Generation = incrementContractGeneration(ct.Generation)
+	coreconfig.IncrementContractGeneration(ct)
 
 	if changed == coreconfig.ResourceChanged {
 		// Update the configuration map with the new contract data.
@@ -251,6 +249,8 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 		logger.Debug("Broker deleted", zap.Int("index", brokerIndex))
 
+		coreconfig.IncrementContractGeneration(ct)
+
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			return err
@@ -283,10 +283,6 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	return nil
 }
 
-func incrementContractGeneration(generation uint64) uint64 {
-	return (generation + 1) % (math.MaxUint64 - 1)
-}
-
 func (r *Reconciler) topicConfig(logger *zap.Logger, broker *eventing.Broker) (*kafka.TopicConfig, *corev1.ConfigMap, error) {
 
 	logger.Debug("broker config", zap.Any("broker.spec.config", broker.Spec.Config))
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 46699529d..008c684ae 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -19,7 +19,6 @@ package sink
 import (
 	"context"
 	"fmt"
-	"math"
 
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
@@ -168,8 +167,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	// Update contract data with the new sink configuration.
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, sinkConfig, sinkIndex, logger)
 
-	// Increment volumeGeneration
-	ct.Generation = incrementGeneration(ct.Generation)
+	coreconfig.IncrementContractGeneration(ct)
 
 	if changed == coreconfig.ResourceChanged {
 		// Update the configuration map with the new contract data.
@@ -234,6 +232,8 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 
 		logger.Debug("Sink deleted", zap.Int("index", sinkIndex))
 
+		coreconfig.IncrementContractGeneration(ct)
+
 		// Update the configuration map with the new contract data.
 		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			return err
@@ -257,10 +257,6 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 	return nil
 }
 
-func incrementGeneration(generation uint64) uint64 {
-	return (generation + 1) % (math.MaxUint64 - 1)
-}
-
 func topicConfigFromSinkSpec(kss *eventing.KafkaSinkSpec) *kafka.TopicConfig {
 	return &kafka.TopicConfig{
 		TopicDetail: sarama.TopicDetail{
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index eecf88a7d..ed89748b5 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -19,7 +19,6 @@ package trigger
 import (
 	"context"
 	"fmt"
-	"math"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
@@ -119,7 +118,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, trigger *eventing.Trigger
 	ct.Resources[brokerIndex].Egresses = deleteTrigger(egresses, triggerIndex)
 
 	// Increment volume generation
-	ct.Generation = incrementGeneration(ct.Generation)
+	coreconfig.IncrementContractGeneration(ct)
 
 	// Update data plane config map.
 	err = r.UpdateDataPlaneConfigMap(ctx, ct, dataPlaneConfigMap)
@@ -256,8 +255,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 
 	changed := coreconfig.AddOrUpdateEgressConfig(ct, brokerIndex, triggerConfig, triggerIndex)
 
-	// Increment volumeGeneration
-	ct.Generation = incrementGeneration(ct.Generation)
+	coreconfig.IncrementContractGeneration(ct)
 
 	if changed == coreconfig.EgressChanged {
 		// Update the configuration map with the new dataPlaneConfig data.
@@ -291,7 +289,3 @@ func isOurBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass
 }
-
-func incrementGeneration(generation uint64) uint64 {
-	return (generation + 1) % (math.MaxUint64 - 1)
-}
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 040dc98d1..95805d3f6 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -42,7 +42,7 @@
     <mycila.license.plugin.version>4.0.rc2</mycila.license.plugin.version>
 
     <!-- dependencies version -->
-    <vertx.version>4.0.0</vertx.version>
+    <vertx.version>4.0.1</vertx.version>
     <cloudevents.sdk.version>2.0.0.RC2</cloudevents.sdk.version>
     <micrometer.version>1.6.3</micrometer.version>
     <opentelemetry.version>0.15.0</opentelemetry.version>
diff --git a/go.mod b/go.mod
index e402baa21..df661c70a 100644
--- a/go.mod
+++ b/go.mod
@@ -23,7 +23,7 @@ require (
 	k8s.io/apiserver v0.19.7
 	k8s.io/client-go v0.19.7
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.20.1-0.20210201175433-d58bd38861b3
-	knative.dev/hack v0.0.0-20210120165453-8d623a0af457
-	knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6
+	knative.dev/eventing v0.20.1-0.20210204140212-817144887919
+	knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
+	knative.dev/pkg v0.0.0-20210203171706-6045ed499615
 )
diff --git a/go.sum b/go.sum
index 083d43952..d53534117 100644
--- a/go.sum
+++ b/go.sum
@@ -1259,14 +1259,18 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73 h1:uJmqzgNWG7XyClnU/mLPBWwfKKF1K
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.20.1-0.20210201175433-d58bd38861b3 h1:sVzkla+ECdBhsp70h3NT9RICyUaspORKGB4WbTafunQ=
-knative.dev/eventing v0.20.1-0.20210201175433-d58bd38861b3/go.mod h1:rWfYu3gwM/8Zu+2kInU2iYk9woIDnLAGhzxQCT5eNow=
+knative.dev/eventing v0.20.1-0.20210204140212-817144887919 h1:BF2OHqr/IpSxkGgjHxC4PfH90SnZwaBwMPy9DkD66Qw=
+knative.dev/eventing v0.20.1-0.20210204140212-817144887919/go.mod h1:rWfYu3gwM/8Zu+2kInU2iYk9woIDnLAGhzxQCT5eNow=
 knative.dev/hack v0.0.0-20210120165453-8d623a0af457 h1:jEBITgx/lQydGncM0uetpv/ZqawRzb2aSfEaYoMeDjM=
 knative.dev/hack v0.0.0-20210120165453-8d623a0af457/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf h1:u4cY4jr2LYvhoz/1HBWEPsMiLkm0HMdDTfmmw1RE8zE=
+knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/pkg v0.0.0-20210125222030-6040b3af4803 h1:L9lY/UztepeiTWys3VyO3Y5wQmYPyYG7eOG9qib6AZY=
 knative.dev/pkg v0.0.0-20210125222030-6040b3af4803/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
 knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6 h1:HIACRvhv/4U7vcFTAakfqYJIlENCSGtTrZOwz/q/A00=
 knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
+knative.dev/pkg v0.0.0-20210203171706-6045ed499615 h1:Ee0n+axzrQXWkJfGl91B5RRmB25R1plTfC7MDw3NxUU=
+knative.dev/pkg v0.0.0-20210203171706-6045ed499615/go.mod h1:X4NPrCo8NK3hbDVan9Vm7mf5io3ZoINakAdrpSXVB08=
 knative.dev/reconciler-test v0.0.0-20210127151131-28c3852c3e9b/go.mod h1:+IqX+lV5lQ6AhHY7fomh3sn6K2kW1cTQvC7US5cJB7k=
 modernc.org/cc v1.0.0/go.mod h1:1Sk4//wdnYJiUIxnW8ddKpaOJCF37yAdqYnkxUpaYxw=
 modernc.org/golex v1.0.0/go.mod h1:b/QX9oBD/LhixY6NDh+IdGv17hgB+51fET1i2kPSmvk=
diff --git a/vendor/knative.dev/hack/library.sh b/vendor/knative.dev/hack/library.sh
index cf56962a8..25b3407e8 100644
--- a/vendor/knative.dev/hack/library.sh
+++ b/vendor/knative.dev/hack/library.sh
@@ -539,7 +539,8 @@ function go_update_deps() {
 
   export GO111MODULE=on
   export GOFLAGS=""
-  export GOSUMDB=off   # Do not use the sum.golang.org service.
+  export GONOSUMDB="${GONOSUMDB:-},knative.dev/*"
+  export GONOPROXY="${GONOPROXY:-},knative.dev/*"
 
   echo "=== Update Deps for Golang"
 
@@ -559,16 +560,6 @@ function go_update_deps() {
 
   if [[ $UPGRADE == 1 ]]; then
     group "Upgrading to ${VERSION}"
-    # From shell parameter expansion:
-    # ${parameter:+word}
-    # If parameter is null or unset, nothing is substituted, otherwise the expansion of word is substituted.
-    # -z is if the length of the string, so skip setting GOPROXY if GOPROXY is already set.
-    if [[ -z ${GOPROXY:+skip} ]]; then
-      export GOPROXY=direct
-      echo "Using 'GOPROXY=direct'."
-    else
-      echo "Respecting 'GOPROXY=${GOPROXY}'."
-    fi
     FLOATING_DEPS+=( $(run_go_tool knative.dev/test-infra/buoy buoy float ${REPO_ROOT_DIR}/go.mod --release ${VERSION} --domain ${DOMAIN}) )
     if [[ ${#FLOATING_DEPS[@]} > 0 ]]; then
       echo "Floating deps to ${FLOATING_DEPS[@]}"
diff --git a/vendor/knative.dev/pkg/network/h2c.go b/vendor/knative.dev/pkg/network/h2c.go
index 73fabee97..f950b9c34 100644
--- a/vendor/knative.dev/pkg/network/h2c.go
+++ b/vendor/knative.dev/pkg/network/h2c.go
@@ -41,8 +41,13 @@ func NewServer(addr string, h http.Handler) *http.Server {
 // to explicitly allow h2c (http2 without TLS) transport.
 // See https://github.com/golang/go/issues/14141 for more details.
 func NewH2CTransport() http.RoundTripper {
+	return newH2CTransport(false)
+}
+
+func newH2CTransport(disableCompression bool) http.RoundTripper {
 	return &http2.Transport{
-		AllowHTTP: true,
+		AllowHTTP:          true,
+		DisableCompression: disableCompression,
 		DialTLS: func(netw, addr string, _ *tls.Config) (net.Conn, error) {
 			return DialWithBackOff(context.Background(),
 				netw, addr)
diff --git a/vendor/knative.dev/pkg/network/transports.go b/vendor/knative.dev/pkg/network/transports.go
index c37b99d57..26ce82395 100644
--- a/vendor/knative.dev/pkg/network/transports.go
+++ b/vendor/knative.dev/pkg/network/transports.go
@@ -94,13 +94,14 @@ func dialBackOffHelper(ctx context.Context, network, address string, bo wait.Bac
 	return nil, fmt.Errorf("timed out dialing after %.2fs", elapsed.Seconds())
 }
 
-func newHTTPTransport(disableKeepAlives bool, maxIdle, maxIdlePerHost int) http.RoundTripper {
+func newHTTPTransport(disableKeepAlives, disableCompression bool, maxIdle, maxIdlePerHost int) http.RoundTripper {
 	transport := http.DefaultTransport.(*http.Transport).Clone()
 	transport.DialContext = DialWithBackOff
 	transport.DisableKeepAlives = disableKeepAlives
 	transport.MaxIdleConns = maxIdle
 	transport.MaxIdleConnsPerHost = maxIdlePerHost
 	transport.ForceAttemptHTTP2 = false
+	transport.DisableCompression = disableCompression
 	return transport
 }
 
@@ -108,7 +109,7 @@ func newHTTPTransport(disableKeepAlives bool, maxIdle, maxIdlePerHost int) http.
 // since it will not cache connections.
 func NewProberTransport() http.RoundTripper {
 	return newAutoTransport(
-		newHTTPTransport(true /*disable keep-alives*/, 0, 0 /*no caching*/),
+		newHTTPTransport(true /*disable keep-alives*/, false /*disable auto-compression*/, 0, 0 /*no caching*/),
 		NewH2CTransport())
 }
 
@@ -116,8 +117,17 @@ func NewProberTransport() http.RoundTripper {
 // based on the request's HTTP version.
 func NewAutoTransport(maxIdle, maxIdlePerHost int) http.RoundTripper {
 	return newAutoTransport(
-		newHTTPTransport(false /*disable keep-alives*/, maxIdle, maxIdlePerHost),
-		NewH2CTransport())
+		newHTTPTransport(false /*disable keep-alives*/, false /*disable auto-compression*/, maxIdle, maxIdlePerHost),
+		newH2CTransport(false /*disable auto-compression*/))
+}
+
+// NewProxyAutoTransport creates a RoundTripper suitable for use by a reverse
+// proxy.  The returned transport uses HTTP or H2C based on the request's HTTP
+// version. The transport has DisableCompression set to true.
+func NewProxyAutoTransport(maxIdle, maxIdlePerHost int) http.RoundTripper {
+	return newAutoTransport(
+		newHTTPTransport(false /*disable keep-alives*/, true /*disable auto-compression*/, maxIdle, maxIdlePerHost),
+		newH2CTransport(true /*disable auto-compression*/))
 }
 
 // AutoTransport uses h2c for HTTP2 requests and falls back to `http.DefaultTransport` for all others
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 16e6858e3..be8183030 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1067,7 +1067,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.20.1-0.20210201175433-d58bd38861b3
+# knative.dev/eventing v0.20.1-0.20210204140212-817144887919
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/configs
@@ -1197,10 +1197,10 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/hack v0.0.0-20210120165453-8d623a0af457
+# knative.dev/hack v0.0.0-20210203173706-8368e1f6eacf
 ## explicit
 knative.dev/hack
-# knative.dev/pkg v0.0.0-20210130001831-ca02ef752ac6
+# knative.dev/pkg v0.0.0-20210203171706-6045ed499615
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index 11c2b57a3..3bdf62bdc 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -17,6 +17,7 @@
 package config
 
 import (
+	"math"
 	"testing"
 
 	"k8s.io/utils/pointer"
@@ -168,3 +169,42 @@ func TestBackoffDelayFromString(t *testing.T) {
 		})
 	}
 }
+
+func TestIncrementContractGeneration(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		generation uint64
+		expected   uint64
+	}{
+		{
+			name:       "min",
+			generation: 0,
+			expected:   1,
+		},
+		{
+			name:       "max",
+			generation: math.MaxUint64 - 2,
+			expected:   0,
+		},
+		{
+			name:       "random",
+			generation: 42,
+			expected:   43,
+		},
+	}
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			ct := &contract.Contract{
+				Generation: tt.generation,
+			}
+			IncrementContractGeneration(ct)
+
+			if ct.Generation != tt.expected {
+				t.Errorf("Got %d expected %d", ct.Generation, tt.expected)
+			}
+		})
+	}
+}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 973e25e71..f869cb06b 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -1533,7 +1533,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &contract.Contract{
 					Resources:  []*contract.Resource{},
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 			OtherTestData: map[string]interface{}{
@@ -1561,7 +1561,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 			Key: testKey,
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &contract.Contract{
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 			OtherTestData: map[string]interface{}{
@@ -1595,7 +1595,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
 				ConfigMapUpdate(&configs, &contract.Contract{
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 			OtherTestData: map[string]interface{}{
@@ -1652,7 +1652,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-					Generation: 5,
+					Generation: 6,
 				}),
 			},
 			OtherTestData: map[string]interface{}{
@@ -1689,7 +1689,7 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
-					Generation: 5,
+					Generation: 6,
 				}),
 			},
 			OtherTestData: map[string]interface{}{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 95891d20b..b39980929 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -830,7 +830,7 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 							BootstrapServers: bootstrapServers,
 						},
 					},
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 		},
@@ -884,7 +884,7 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 							BootstrapServers: bootstrapServers,
 						},
 					},
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 		},
@@ -950,7 +950,7 @@ func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
 							BootstrapServers: bootstrapServers,
 						},
 					},
-					Generation: 1,
+					Generation: 2,
 				}),
 			},
 			OtherTestData: map[string]interface{}{