Repository: knative-sandbox/eventing-kafka-broker
Pull request #188 (2020-09-18 05:28:12)

[data-plane]: Bump vertx.version from 4.0.0.Beta2 to 4.0.0.Beta3 in /data-plane
Bumps `vertx.version` from 4.0.0.Beta2 to 4.0.0.Beta3.
Updates `vertx-stack-depchain` from 4.0.0.Beta2 to 4.0.0.Beta3

Updates `vertx-junit5` from 4.0.0.Beta2 to 4.0.0.Beta3


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/188#" title="Author self-approved">dependabot[bot]</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **grantr**
You can assign the PR to them by writing `/assign @grantr` in a comment when ready.

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).

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/master/OWNERS)**

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":["grantr"]} -->
knative-prow-robot: Hi @dependabot[bot]. Thanks for your PR.

I'm waiting for a [knative-sandbox](https://github.com/orgs/knative-sandbox/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://github.com/orgs/knative-sandbox/people) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details>

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


Patch:
diff --git a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_lifecycle.go b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_lifecycle.go
index 2badc77ed..bdd6e5814 100644
--- a/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_lifecycle.go
+++ b/control-plane/pkg/apis/eventing/v1alpha1/kafka_sink_lifecycle.go
@@ -19,16 +19,14 @@ package v1alpha1
 import "knative.dev/pkg/apis"
 
 const (
-	ConditionAddressable      apis.ConditionType = "Addressable"
-	ConditionTopicReady       apis.ConditionType = "TopicReady"
-	ConditionConfigMapUpdated apis.ConditionType = "ConfigMapUpdated"
+	ConditionAddressable apis.ConditionType = "Addressable"
 )
 
-var conditionSet = apis.NewLivingConditionSet(
-	ConditionAddressable,
-	ConditionTopicReady,
-	ConditionConfigMapUpdated,
-)
+var conditionSet apis.ConditionSet
+
+func RegisterConditionSet(cs apis.ConditionSet) {
+	conditionSet = cs
+}
 
 func (ks *KafkaSink) GetConditionSet() apis.ConditionSet {
 	return conditionSet
diff --git a/control-plane/pkg/core/config/resource.go b/control-plane/pkg/core/config/resource.go
index d6b12ba1b..9026fa4e1 100644
--- a/control-plane/pkg/core/config/resource.go
+++ b/control-plane/pkg/core/config/resource.go
@@ -54,3 +54,19 @@ func AddOrUpdateResourceConfig(contract *contract.Contract, resource *contract.R
 		logger.Debug("Resource doesn't exist")
 	}
 }
+
+// DeleteResource deletes the resource at the given index from Resources.
+func DeleteResource(ct *contract.Contract, index int) {
+
+	if len(ct.Resources) == 1 {
+		*ct = contract.Contract{
+			Generation: ct.Generation,
+		}
+		return
+	}
+
+	// replace the resource to be deleted with the last one.
+	ct.Resources[index] = ct.Resources[len(ct.Resources)-1]
+	// truncate the array.
+	ct.Resources = ct.Resources[:len(ct.Resources)-1]
+}
diff --git a/control-plane/pkg/core/config/utils.go b/control-plane/pkg/core/config/utils.go
index ef2b5881b..efc53dbe7 100644
--- a/control-plane/pkg/core/config/utils.go
+++ b/control-plane/pkg/core/config/utils.go
@@ -18,6 +18,7 @@ package config
 
 import (
 	"fmt"
+
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index 2d9c21f0a..a0e355ec1 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -48,8 +48,12 @@ var ConditionSet = apis.NewLivingConditionSet(
 )
 
 const (
+	TopicOwnerAnnotation = "eventing.knative.dev/topic.owner"
+
 	ReasonDataPlaneNotAvailable  = "Data plane not available"
 	MessageDataPlaneNotAvailable = "Did you install the data plane for this component?"
+
+	ReasonTopicNotPresent = "Topic is not present"
 )
 
 type Object interface {
@@ -145,7 +149,17 @@ func (manager *StatusConditionManager) FailedToCreateTopic(topic string, err err
 	return fmt.Errorf("failed to create topic: %s: %w", topic, err)
 }
 
-func (manager *StatusConditionManager) TopicCreated(topic string) {
+func (manager *StatusConditionManager) TopicReady(topic string) {
+
+	if owner, ok := manager.Object.GetStatus().Annotations[TopicOwnerAnnotation]; ok {
+		manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrueWithReason(
+			ConditionTopicReady,
+			fmt.Sprintf("Topic %s (owner %s)", topic, owner),
+			"",
+		)
+
+		return
+	}
 
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrueWithReason(
 		ConditionTopicReady,
@@ -206,3 +220,25 @@ func (manager *StatusConditionManager) FailedToResolveConfig(err error) reconcil
 func (manager *StatusConditionManager) ConfigResolved() {
 	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkTrue(ConditionConfigParsed)
 }
+
+func (manager *StatusConditionManager) TopicNotPresentOrInvalidErr(err error) error {
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		ConditionTopicReady,
+		ReasonTopicNotPresent,
+		err.Error(),
+	)
+
+	return fmt.Errorf("topic is not present: %w", err)
+}
+
+func (manager *StatusConditionManager) TopicNotPresentOrInvalid() error {
+
+	manager.Object.GetConditionSet().Manage(manager.Object.GetStatus()).MarkFalse(
+		ConditionTopicReady,
+		ReasonTopicNotPresent,
+		"Check topic configuration",
+	)
+
+	return fmt.Errorf("topic is not present: check topic configuration")
+
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index d55113f95..f79bfbebd 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -19,6 +19,7 @@ package base
 import (
 	"context"
 	"fmt"
+
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index b428946f5..ee65ec76c 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -19,11 +19,12 @@ package broker
 import (
 	"context"
 	"fmt"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"math"
 	"strings"
 	"sync"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
@@ -106,7 +107,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	if err != nil {
 		return statusConditionManager.FailedToCreateTopic(topic, err)
 	}
-	statusConditionManager.TopicCreated(topic)
+	statusConditionManager.TopicReady(topic)
 
 	logger.Debug("Topic created", zap.Any("topic", topic))
 
@@ -119,14 +120,14 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	logger.Debug("Got contract config map")
 
 	// Get contract data.
-	contract, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
-	if err != nil && contract == nil {
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil && ct == nil {
 		return statusConditionManager.FailedToGetDataFromConfigMap(err)
 	}
 
 	logger.Debug(
 		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(contract)),
+		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(ct)),
 	)
 
 	// Get resource configuration.
@@ -135,15 +136,15 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.FailedToGetConfig(err)
 	}
 
-	brokerIndex := coreconfig.FindResource(contract, broker.UID)
+	brokerIndex := coreconfig.FindResource(ct, broker.UID)
 	// Update contract data with the new contract configuration
-	coreconfig.AddOrUpdateResourceConfig(contract, brokerResource, brokerIndex, logger)
+	coreconfig.AddOrUpdateResourceConfig(ct, brokerResource, brokerIndex, logger)
 
 	// Increment volumeGeneration
-	contract.Generation = incrementContractGeneration(contract.Generation)
+	ct.Generation = incrementContractGeneration(ct.Generation)
 
 	// Update the configuration map with the new contract data.
-	if err := r.UpdateDataPlaneConfigMap(ctx, contract, contractConfigMap); err != nil {
+	if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 		logger.Error("failed to update data plane config map", zap.Error(
 			statusConditionManager.FailedToUpdateConfigMap(err),
 		))
@@ -160,7 +161,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	// 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, contract.Generation); err != nil {
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
 		logger.Error("Failed to update receiver pod annotation", zap.Error(
 			statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
 		))
@@ -170,7 +171,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	logger.Debug("Updated receiver pod annotation")
 
 	// Update volume generation annotation of dispatcher pods
-	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, contract.Generation); err != nil {
+	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.
@@ -206,24 +207,24 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 	logger.Debug("Got contract config map")
 
 	// Get contract data.
-	contract, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
 	if err != nil {
 		return fmt.Errorf("failed to get contract: %w", err)
 	}
 
 	logger.Debug(
 		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(contract)),
+		zap.Any(base.ContractLogKey, (*log.ContractMarshaller)(ct)),
 	)
 
-	brokerIndex := coreconfig.FindResource(contract, broker.UID)
+	brokerIndex := coreconfig.FindResource(ct, broker.UID)
 	if brokerIndex != coreconfig.NoResource {
-		deleteBroker(contract, brokerIndex)
+		coreconfig.DeleteResource(ct, brokerIndex)
 
 		logger.Debug("Broker deleted", zap.Int("index", brokerIndex))
 
 		// Update the configuration map with the new contract data.
-		if err := r.UpdateDataPlaneConfigMap(ctx, contract, contractConfigMap); err != nil {
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
 			return err
 		}
 
@@ -369,20 +370,6 @@ func (r *Reconciler) SetDefaultTopicDetails(topicDetail sarama.TopicDetail) {
 	r.KafkaDefaultTopicDetails = topicDetail
 }
 
-func deleteBroker(c *contract.Contract, index int) {
-	if len(c.Resources) == 1 {
-		*c = contract.Contract{
-			Generation: c.Generation,
-		}
-		return
-	}
-
-	// replace the resource to be deleted with the last one.
-	c.Resources[index] = c.Resources[len(c.Resources)-1]
-	// truncate the array.
-	c.Resources = c.Resources[:len(c.Resources)-1]
-}
-
 func (r *Reconciler) getDefaultBootstrapServersOrFail() ([]string, error) {
 	r.bootstrapServersLock.RLock()
 	defer r.bootstrapServersLock.RUnlock()
diff --git a/control-plane/pkg/reconciler/kafka/topic.go b/control-plane/pkg/reconciler/kafka/topic.go
index 6e9ee2499..ac25f6155 100644
--- a/control-plane/pkg/reconciler/kafka/topic.go
+++ b/control-plane/pkg/reconciler/kafka/topic.go
@@ -110,3 +110,23 @@ func (f NewClusterAdminFunc) DeleteTopic(topic string, bootstrapServers []string
 
 	return DeleteTopic(kafkaClusterAdmin, topic)
 }
+
+func (f NewClusterAdminFunc) IsTopicPresentAndValid(topic string, bootstrapServers []string) (bool, error) {
+
+	kafkaClusterAdmin, err := GetClusterAdmin(f, bootstrapServers)
+	if err != nil {
+		return false, err
+	}
+	defer kafkaClusterAdmin.Close()
+
+	metadata, err := kafkaClusterAdmin.DescribeTopics([]string{topic})
+	if err != nil {
+		return false, fmt.Errorf("failed to describe topic %s: %w", topic, err)
+	}
+
+	return isValidSingleTopicMetadata(metadata, topic), nil
+}
+
+func isValidSingleTopicMetadata(metadata []*sarama.TopicMetadata, topic string) bool {
+	return len(metadata) == 1 && metadata[0].Name == topic && !metadata[0].IsInternal
+}
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index 121bbe642..298157afb 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -28,6 +28,7 @@ import (
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/logging"
 
+	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	sinkinformer "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/informers/eventing/v1alpha1/kafkasink"
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
@@ -36,6 +37,8 @@ import (
 
 func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
 
+	eventing.RegisterConditionSet(base.ConditionSet)
+
 	logger := logging.FromContext(ctx)
 
 	reconciler := &Reconciler{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 49ca3d789..af8668788 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -18,9 +18,11 @@ package sink
 
 import (
 	"context"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"fmt"
 	"math"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"github.com/Shopify/sarama"
 	"go.uber.org/zap"
 	corelisters "k8s.io/client-go/listers/core/v1"
@@ -37,6 +39,11 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
 )
 
+const (
+	ExternalTopicOwner   = "external"
+	ControllerTopicOwner = "kafkasink-controller"
+)
+
 type Reconciler struct {
 	*base.Reconciler
 
@@ -71,21 +78,36 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	}
 	statusConditionManager.DataPlaneAvailable()
 
+	if ks.GetStatus().Annotations == nil {
+		ks.GetStatus().Annotations = make(map[string]string, 1)
+	}
+
 	if ks.Spec.NumPartitions != nil && ks.Spec.ReplicationFactor != nil {
-		topicConfig := &kafka.TopicConfig{
-			TopicDetail: sarama.TopicDetail{
-				NumPartitions:     *ks.Spec.NumPartitions,
-				ReplicationFactor: *ks.Spec.ReplicationFactor,
-			},
-			BootstrapServers: ks.Spec.BootstrapServers,
-		}
+
+		ks.GetStatus().Annotations[base.TopicOwnerAnnotation] = ControllerTopicOwner
+
+		topicConfig := topicConfigFromSinkSpec(&ks.Spec)
 
 		topic, err := r.ClusterAdmin.CreateTopic(logger, ks.Spec.Topic, topicConfig)
 		if err != nil {
 			return statusConditionManager.FailedToCreateTopic(topic, err)
 		}
+	} else {
+
+		// If the topic is externally managed, we need to make sure that the topic exists and it's valid.
+
+		ks.GetStatus().Annotations[base.TopicOwnerAnnotation] = ExternalTopicOwner
+
+		isPresentAndValid, err := r.ClusterAdmin.IsTopicPresentAndValid(ks.Spec.Topic, ks.Spec.BootstrapServers)
+		if err != nil {
+			return statusConditionManager.TopicNotPresentOrInvalidErr(err)
+		}
+		if !isPresentAndValid {
+			// The topic might be invalid.
+			return statusConditionManager.TopicNotPresentOrInvalid()
+		}
 	}
-	statusConditionManager.TopicCreated(ks.Spec.Topic)
+	statusConditionManager.TopicReady(ks.Spec.Topic)
 
 	logger.Debug("Topic created", zap.Any("topic", ks.Spec.Topic))
 
@@ -121,6 +143,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
 	}
+	statusConditionManager.ConfigResolved()
 
 	sinkIndex := coreconfig.FindResource(ct, ks.UID)
 	// Update contract data with the new sink configuration.
@@ -161,12 +184,62 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, ks *eventing.KafkaSink) r
 
 func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) error {
 
-	// logger := log.Logger(ctx, "finalize", ks)
+	logger := log.Logger(ctx, "finalize", ks)
+
+	// Get contract config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to get contract config map %s: %w", r.Configs.DataPlaneConfigMapAsString(), err)
+	}
+
+	logger.Debug("Got contract config map")
+
+	// Get contract data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil {
+		return fmt.Errorf("failed to get contract: %w", err)
+	}
+
+	logger.Debug(
+		"Got contract data from config map",
+		zap.Any("contract", (*log.ContractMarshaller)(ct)),
+	)
+
+	sinkIndex := coreconfig.FindResource(ct, ks.UID)
+	if sinkIndex != coreconfig.NoResource {
+		coreconfig.DeleteResource(ct, sinkIndex)
+
+		logger.Debug("Sink deleted", zap.Int("index", sinkIndex))
+
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			return err
+		}
+
+		logger.Debug("Sinks config map updated")
+	}
+
+	if ks.GetStatus().Annotations[base.TopicOwnerAnnotation] == ControllerTopicOwner {
+		topic, err := r.ClusterAdmin.DeleteTopic(ks.Spec.Topic, ks.Spec.BootstrapServers)
+		if err != nil {
+			return err
+		}
+		logger.Debug("Topic deleted", zap.String("topic", topic))
+	}
 
-	// TODO implement finalizer
 	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{
+			NumPartitions:     *kss.NumPartitions,
+			ReplicationFactor: *kss.ReplicationFactor,
+		},
+		BootstrapServers: kss.BootstrapServers,
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index 8de411889..b322f3ea9 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -19,9 +19,10 @@ package trigger
 import (
 	"context"
 	"fmt"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"math"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"go.uber.org/zap"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index ab8ad2fc9..3aefccc0a 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -38,7 +38,7 @@
     <maven.enforcer.plugin.version>3.0.0-M3</maven.enforcer.plugin.version>
 
     <!-- dependencies version -->
-    <vertx.version>4.0.0.Beta2</vertx.version>
+    <vertx.version>4.0.0.Beta3</vertx.version>
     <cloudevents.sdk.version>2.0.0-milestone2</cloudevents.sdk.version>
     <jackson.version>2.11.2</jackson.version>
     <protobuf.version>3.13.0</protobuf.version>
@@ -47,7 +47,7 @@
     <net.logstash.logback.encoder.version>6.4</net.logstash.logback.encoder.version>
     <assertj.version>3.17.2</assertj.version>
     <junit.jupiter.version>5.7.0</junit.jupiter.version>
-    <mokito.junit.jupiter.version>3.5.10</mokito.junit.jupiter.version>
+    <mokito.junit.jupiter.version>3.5.11</mokito.junit.jupiter.version>
   </properties>
 
   <modules>
diff --git a/go.mod b/go.mod
index c5e291044..c0055096e 100644
--- a/go.mod
+++ b/go.mod
@@ -20,9 +20,9 @@ require (
 	k8s.io/client-go v11.0.1-0.20190805182717-6502b5e7b1b5+incompatible
 	k8s.io/kubernetes v1.14.7
 	k8s.io/utils v0.0.0-20200603063816-c1c6865ac451
-	knative.dev/eventing v0.17.1-0.20200914161402-2237b8b18606
-	knative.dev/pkg v0.0.0-20200914134901-948ac9f2ce99
-	knative.dev/test-infra v0.0.0-20200911201000-3f90e7c8f2fa
+	knative.dev/eventing v0.17.1-0.20200917130042-886594db5d01
+	knative.dev/pkg v0.0.0-20200915092441-a5e2edc217c0
+	knative.dev/test-infra v0.0.0-20200916223841-d362984af929
 )
 
 replace (
diff --git a/go.sum b/go.sum
index efda0035d..5e39b827d 100644
--- a/go.sum
+++ b/go.sum
@@ -1865,8 +1865,8 @@ k8s.io/utils v0.0.0-20200603063816-c1c6865ac451 h1:v8ud2Up6QK1lNOKFgiIVrZdMg7Mpm
 k8s.io/utils v0.0.0-20200603063816-c1c6865ac451/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/caching v0.0.0-20190719140829-2032732871ff/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
 knative.dev/caching v0.0.0-20200116200605-67bca2c83dfa/go.mod h1:dHXFU6CGlLlbzaWc32g80cR92iuBSpsslDNBWI8C7eg=
-knative.dev/eventing v0.17.1-0.20200914161402-2237b8b18606 h1:y8xwDDPD0iJ2C6BxTQY9UuI+iBqIS6up5uwUHKJVRq8=
-knative.dev/eventing v0.17.1-0.20200914161402-2237b8b18606/go.mod h1:Xbh3tPsCNXXHS3G0+OeSPduhEC7B1eCfMziB29TmUms=
+knative.dev/eventing v0.17.1-0.20200917130042-886594db5d01 h1:hWfizaxIOneFoqrrTkTjx0EWt43yPecEwFRFng6+NFg=
+knative.dev/eventing v0.17.1-0.20200917130042-886594db5d01/go.mod h1:D6q7wA33iRBz2hwTYiURGFQoWXSoJ+TQpq/0bDuCgTA=
 knative.dev/eventing-contrib v0.6.1-0.20190723221543-5ce18048c08b/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/eventing-contrib v0.11.2/go.mod h1:SnXZgSGgMSMLNFTwTnpaOH7hXDzTFtw0J8OmHflNx3g=
 knative.dev/pkg v0.0.0-20191101194912-56c2594e4f11/go.mod h1:pgODObA1dTyhNoFxPZTTjNWfx6F0aKsKzn+vaT9XO/Q=
@@ -1877,16 +1877,18 @@ knative.dev/pkg v0.0.0-20200505191044-3da93ebb24c2/go.mod h1:Q6sL35DdGs8hIQZKdaC
 knative.dev/pkg v0.0.0-20200515002500-16d7b963416f/go.mod h1:tMOHGbxtRz8zYFGEGpV/bpoTEM1o89MwYFC4YJXl3GY=
 knative.dev/pkg v0.0.0-20200528142800-1c6815d7e4c9/go.mod h1:QgNZTxnwpB/oSpNcfnLVlw+WpEwwyKAvJlvR3hgeltA=
 knative.dev/pkg v0.0.0-20200711004937-22502028e31a/go.mod h1:AqAJV6rYi8IGikDjJ/9ZQd9qKdkXVlesVnVjwx62YB8=
-knative.dev/pkg v0.0.0-20200914134901-948ac9f2ce99 h1:1FL9I9NBJEPGdabcCPbqd/aP2UqQd08PhFORy+RBaoY=
-knative.dev/pkg v0.0.0-20200914134901-948ac9f2ce99/go.mod h1:igZfLQk2QpRo+cxbH4NfsgbdavqsXr1rMr818/xVu3c=
+knative.dev/pkg v0.0.0-20200915092441-a5e2edc217c0 h1:9ETJvpaWmhEcTG/wx2syd6FbJpD2u/PGabD1CcYLf9M=
+knative.dev/pkg v0.0.0-20200915092441-a5e2edc217c0/go.mod h1:ZV1kWD8UMvELVVDld0PDRdjwdCsGwEwf1YUvGGDUl2c=
 knative.dev/test-infra v0.0.0-20200407185800-1b88cb3b45a5/go.mod h1:xcdUkMJrLlBswIZqL5zCuBFOC22WIPMQoVX1L35i0vQ=
 knative.dev/test-infra v0.0.0-20200505052144-5ea2f705bb55/go.mod h1:WqF1Azka+FxPZ20keR2zCNtiQA1MP9ZB4BH4HuI+SIU=
 knative.dev/test-infra v0.0.0-20200513011557-d03429a76034/go.mod h1:aMif0KXL4g19YCYwsy4Ocjjz5xgPlseYV+B95Oo4JGE=
 knative.dev/test-infra v0.0.0-20200519015156-82551620b0a9/go.mod h1:A5b2OAXTOeHT3hHhVQm3dmtbuWvIDP7qzgtqxA3/2pE=
 knative.dev/test-infra v0.0.0-20200707183444-aed09e56ddc7/go.mod h1:RjYAhXnZqeHw9+B0zsbqSPlae0lCvjekO/nw5ZMpLCs=
-knative.dev/test-infra v0.0.0-20200909211651-72eb6ae3c773/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
 knative.dev/test-infra v0.0.0-20200911201000-3f90e7c8f2fa h1:p1gXSaviQgkr3yKzcKO0+xW5qOtTnOPKqgFviTQ/qoM=
 knative.dev/test-infra v0.0.0-20200911201000-3f90e7c8f2fa/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
+knative.dev/test-infra v0.0.0-20200914203402-489e9e80675b/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
+knative.dev/test-infra v0.0.0-20200916223841-d362984af929 h1:jCHClMehwRQJ1BU4YPWX5eHjnJNFhYHJNsYIbJtHGZA=
+knative.dev/test-infra v0.0.0-20200916223841-d362984af929/go.mod h1:Pmg2c7Z7q7BGFUV/GOpU5BlrD3ePJft4MPqx8AYBplc=
 mvdan.cc/interfacer v0.0.0-20180901003855-c20040233aed/go.mod h1:Xkxe497xwlCKkIaQYRfC7CSLworTXY9RMqwhhCm+8Nc=
 mvdan.cc/lint v0.0.0-20170908181259-adc824a0674b/go.mod h1:2odslEg/xrtNQqCYg2/jCoyKnw3vv5biOc3JnIcYfL4=
 mvdan.cc/unparam v0.0.0-20190720180237-d51796306d8f/go.mod h1:4G1h5nDURzA3bwVMZIVpwbkw+04kSxk3rAtzlimaUJw=
diff --git a/vendor/modules.txt b/vendor/modules.txt
index bd348220e..0410b51be 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1044,7 +1044,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.17.1-0.20200914161402-2237b8b18606
+# knative.dev/eventing v0.17.1-0.20200917130042-886594db5d01
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/configs
@@ -1166,7 +1166,7 @@ knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
 knative.dev/eventing/test/test_images/sequencestepper
 knative.dev/eventing/test/test_images/transformevents
-# knative.dev/pkg v0.0.0-20200914134901-948ac9f2ce99
+# knative.dev/pkg v0.0.0-20200915092441-a5e2edc217c0
 ## explicit
 knative.dev/pkg/apis
 knative.dev/pkg/apis/duck
@@ -1253,7 +1253,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/test-infra v0.0.0-20200911201000-3f90e7c8f2fa
+# knative.dev/test-infra v0.0.0-20200916223841-d362984af929
 ## explicit
 knative.dev/test-infra/scripts
 # sigs.k8s.io/structured-merge-diff/v3 v3.0.1-0.20200706213357-43c19bbb7fba


Test patch:
diff --git a/control-plane/pkg/core/config/egress_test.go b/control-plane/pkg/core/config/egress_test.go
index c94e14ed6..6093ddf4e 100644
--- a/control-plane/pkg/core/config/egress_test.go
+++ b/control-plane/pkg/core/config/egress_test.go
@@ -17,9 +17,10 @@
 package config
 
 import (
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"k8s.io/apimachinery/pkg/types"
 )
 
diff --git a/control-plane/pkg/core/config/resource_test.go b/control-plane/pkg/core/config/resource_test.go
index 02a25eeac..b61dc9edc 100644
--- a/control-plane/pkg/core/config/resource_test.go
+++ b/control-plane/pkg/core/config/resource_test.go
@@ -17,9 +17,11 @@
 package config
 
 import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/testing/protocmp"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
@@ -279,3 +281,114 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 		})
 	}
 }
+
+func TestDeleteResource(t *testing.T) {
+
+	tests := []struct {
+		name  string
+		ct    *contract.Contract
+		index int
+		want  contract.Contract
+	}{
+		{
+			name: "1 resource",
+			ct: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+				},
+				Generation: 200,
+			},
+			index: 0,
+			want: contract.Contract{
+				Generation: 200,
+			},
+		},
+		{
+			name: "2 resource",
+			ct: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+					{
+						Id: "2",
+					},
+				},
+			},
+			index: 0,
+			want: contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "2",
+					},
+				},
+			},
+		},
+		{
+			name: "3 resource - delete last",
+			ct: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+					{
+						Id: "2",
+					},
+					{
+						Id: "3",
+					},
+				},
+			},
+			index: 2,
+			want: contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+					{
+						Id: "2",
+					},
+				},
+			},
+		},
+		{
+			name: "3 broker - middle",
+			ct: &contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+					{
+						Id: "2",
+					},
+					{
+						Id: "3",
+					},
+				},
+				Generation: 200,
+			},
+			index: 1,
+			want: contract.Contract{
+				Resources: []*contract.Resource{
+					{
+						Id: "1",
+					},
+					{
+						Id: "3",
+					},
+				},
+				Generation: 200,
+			},
+		},
+	}
+	for i := range tests {
+		t.Run(tests[i].name, func(t *testing.T) {
+
+			DeleteResource(tests[i].ct, tests[i].index)
+
+			assert.Equal(t, tests[i].ct, &tests[i].want)
+		})
+	}
+}
diff --git a/control-plane/pkg/core/config/utils_test.go b/control-plane/pkg/core/config/utils_test.go
index a73a41bf1..74c280150 100644
--- a/control-plane/pkg/core/config/utils_test.go
+++ b/control-plane/pkg/core/config/utils_test.go
@@ -17,9 +17,10 @@
 package config
 
 import (
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	eventing "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 )
 
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index c7c747e57..d8c32f55d 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -19,12 +19,13 @@ package broker_test // different package name due to import cycles. (broker -> t
 import (
 	"context"
 	"fmt"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"net/url"
 	"sync"
 	"testing"
 	"time"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/assert"
 	corev1 "k8s.io/api/core/v1"
diff --git a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
index 4bfcee5ea..9ee01a2ac 100644
--- a/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
+++ b/control-plane/pkg/reconciler/kafka/testing/admin_mock.go
@@ -23,6 +23,11 @@ type MockKafkaClusterAdmin struct {
 	ExpectedClose      bool
 	ExpectedCloseError error
 
+	// DescribeTopics
+	ExpectedTopics                         []string
+	ExpectedErrorOnDescribeTopics          error
+	ExpectedTopicsMetadataOnDescribeTopics []*sarama.TopicMetadata
+
 	T *testing.T
 }
 
@@ -43,7 +48,12 @@ func (m *MockKafkaClusterAdmin) ListTopics() (map[string]sarama.TopicDetail, err
 }
 
 func (m *MockKafkaClusterAdmin) DescribeTopics(topics []string) (metadata []*sarama.TopicMetadata, err error) {
-	panic("implement me")
+
+	if diff := cmp.Diff(topics, m.ExpectedTopics); diff != "" {
+		m.T.Errorf("unexpected topics (-want, +got) %s", diff)
+	}
+
+	return m.ExpectedTopicsMetadataOnDescribeTopics, m.ExpectedErrorOnDescribeTopics
 }
 
 func (m *MockKafkaClusterAdmin) DeleteTopic(topic string) error {
diff --git a/control-plane/pkg/reconciler/kafka/topic_test.go b/control-plane/pkg/reconciler/kafka/topic_test.go
index 6f72f373b..e68bf09f0 100644
--- a/control-plane/pkg/reconciler/kafka/topic_test.go
+++ b/control-plane/pkg/reconciler/kafka/topic_test.go
@@ -327,3 +327,113 @@ func TestNewClusterAdminFuncDeleteTopicCloseClusterAdmin(t *testing.T) {
 
 	assert.True(t, ca.ExpectedClose, "expected call to Close() on ClusterAdmin")
 }
+
+func TestNewClusterAdminFuncIsTopicPresent(t *testing.T) {
+	type args struct {
+		topic            string
+		bootstrapServers []string
+	}
+	tests := []struct {
+		name    string
+		f       NewClusterAdminFunc
+		args    args
+		want    bool
+		wantErr bool
+	}{
+		{
+			name: "topic exists",
+			f: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopics: []string{"topic-name-1"},
+					ExpectedTopicsMetadataOnDescribeTopics: []*sarama.TopicMetadata{
+						{
+							Name:       "topic-name-1",
+							IsInternal: false,
+						},
+					},
+					T: t,
+				}, nil
+			},
+			args: args{
+				topic:            "topic-name-1",
+				bootstrapServers: []string{},
+			},
+			want:    true,
+			wantErr: false,
+		},
+		{
+			name: "topic exists - internal topic",
+			f: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopics: []string{"topic-name-1"},
+					ExpectedTopicsMetadataOnDescribeTopics: []*sarama.TopicMetadata{
+						{
+							Name:       "topic-name-1",
+							IsInternal: true,
+						},
+					},
+					T: t,
+				}, nil
+			},
+			args: args{
+				topic:            "topic-name-1",
+				bootstrapServers: []string{},
+			},
+			want:    false,
+			wantErr: false,
+		},
+		{
+			name: "DescribeTopics returns error",
+			f: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+				return &kafkatesting.MockKafkaClusterAdmin{
+					ExpectedTopics:                []string{"topic-name-1"},
+					ExpectedErrorOnDescribeTopics: fmt.Errorf("error"),
+					T:                             t,
+				}, nil
+			},
+			args: args{
+				topic:            "topic-name-1",
+				bootstrapServers: []string{},
+			},
+			want:    false,
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := tt.f.IsTopicPresentAndValid(tt.args.topic, tt.args.bootstrapServers)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("IsTopicPresentAndValid() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("IsTopicPresentAndValid() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestNewClusterAdminFuncIsTopicPresentCloseClusterAdmin(t *testing.T) {
+
+	ca := &kafkatesting.MockKafkaClusterAdmin{
+		ExpectedTopics: []string{"topic-name-1"},
+		ExpectedTopicsMetadataOnDescribeTopics: []*sarama.TopicMetadata{
+			{
+				Name:       "topic-name-1",
+				IsInternal: false,
+			},
+		},
+		ExpectedClose: false,
+		T:             t,
+	}
+
+	f := NewClusterAdminFunc(func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
+		return ca, nil
+	})
+
+	got, err := f.IsTopicPresentAndValid("topic-name-1", []string{})
+
+	assert.Nil(t, err, "IsTopicPresentAndValid() error = %v, wantErr %v", err, false)
+	assert.True(t, got, "IsTopicPresentAndValid() got = %v, want %v", got, true)
+	assert.True(t, ca.ExpectedClose, "expected call to Close() on ClusterAdmin")
+}
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index e4a66cbd3..2a86afc58 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -14,14 +14,16 @@
  * limitations under the License.
  */
 
-package sink
+package sink_test
 
 import (
 	"context"
 	"fmt"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"io"
 	"testing"
 
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"github.com/Shopify/sarama"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -41,6 +43,7 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/broker"
 	kafkatesting "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka/testing"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 	. "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/testing"
 )
 
@@ -49,10 +52,13 @@ const (
 
 	bootstrapServers = "kafka-1:9092,kafka-2:9093"
 
-	wantTopicName          = "wantTopicName"
-	wantErrorOnCreateTopic = "wantErrorOnCreateTopic"
-	wantErrorOnDeleteTopic = "wantErrorOnDeleteTopic"
-	ExpectedTopicDetail    = "expectedTopicDetail"
+	wantTopicName                  = "wantTopicName"
+	wantErrorOnCreateTopic         = "wantErrorOnCreateTopic"
+	wantErrorOnDeleteTopic         = "wantErrorOnDeleteTopic"
+	ExpectedTopicDetail            = "expectedTopicDetail"
+	ExpectedTopicsOnDescribeTopics = "expectedTopicsOnDescribeTopics"
+	ExpectedTopicIsPresent         = "expectedTopicIsPresent"
+	ExpectedErrorOnDescribeTopics  = "expectedErrorOnDescribeTopics"
 
 	TopicPrefix = "knative-sink-"
 )
@@ -72,6 +78,8 @@ var (
 
 func TestSinkReconciler(t *testing.T) {
 
+	v1alpha1.RegisterConditionSet(base.ConditionSet)
+
 	t.Parallel()
 
 	for _, f := range Formats {
@@ -89,7 +97,9 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 		{
 			Name: "Reconciled normal",
 			Objects: []runtime.Object{
-				NewSink(),
+				NewSink(
+					SinkControllerOwnsTopic,
+				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
 					base.VolumeGenerationAnnotationKey: "1",
@@ -123,20 +133,137 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&configs.Env),
 					),
 				},
 			},
 		},
+		{
+			Name: "Reconciled normal - no topic owner",
+			Objects: []runtime.Object{
+				NewSink(
+					SinkControllerDontOwnTopic,
+					func(sink *v1alpha1.KafkaSink) {
+						sink.Spec.ReplicationFactor = nil
+						sink.Spec.NumPartitions = nil
+					},
+				),
+				NewConfigMap(&configs, nil),
+				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:               SinkUUID,
+							Topics:           []string{SinkTopic()},
+							BootstrapServers: bootstrapServers,
+							Ingress: &contract.Ingress{
+								ContentMode: contract.ContentMode_STRUCTURED,
+								IngressType: &contract.Ingress_Path{
+									Path: receiver.Path(SinkNamespace, SinkName),
+								},
+							},
+						},
+					},
+					Generation: 1,
+				}),
+				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						SinkControllerDontOwnTopic,
+						func(sink *v1alpha1.KafkaSink) {
+							sink.Spec.ReplicationFactor = nil
+							sink.Spec.NumPartitions = nil
+						},
+						InitSinkConditions,
+						SinkDataPlaneAvailable,
+						SinkConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ExternalTopicOwner),
+						SinkAddressable(&configs.Env),
+					),
+				},
+			},
+		},
+		{
+			Name: "No topic owner - topic present err",
+			Objects: []runtime.Object{
+				NewSink(
+					SinkControllerDontOwnTopic,
+					func(sink *v1alpha1.KafkaSink) {
+						sink.Spec.ReplicationFactor = nil
+						sink.Spec.NumPartitions = nil
+					},
+				),
+				NewConfigMap(&configs, nil),
+				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+					"annotation_to_preserve":           "value_to_preserve",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"topic is not present: "+SinkNotPresentErrFormat,
+					SinkTopic(), io.EOF,
+				),
+			},
+			WantErr: true,
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						SinkControllerDontOwnTopic,
+						func(sink *v1alpha1.KafkaSink) {
+							sink.Spec.ReplicationFactor = nil
+							sink.Spec.NumPartitions = nil
+						},
+						InitSinkConditions,
+						SinkDataPlaneAvailable,
+						SinkTopicNotPresentErr(SinkTopic(), io.EOF),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				ExpectedErrorOnDescribeTopics: io.EOF,
+			},
+		},
 		{
 			Name: "Reconciled normal - set topic and bootstrap servers",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					func(sink *v1alpha1.KafkaSink) {
 						sink.Spec.Topic = "my-topic-1"
 						sink.Spec.BootstrapServers = []string{"kafka-broker:10000"}
@@ -175,15 +302,17 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						func(sink *v1alpha1.KafkaSink) {
 							sink.Spec.Topic = "my-topic-1"
 							sink.Spec.BootstrapServers = []string{"kafka-broker:10000"}
 						},
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers([]string{"kafka-broker:10000"}),
+						SinkTopicReadyWithOwner("my-topic-1", sink.ControllerTopicOwner),
 						SinkConfigMapUpdatedReady(&configs.Env),
-						SinkTopicReadyWithName("my-topic-1"),
 						SinkAddressable(&configs.Env),
 					),
 				},
@@ -196,6 +325,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			Name: "Failed to create topic",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
 				SinkReceiverPod(configs.SystemNamespace, nil),
@@ -217,6 +347,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
 						BootstrapServers(bootstrapServersArr),
@@ -232,6 +363,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			Name: "Config map not found - create config map",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
 				NewService(),
@@ -273,11 +405,14 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&configs.Env),
 					),
 				},
@@ -287,6 +422,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			Name: "Reconciled normal - config map not readable",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
 				NewConfigMap(&configs, []byte(`{"hello": "world"}`)),
@@ -318,11 +454,14 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&configs.Env),
 					),
 				},
@@ -332,6 +471,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			Name: "Reconciled normal - preserve config map previous state",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
 				NewConfigMapFromContract(&contract.Contract{
@@ -386,11 +526,14 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
 						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&configs.Env),
 					),
 				},
@@ -400,6 +543,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			Name: "Reconciled normal - update existing broker while preserving others",
 			Objects: []runtime.Object{
 				NewSink(
+					SinkControllerOwnsTopic,
 					BootstrapServers(bootstrapServersArr),
 				),
 				NewConfigMapFromContract(&contract.Contract{
@@ -452,11 +596,13 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
 				{
 					Object: NewSink(
+						SinkControllerOwnsTopic,
 						InitSinkConditions,
 						SinkDataPlaneAvailable,
+						SinkConfigParsed,
 						BootstrapServers(bootstrapServersArr),
 						SinkConfigMapUpdatedReady(&configs.Env),
-						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
 						SinkAddressable(&configs.Env),
 					),
 				},
@@ -498,6 +644,212 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 	useTable(t, table, &configs)
 }
 
+func TestSinkFinalizer(t *testing.T) {
+
+	v1alpha1.RegisterConditionSet(base.ConditionSet)
+
+	t.Parallel()
+
+	for _, f := range Formats {
+		sinkFinalization(t, f, *DefaultConfigs)
+	}
+}
+
+func sinkFinalization(t *testing.T, format string, configs broker.Configs) {
+
+	testKey := fmt.Sprintf("%s/%s", SinkNamespace, SinkName)
+
+	configs.DataPlaneConfigFormat = format
+
+	table := TableTest{
+		{
+			Name: "Reconciled normal - topic externally controlled",
+			Objects: []runtime.Object{
+				NewDeletedSink(
+					SinkControllerDontOwnTopic,
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+						{
+							Id:     SinkUUID,
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &configs),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+			},
+		},
+		{
+			Name: "Reconciled normal - topic controlled by us",
+			Objects: []runtime.Object{
+				NewDeletedSink(
+					SinkControllerOwnsTopic,
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+						{
+							Id:     SinkUUID,
+							Topics: []string{"topic-2"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &configs),
+			},
+			Key: testKey,
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+			},
+		},
+		{
+			Name: "Reconciled normal - topic controlled by us - error deleting topic",
+			Objects: []runtime.Object{
+				NewDeletedSink(
+					SinkControllerOwnsTopic,
+					func(sink *v1alpha1.KafkaSink) {
+						sink.Spec.Topic = "topic-2"
+					},
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+						{
+							Id:     SinkUUID,
+							Topics: []string{"topic-2"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &configs),
+			},
+			Key:     testKey,
+			WantErr: true,
+			WantEvents: []string{
+				Eventf(
+					corev1.EventTypeWarning,
+					"InternalError",
+					"failed to delete topic %s: %v",
+					"topic-2", deleteTopicError,
+				),
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(&configs, &contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Id:     SinkUUID + "a",
+							Topics: []string{"topic"},
+							Ingress: &contract.Ingress{
+								IngressType: &contract.Ingress_Path{
+									Path: "path",
+								},
+								ContentMode: contract.ContentMode_STRUCTURED,
+							},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}),
+			},
+			OtherTestData: map[string]interface{}{
+				wantErrorOnDeleteTopic: deleteTopicError,
+				wantTopicName:          "topic-2",
+			},
+		},
+	}
+
+	for i := range table {
+		table[i].Name = table[i].Name + " - " + format
+	}
+
+	useTable(t, table, &configs)
+}
+
 func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 
 	table.Test(t, NewFactory(configs, func(ctx context.Context, listers *Listers, configs *broker.Configs, row *TableRow) controller.Reconciler {
@@ -527,7 +879,31 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 			expectedTopicDetail = td.(sarama.TopicDetail)
 		}
 
-		reconciler := &Reconciler{
+		expectedTopicsOnDescribeTopics := []string{SinkTopic()}
+		if et, ok := row.OtherTestData[ExpectedTopicsOnDescribeTopics]; ok {
+			expectedTopicsOnDescribeTopics = et.([]string)
+		}
+		expectedTopicIsPresent := true
+		if isPresent, ok := row.OtherTestData[ExpectedTopicIsPresent]; ok {
+			expectedTopicIsPresent = isPresent.(bool)
+		}
+
+		var metadata []*sarama.TopicMetadata
+		if expectedTopicIsPresent {
+			for _, topic := range expectedTopicsOnDescribeTopics {
+				metadata = append(metadata, &sarama.TopicMetadata{
+					Name:       topic,
+					IsInternal: false,
+				})
+			}
+		}
+
+		var errorOnDescribeTopics error
+		if isPresentError, ok := row.OtherTestData[ExpectedErrorOnDescribeTopics]; ok {
+			errorOnDescribeTopics = isPresentError.(error)
+		}
+
+		reconciler := &sink.Reconciler{
 			Reconciler: &base.Reconciler{
 				KubeClient:                  kubeclient.Get(ctx),
 				PodLister:                   listers.GetPodLister(),
@@ -540,11 +916,14 @@ func useTable(t *testing.T, table TableTest, configs *broker.Configs) {
 			ConfigMapLister: listers.GetConfigMapLister(),
 			ClusterAdmin: func(addrs []string, config *sarama.Config) (sarama.ClusterAdmin, error) {
 				return &kafkatesting.MockKafkaClusterAdmin{
-					ExpectedTopicName:   expectedTopicName,
-					ExpectedTopicDetail: expectedTopicDetail,
-					ErrorOnCreateTopic:  onCreateTopicError,
-					ErrorOnDeleteTopic:  onDeleteTopicError,
-					T:                   t,
+					ExpectedTopicName:                      expectedTopicName,
+					ExpectedTopicDetail:                    expectedTopicDetail,
+					ErrorOnCreateTopic:                     onCreateTopicError,
+					ErrorOnDeleteTopic:                     onDeleteTopicError,
+					ExpectedTopics:                         expectedTopicsOnDescribeTopics,
+					ExpectedTopicsMetadataOnDescribeTopics: metadata,
+					ExpectedErrorOnDescribeTopics:          errorOnDescribeTopics,
+					T:                                      t,
 				}, nil
 			},
 			Configs: &configs.Env,
diff --git a/control-plane/pkg/reconciler/testing/objects_sink.go b/control-plane/pkg/reconciler/testing/objects_sink.go
index 137d42a9a..4eb8ed251 100644
--- a/control-plane/pkg/reconciler/testing/objects_sink.go
+++ b/control-plane/pkg/reconciler/testing/objects_sink.go
@@ -19,6 +19,7 @@ package testing
 import (
 	"context"
 	"fmt"
+	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -33,6 +34,7 @@ import (
 	"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/kafka"
+	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/sink"
 )
 
 const (
@@ -43,6 +45,8 @@ const (
 	SinkNumPartitions     = 10
 	SinkReplicationFactor = 3
 
+	SinkNotPresentErrFormat = "failed to describe topic %s: %v"
+
 	topicPrefix = "knative-sink-"
 )
 
@@ -76,6 +80,17 @@ func NewSink(options ...SinkOption) runtime.Object {
 	return sink
 }
 
+func NewDeletedSink(options ...SinkOption) runtime.Object {
+	return NewSink(
+		append(
+			options,
+			func(sink *eventing.KafkaSink) {
+				sink.DeletionTimestamp = &metav1.Time{Time: time.Now()}
+			},
+		)...,
+	)
+}
+
 func SinkTopic() string {
 	return kafka.Topic(topicPrefix, &metav1.ObjectMeta{
 		Name:      SinkName,
@@ -113,10 +128,33 @@ func SinkTopicReadyWithName(topic string) func(sink *eventing.KafkaSink) {
 	}
 }
 
+func SinkTopicReadyWithOwner(topic, owner string) func(sink *eventing.KafkaSink) {
+	return func(sink *eventing.KafkaSink) {
+		sink.GetConditionSet().Manage(sink.GetStatus()).MarkTrueWithReason(
+			base.ConditionTopicReady,
+			fmt.Sprintf("Topic %s (owner %s)", topic, owner),
+			"",
+		)
+	}
+}
 func SinkTopicReady(sink *eventing.KafkaSink) {
 	SinkTopicReadyWithName(SinkTopic())(sink)
 }
 
+func SinkConfigParsed(sink *eventing.KafkaSink) {
+	sink.GetConditionSet().Manage(sink.GetStatus()).MarkTrue(base.ConditionConfigParsed)
+}
+
+func SinkTopicNotPresentErr(topic string, err error) func(sink *eventing.KafkaSink) {
+	return func(sink *eventing.KafkaSink) {
+		sink.GetConditionSet().Manage(sink.GetStatus()).MarkFalse(
+			base.ConditionTopicReady,
+			base.ReasonTopicNotPresent,
+			fmt.Sprintf(SinkNotPresentErrFormat, topic, err),
+		)
+	}
+}
+
 func SinkDataPlaneAvailable(sink *eventing.KafkaSink) {
 	sink.GetConditionSet().Manage(sink.GetStatus()).MarkTrue(base.ConditionDataPlaneAvailable)
 }
@@ -129,6 +167,22 @@ func SinkDataPlaneNotAvailable(sink *eventing.KafkaSink) {
 	)
 }
 
+func SinkControllerOwnsTopic(sink *eventing.KafkaSink) {
+	allocateStatusAnnotations(sink)
+	sink.GetStatus().Annotations[base.TopicOwnerAnnotation] = sinkreconciler.ControllerTopicOwner
+}
+
+func SinkControllerDontOwnTopic(sink *eventing.KafkaSink) {
+	allocateStatusAnnotations(sink)
+	sink.GetStatus().Annotations[base.TopicOwnerAnnotation] = sinkreconciler.ExternalTopicOwner
+}
+
+func allocateStatusAnnotations(sink *eventing.KafkaSink) {
+	if sink.GetStatus().Annotations == nil {
+		sink.GetStatus().Annotations = make(map[string]string, 1)
+	}
+}
+
 func SinkAddressable(configs *config.Env) func(sink *eventing.KafkaSink) {
 
 	return func(sink *eventing.KafkaSink) {
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index c827b3a01..ef1413652 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -19,9 +19,10 @@ package trigger
 import (
 	"context"
 	"fmt"
+	"testing"
+
 	"google.golang.org/protobuf/testing/protocmp"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
-	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
diff --git a/data-plane/tests/pom.xml b/data-plane/tests/pom.xml
index e788e68d8..58856af7b 100644
--- a/data-plane/tests/pom.xml
+++ b/data-plane/tests/pom.xml
@@ -28,7 +28,7 @@
   <artifactId>tests</artifactId>
 
   <properties>
-    <debezium.version>1.2.3.Final</debezium.version>
+    <debezium.version>1.2.4.Final</debezium.version>
     <kafka.version>2.6.0</kafka.version>
   </properties>
 
diff --git a/test/pkg/observability/watch_config_map.go b/test/pkg/observability/watch_config_map.go
index b651806ba..360e8290c 100644
--- a/test/pkg/observability/watch_config_map.go
+++ b/test/pkg/observability/watch_config_map.go
@@ -18,11 +18,12 @@ package observability
 
 import (
 	"context"
-	"google.golang.org/protobuf/encoding/protojson"
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
 	"log"
 	"sync"
 
+	"google.golang.org/protobuf/encoding/protojson"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+
 	"github.com/google/go-cmp/cmp"
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
index 6c309fe6c..22cc84470 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_control_plane_test_helper.go
@@ -136,7 +136,17 @@ func triggerV1Beta1ReadyAfterBrokerIncludesSubURI(t *testing.T, triggerName, bro
 
 	client.WaitForResourceReadyOrFail(triggerName, testlib.TriggerTypeMeta)
 
-	tr, err := client.Eventing.EventingV1beta1().Triggers(client.Namespace).Get(context.Background(), triggerName, metav1.GetOptions{})
+	var tr *eventingv1beta1.Trigger
+	triggers := client.Eventing.EventingV1beta1().Triggers(client.Namespace)
+	err = client.RetryWebhookErrors(func(attempts int) (err error) {
+		var e error
+		client.T.Logf("Getting v1beta1 trigger %s", triggerName)
+		tr, e = triggers.Get(context.Background(), triggerName, metav1.GetOptions{})
+		if e != nil {
+			client.T.Logf("Failed to get trigger %q: %v", triggerName, e)
+		}
+		return err
+	})
 	if err != nil {
 		t.Fatalf("Error: Could not get trigger %s: %v", triggerName, err)
 	}
@@ -147,7 +157,17 @@ func triggerV1Beta1ReadyAfterBrokerIncludesSubURI(t *testing.T, triggerName, bro
 
 func triggerV1Beta1ReadyIncludesSubURI(t *testing.T, triggerName string, client *testlib.Client) {
 	client.WaitForResourceReadyOrFail(triggerName, testlib.TriggerTypeMeta)
-	tr, err := client.Eventing.EventingV1beta1().Triggers(client.Namespace).Get(context.Background(), triggerName, metav1.GetOptions{})
+	var tr *eventingv1beta1.Trigger
+	triggers := client.Eventing.EventingV1beta1().Triggers(client.Namespace)
+	err := client.RetryWebhookErrors(func(attempts int) (err error) {
+		var e error
+		client.T.Logf("Getting v1beta1 trigger %s", triggerName)
+		tr, e = triggers.Get(context.Background(), triggerName, metav1.GetOptions{})
+		if e != nil {
+			client.T.Logf("Failed to get trigger %q: %v", triggerName, e)
+		}
+		return err
+	})
 	if err != nil {
 		t.Fatalf("Error: Could not get trigger %s: %v", triggerName, err)
 	}
diff --git a/vendor/knative.dev/eventing/test/conformance/helpers/broker_data_plane_test_helper.go b/vendor/knative.dev/eventing/test/conformance/helpers/broker_data_plane_test_helper.go
index 6dcc47142..9facba445 100644
--- a/vendor/knative.dev/eventing/test/conformance/helpers/broker_data_plane_test_helper.go
+++ b/vendor/knative.dev/eventing/test/conformance/helpers/broker_data_plane_test_helper.go
@@ -40,7 +40,6 @@ var podMeta = metav1.TypeMeta{
 
 func BrokerDataPlaneSetupHelper(ctx context.Context, client *testlib.Client, brokerClass string, brokerTestRunner testlib.ComponentsTestRunner) *eventingv1beta1.Broker {
 	var broker *eventingv1beta1.Broker
-	var err error
 	brokerName := brokerTestRunner.ComponentName
 	brokerNamespace := brokerTestRunner.ComponentNamespace
 	if brokerName == "" || brokerNamespace == "" {
@@ -54,7 +53,17 @@ func BrokerDataPlaneSetupHelper(ctx context.Context, client *testlib.Client, bro
 		)
 		client.WaitForResourceReadyOrFail(broker.Name, testlib.BrokerTypeMeta)
 	} else {
-		if broker, err = client.Eventing.EventingV1beta1().Brokers(brokerNamespace).Get(ctx, brokerName, metav1.GetOptions{}); err != nil {
+		brokers := client.Eventing.EventingV1beta1().Brokers(brokerNamespace)
+		err := client.RetryWebhookErrors(func(attempts int) (err error) {
+			var e error
+			client.T.Logf("Getting v1beta1 broker %s", brokerName)
+			broker, e = brokers.Get(context.Background(), brokerName, metav1.GetOptions{})
+			if e != nil {
+				client.T.Logf("Failed to get broker %q: %v", brokerName, e)
+			}
+			return err
+		})
+		if err != nil {
 			client.T.Fatalf("Could not Get broker %s/%s: %v", brokerNamespace, brokerName, err)
 		}
 	}
diff --git a/vendor/knative.dev/pkg/test/ingress/ingress.go b/vendor/knative.dev/pkg/test/ingress/ingress.go
index 7cf8f9240..6b5094151 100644
--- a/vendor/knative.dev/pkg/test/ingress/ingress.go
+++ b/vendor/knative.dev/pkg/test/ingress/ingress.go
@@ -33,7 +33,12 @@ const (
 )
 
 // GetIngressEndpoint gets the ingress public IP or hostname.
-func GetIngressEndpoint(ctx context.Context, kubeClientset *kubernetes.Clientset) (string, error) {
+// address - is the endpoint to which we should actually connect.
+// portMap - translates the request's port to the port on address to which the caller
+//    should connect.  This is used when the resolution to address goes through some
+//    sort of port-mapping, e.g. Kubernetes node ports.
+// err - an error when address/portMap cannot be established.
+func GetIngressEndpoint(ctx context.Context, kubeClientset *kubernetes.Clientset, endpointOverride string) (address string, portMap func(string) string, err error) {
 	ingressName := istioIngressName
 	if gatewayOverride := os.Getenv("GATEWAY_OVERRIDE"); gatewayOverride != "" {
 		ingressName = gatewayOverride
@@ -45,13 +50,25 @@ func GetIngressEndpoint(ctx context.Context, kubeClientset *kubernetes.Clientset
 
 	ingress, err := kubeClientset.CoreV1().Services(ingressNamespace).Get(ctx, ingressName, metav1.GetOptions{})
 	if err != nil {
-		return "", err
+		return "", nil, err
+	}
+
+	// If an override is provided, use it
+	if endpointOverride != "" {
+		return endpointOverride, func(port string) string {
+			for _, sp := range ingress.Spec.Ports {
+				if fmt.Sprint(sp.Port) == port {
+					return fmt.Sprint(sp.NodePort)
+				}
+			}
+			return port
+		}, nil
 	}
 	endpoint, err := EndpointFromService(ingress)
 	if err != nil {
-		return "", err
+		return "", nil, err
 	}
-	return endpoint, nil
+	return endpoint, func(in string) string { return in }, nil
 }
 
 // EndpointFromService extracts the endpoint from the service's ingress.
diff --git a/vendor/knative.dev/pkg/test/spoof/spoof.go b/vendor/knative.dev/pkg/test/spoof/spoof.go
index fa285782d..3278ca346 100644
--- a/vendor/knative.dev/pkg/test/spoof/spoof.go
+++ b/vendor/knative.dev/pkg/test/spoof/spoof.go
@@ -26,7 +26,6 @@ import (
 	"io/ioutil"
 	"net"
 	"net/http"
-	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -60,17 +59,8 @@ func (r *Response) String() string {
 	return fmt.Sprintf("status: %d, body: %s, headers: %v", r.StatusCode, string(r.Body), r.Header)
 }
 
-// Interface defines the actions that can be performed by the spoofing client.
-type Interface interface {
-	Do(*http.Request, ...ErrorRetryChecker) (*Response, error)
-	Poll(*http.Request, ResponseChecker, ...ErrorRetryChecker) (*Response, error)
-}
-
 // https://medium.com/stupid-gopher-tricks/ensuring-go-interface-satisfaction-at-compile-time-1ed158e8fa17
-var (
-	_           Interface = (*SpoofingClient)(nil)
-	dialContext           = (&net.Dialer{}).DialContext
-)
+var dialContext = (&net.Dialer{}).DialContext
 
 // ResponseChecker is used to determine when SpoofinClient.Poll is done polling.
 // This allows you to predicate wait.PollImmediate on the request's http.Response.
@@ -107,10 +97,9 @@ func New(
 	domain string,
 	resolvable bool,
 	endpointOverride string,
-	requestInterval time.Duration,
-	requestTimeout time.Duration,
+	requestInterval, requestTimeout time.Duration,
 	opts ...TransportOption) (*SpoofingClient, error) {
-	endpoint, err := ResolveEndpoint(ctx, kubeClientset, domain, resolvable, endpointOverride)
+	endpoint, mapper, err := ResolveEndpoint(ctx, kubeClientset, domain, resolvable, endpointOverride)
 	if err != nil {
 		return nil, fmt.Errorf("failed get the cluster endpoint: %w", err)
 	}
@@ -119,13 +108,13 @@ func New(
 	logf("Spoofing %s -> %s", domain, endpoint)
 	transport := &http.Transport{
 		DialContext: func(ctx context.Context, network, addr string) (conn net.Conn, e error) {
-			spoofed := addr
-			if i := strings.LastIndex(addr, ":"); i != -1 && domain == addr[:i] {
-				// The original hostname:port is spoofed by replacing the hostname by the value
-				// returned by ResolveEndpoint.
-				spoofed = endpoint + ":" + addr[i+1:]
+			_, port, err := net.SplitHostPort(addr)
+			if err != nil {
+				return nil, err
 			}
-			return dialContext(ctx, network, spoofed)
+			// The original hostname:port is spoofed by replacing the hostname by the value
+			// returned by ResolveEndpoint.
+			return dialContext(ctx, network, net.JoinHostPort(endpoint, mapper(port)))
 		},
 	}
 
@@ -139,28 +128,25 @@ func New(
 		Propagation: tracecontextb3.TraceContextB3Egress,
 	}
 
-	sc := SpoofingClient{
+	sc := &SpoofingClient{
 		Client:          &http.Client{Transport: roundTripper},
 		RequestInterval: requestInterval,
 		RequestTimeout:  requestTimeout,
 		Logf:            logf,
 	}
-	return &sc, nil
+	return sc, nil
 }
 
 // ResolveEndpoint resolves the endpoint address considering whether the domain is resolvable and taking into
 // account whether the user overrode the endpoint address externally
-func ResolveEndpoint(ctx context.Context, kubeClientset *kubernetes.Clientset, domain string, resolvable bool, endpointOverride string) (string, error) {
+func ResolveEndpoint(ctx context.Context, kubeClientset *kubernetes.Clientset, domain string, resolvable bool, endpointOverride string) (string, func(string) string, error) {
+	id := func(in string) string { return in }
 	// If the domain is resolvable, it can be used directly
 	if resolvable {
-		return domain, nil
-	}
-	// If an override is provided, use it
-	if endpointOverride != "" {
-		return endpointOverride, nil
+		return domain, id, nil
 	}
 	// Otherwise, use the actual cluster endpoint
-	return ingress.GetIngressEndpoint(ctx, kubeClientset)
+	return ingress.GetIngressEndpoint(ctx, kubeClientset, endpointOverride)
 }
 
 // Do dispatches to the underlying http.Client.Do, spoofing domains as needed
@@ -170,7 +156,9 @@ func (sc *SpoofingClient) Do(req *http.Request, errorRetryCheckers ...ErrorRetry
 	return sc.Poll(req, func(*Response) (bool, error) { return true, nil }, errorRetryCheckers...)
 }
 
-// Poll executes an http request until it satisfies the inState condition or encounters an error.
+// Poll executes an http request until it satisfies the inState condition or, if there's an error,
+// none of the error retry checkers permit a retry.
+// If no retry checkers are specified `DefaultErrorRetryChecker` will be used.
 func (sc *SpoofingClient) Poll(req *http.Request, inState ResponseChecker, errorRetryCheckers ...ErrorRetryChecker) (*Response, error) {
 	if len(errorRetryCheckers) == 0 {
 		errorRetryCheckers = []ErrorRetryChecker{DefaultErrorRetryChecker}
diff --git a/vendor/knative.dev/test-infra/scripts/e2e-tests.sh b/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
index 51356f998..19a7b6eec 100644
--- a/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
+++ b/vendor/knative.dev/test-infra/scripts/e2e-tests.sh
@@ -167,10 +167,12 @@ function initialize() {
 
   (( IS_PROW )) && [[ -z "${GCP_PROJECT_ID:-}" ]] && IS_BOSKOS=1
 
-  if (( SKIP_ISTIO_ADDON )); then
-    custom_flags+=("--addons=NodeLocalDNS")
-  else
-    custom_flags+=("--addons=Istio,NodeLocalDNS")
+  if [[ "${CLOUD_PROVIDER}" == "gke" ]]; then
+    if (( SKIP_ISTIO_ADDON )); then
+      custom_flags+=("--addons=NodeLocalDNS")
+    else
+      custom_flags+=("--addons=Istio,NodeLocalDNS")
+    fi
   fi
 
   readonly IS_BOSKOS
diff --git a/vendor/knative.dev/test-infra/scripts/infra-library.sh b/vendor/knative.dev/test-infra/scripts/infra-library.sh
index 1ab61d3e8..a27934997 100644
--- a/vendor/knative.dev/test-infra/scripts/infra-library.sh
+++ b/vendor/knative.dev/test-infra/scripts/infra-library.sh
@@ -124,8 +124,10 @@ function create_test_cluster() {
 # Parameters: $1 - extra cluster creation flags
 #             $2 - test command to run by the kubetest2 tester
 function create_kind_test_cluster() {
-  # TODO(chizhg): implement with kubetest2
-  return 0
+  local -n _custom_flags=$1
+  local -n _test_command=$2
+
+  kubetest2 kind "${_custom_flags[@]}" --up --down --test=exec -- "${_test_command[@]}"
 }
 
 # Create a GKE test cluster with kubetest2 and run the test command.