Repository: knative-sandbox/eventing-kafka-broker
Pull request #2886 (2023-01-12 14:36:07)

Tests for broker broker non existing external topic
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- two more tests for non-existing external topic


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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2886#" title="Author self-approved">matzew</a>*

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

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

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

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

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

> :exclamation: Current head 3dbf8bc differs from pull request most recent head 89ed24a. Consider uploading reports for the commit 89ed24a to get more accurate results

```diff
@@             Coverage Diff              @@
##               main    #2886      +/-   ##
============================================
- Coverage     63.57%   58.10%   -5.48%     
============================================
  Files           155       78      -77     
  Lines         10859     8240    -2619     
  Branches        231        0     -231     
============================================
- Hits           6904     4788    -2116     
+ Misses         3453     3088     -365     
+ Partials        502      364     -138     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `?` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...oker/dispatcher/impl/consumer/KeyDeserializer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LZXlEZXNlcmlhbGl6ZXIuamF2YQ==) | | |
| [...atcher/impl/consumer/KafkaConsumerRecordUtils.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9LYWZrYUNvbnN1bWVyUmVjb3JkVXRpbHMuamF2YQ==) | | |
| [...tive/eventing/kafka/broker/core/utils/Logging.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvTG9nZ2luZy5qYXZh) | | |
| [...cher/impl/filter/subscriptionsapi/CeSqlFilter.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9maWx0ZXIvc3Vic2NyaXB0aW9uc2FwaS9DZVNxbEZpbHRlci5qYXZh) | | |
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | | |
| [...eventing/kafka/broker/receiver/RequestContext.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9SZXF1ZXN0Q29udGV4dC5qYXZh) | | |
| [.../broker/receiver/main/ReceiverVerticleFactory.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9tYWluL1JlY2VpdmVyVmVydGljbGVGYWN0b3J5LmphdmE=) | | |
| [...enting/kafka/broker/core/utils/Configurations.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvdXRpbHMvQ29uZmlndXJhdGlvbnMuamF2YQ==) | | |
| [...patcher/impl/consumer/OrderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9PcmRlcmVkQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | | |
| [...venting/kafka/broker/receiver/IngressProducer.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9JbmdyZXNzUHJvZHVjZXIuamF2YQ==) | | |
| ... and [67 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2886?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @matzew: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | 89ed24a04ae5bf0d85de5ec46794f02cd87a0e7c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2886/build-tests_eventing-kafka-broker_main/1613545651205836800) | true | `/test build-tests`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
build-tests_eventing-kafka-broker_main | 89ed24a04ae5bf0d85de5ec46794f02cd87a0e7c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2886/build-tests_eventing-kafka-broker_main/1613545651205836800) | true | `/test build-tests`
channel-integration-tests-sasl-plain_eventing-kafka-broker_main | 89ed24a04ae5bf0d85de5ec46794f02cd87a0e7c | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2886/channel-integration-tests-sasl-plain_eventing-kafka-broker_main/1613545667521679360) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2886#pullrequestreview-1246037180" title="LGTM">creydr</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2886#" title="Author self-approved">matzew</a>*

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->

Patch:
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index 22f325589..4cd099af5 100644
--- a/control-plane/pkg/reconciler/consumer/consumer.go
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -71,7 +71,7 @@ func (r *Reconciler) ReconcileKind(ctx context.Context, c *kafkainternals.Consum
 		return nil // Resource will get queued once we have all resources to build the contract.
 	}
 
-	bound, err := r.schedule(ctx, logger, c, addResource(resourceCt))
+	bound, err := r.schedule(ctx, logger, c, addResource(resourceCt), IsPodNotRunning)
 	if err != nil {
 		return c.MarkBindFailed(err)
 	}
@@ -89,14 +89,14 @@ func (r *Reconciler) FinalizeKind(ctx context.Context, c *kafkainternals.Consume
 
 	logger := logging.FromContext(ctx).Desugar()
 
-	if _, err := r.schedule(ctx, logger, c, removeResource); err != nil {
+	if _, err := r.schedule(ctx, logger, c, removeResource, FalseAnyStatus); err != nil {
 		return c.MarkBindFailed(err)
 	}
 
 	return nil
 }
 
-func (r Reconciler) reconcileContractResource(ctx context.Context, c *kafkainternals.Consumer) (*contract.Resource, error) {
+func (r *Reconciler) reconcileContractResource(ctx context.Context, c *kafkainternals.Consumer) (*contract.Resource, error) {
 	egress, err := r.reconcileContractEgress(ctx, c)
 	if err != nil {
 		return nil, fmt.Errorf("failed to reconcile egress: %w", err)
@@ -131,7 +131,7 @@ func (r Reconciler) reconcileContractResource(ctx context.Context, c *kafkainter
 	return resource, nil
 }
 
-func (r Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainternals.Consumer) (*contract.Egress, error) {
+func (r *Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainternals.Consumer) (*contract.Egress, error) {
 	destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Subscriber, c)
 	if err != nil {
 		return nil, fmt.Errorf("failed to resolve subscriber: %w", err)
@@ -176,7 +176,7 @@ func (r Reconciler) reconcileContractEgress(ctx context.Context, c *kafkainterna
 	return egress, nil
 }
 
-func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consumer, resource *contract.Resource) error {
+func (r *Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consumer, resource *contract.Resource) error {
 	if c.Spec.Auth == nil {
 		return nil
 	}
@@ -216,7 +216,7 @@ func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consume
 	return nil
 }
 
-func (r Reconciler) SecretProviderFunc() security.SecretProviderFunc {
+func (r *Reconciler) SecretProviderFunc() security.SecretProviderFunc {
 	return security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient)
 }
 
@@ -227,7 +227,7 @@ func reconcileCEOverrides(c *kafkainternals.Consumer) *contract.CloudEventOverri
 	return &contract.CloudEventOverrides{Extensions: c.Spec.CloudEventOverrides.Extensions}
 }
 
-func (r Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (*contract.Reference, error) {
+func (r *Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (*contract.Reference, error) {
 
 	cg, err := r.ConsumerGroupLister.ConsumerGroups(c.GetNamespace()).Get(c.GetConsumerGroup().Name)
 	if apierrors.IsNotFound(err) {
@@ -259,7 +259,7 @@ func reconcileDeliveryOrder(c *kafkainternals.Consumer) contract.DeliveryOrder {
 	return contract.DeliveryOrder_UNORDERED
 }
 
-func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternals.Consumer, egress *contract.Egress) error {
+func (r *Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternals.Consumer, egress *contract.Egress) error {
 	if c.Spec.Reply == nil {
 		return nil
 	}
@@ -307,16 +307,16 @@ func addResource(resource *contract.Resource) contractMutatorFunc {
 func removeResource(_ *zap.Logger, ct *contract.Contract, c *kafkainternals.Consumer) int {
 	idx := coreconfig.FindResource(ct, c.GetUID())
 	if idx == coreconfig.NoResource {
-		return coreconfig.NoResource
+		return coreconfig.ResourceUnchanged
 	}
 	coreconfig.DeleteResource(ct, idx)
-	return idx
+	return coreconfig.ResourceChanged
 }
 
 // schedule mutates the ConfigMap associated with the pod specified by Consumer.Spec.PodBind.
 //
 // The actual mutation is done by calling the provided contractMutatorFunc.
-func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkainternals.Consumer, mutatorFunc contractMutatorFunc) (bool, error) {
+func (r *Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkainternals.Consumer, mutatorFunc contractMutatorFunc, shouldWait PodStatusWaitFunc) (bool, error) {
 	// Get the data plane pod when the Consumer should be scheduled.
 	p, err := r.PodLister.Pods(c.Spec.PodBind.PodNamespace).Get(c.Spec.PodBind.PodName)
 	if apierrors.IsNotFound(err) {
@@ -345,7 +345,7 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 	// get or create the associated ConfigMap, since
 	// it won't become ready until we have created the
 	// ConfigMap
-	if p.Status.Phase != corev1.PodRunning {
+	if shouldWait(p) {
 		return false, nil
 	}
 
@@ -367,7 +367,7 @@ func (r Reconciler) schedule(ctx context.Context, logger *zap.Logger, c *kafkain
 	return true, b.UpdatePodsAnnotation(ctx, logger, "dispatcher" /* component, for logging */, ct.Generation, []*corev1.Pod{p})
 }
 
-func (r Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconciler {
+func (r *Reconciler) commonReconciler(p *corev1.Pod, cmName string) base.Reconciler {
 	return base.Reconciler{
 		KubeClient:                    r.KubeClient,
 		PodLister:                     r.PodLister,
@@ -403,3 +403,13 @@ func (r *Reconciler) trackAuthContext(c *kafkainternals.Consumer, auth *kafkaint
 
 	return security.TrackNetSpecSecrets(r.Tracker, auth.NetSpec, c)
 }
+
+type PodStatusWaitFunc func(p *corev1.Pod) bool
+
+func IsPodNotRunning(p *corev1.Pod) bool {
+	return p.Status.Phase != corev1.PodRunning
+}
+
+func FalseAnyStatus(*corev1.Pod) bool {
+	return false
+}


Test patch:
diff --git a/control-plane/pkg/reconciler/consumer/consumer_test.go b/control-plane/pkg/reconciler/consumer/consumer_test.go
index 5530ce702..6f3e1e9b2 100644
--- a/control-plane/pkg/reconciler/consumer/consumer_test.go
+++ b/control-plane/pkg/reconciler/consumer/consumer_test.go
@@ -545,6 +545,195 @@ func TestReconcileKind(t *testing.T) {
 				},
 			},
 		},
+		{
+			Name: "Finalized normal",
+			Objects: []runtime.Object{
+				NewService(),
+				NewConsumerGroup(ConsumerGroupOwnerRef(SourceAsOwnerReference())),
+				NewDispatcherPod("p1"),
+				NewConsumer(1,
+					ConsumerDeletedTimeStamp(),
+					ConsumerUID(ConsumerUUID),
+					ConsumerSpec(NewConsumerSpec(
+						ConsumerTopics(SourceTopics...),
+						ConsumerConfigs(
+							ConsumerBootstrapServersConfig(SourceBootstrapServers),
+							ConsumerGroupIdConfig(SourceConsumerGroup),
+						),
+						ConsumerSubscriber(NewSourceSinkReference()),
+						ConsumerVReplicas(1),
+						ConsumerPlacement(kafkainternals.PodBind{PodName: "p1", PodNamespace: SystemNamespace}),
+					)),
+					ConsumerOwnerRef(ConsumerGroupAsOwnerRef()),
+				),
+				NewConfigMapFromContract(
+					&contract.Contract{
+						Generation: 1,
+						Resources: []*contract.Resource{
+							{
+								Uid:              ConsumerUUID + "a",
+								Topics:           SourceTopics,
+								BootstrapServers: SourceBootstrapServers,
+								Egresses: []*contract.Egress{{
+									ConsumerGroup: SourceConsumerGroup,
+									Destination:   ServiceURL,
+									ReplyStrategy: nil,
+									Filter:        nil,
+									Uid:           ConsumerUUID,
+									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+									KeyType:       0,
+									VReplicas:     1,
+									Reference: &contract.Reference{
+										Uuid:      SourceUUID,
+										Namespace: ConsumerNamespace,
+										Name:      SourceName,
+									},
+									FeatureFlags: defaultContractFeatureFlags,
+								}},
+								Auth:                nil,
+								CloudEventOverrides: nil,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							},
+							{
+								Uid:              ConsumerUUID,
+								Topics:           SourceTopics,
+								BootstrapServers: SourceBootstrapServers,
+								Egresses: []*contract.Egress{{
+									ConsumerGroup: SourceConsumerGroup,
+									Destination:   ServiceURL,
+									ReplyStrategy: nil,
+									Filter:        nil,
+									Uid:           ConsumerUUID,
+									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+									KeyType:       0,
+									VReplicas:     1,
+									Reference: &contract.Reference{
+										Uuid:      SourceUUID,
+										Namespace: ConsumerNamespace,
+										Name:      SourceName,
+									},
+									FeatureFlags: defaultContractFeatureFlags,
+								}},
+								Auth:                nil,
+								CloudEventOverrides: nil,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							},
+							{
+								Uid:              ConsumerUUID + "b",
+								Topics:           SourceTopics,
+								BootstrapServers: SourceBootstrapServers,
+								Egresses: []*contract.Egress{{
+									ConsumerGroup: SourceConsumerGroup,
+									Destination:   ServiceURL,
+									ReplyStrategy: nil,
+									Filter:        nil,
+									Uid:           ConsumerUUID,
+									DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+									KeyType:       0,
+									VReplicas:     1,
+									Reference: &contract.Reference{
+										Uuid:      SourceUUID,
+										Namespace: ConsumerNamespace,
+										Name:      SourceName,
+									},
+									FeatureFlags: defaultContractFeatureFlags,
+								}},
+								Auth:                nil,
+								CloudEventOverrides: nil,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+							},
+						},
+					},
+					SystemNamespace,
+					"p1",
+					base.Json,
+				),
+			},
+			Key:                     testKey,
+			SkipNamespaceValidation: true, // WantCreates compare the broker namespace with configmap namespace, so skip it
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				ConfigMapUpdate(SystemNamespace, "p1", base.Json, &contract.Contract{
+					Generation: 2,
+					Resources: []*contract.Resource{
+						{
+							Uid:              ConsumerUUID + "a",
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: SourceConsumerGroup,
+								Destination:   ServiceURL,
+								ReplyStrategy: nil,
+								Filter:        nil,
+								Uid:           ConsumerUUID,
+								DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+								KeyType:       0,
+								VReplicas:     1,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+								FeatureFlags: defaultContractFeatureFlags,
+							}},
+							Auth:                nil,
+							CloudEventOverrides: nil,
+							Reference: &contract.Reference{
+								Uuid:      SourceUUID,
+								Namespace: ConsumerNamespace,
+								Name:      SourceName,
+							},
+						},
+						{
+							Uid:              ConsumerUUID + "b",
+							Topics:           SourceTopics,
+							BootstrapServers: SourceBootstrapServers,
+							Egresses: []*contract.Egress{{
+								ConsumerGroup: SourceConsumerGroup,
+								Destination:   ServiceURL,
+								ReplyStrategy: nil,
+								Filter:        nil,
+								Uid:           ConsumerUUID,
+								DeliveryOrder: contract.DeliveryOrder_UNORDERED,
+								KeyType:       0,
+								VReplicas:     1,
+								Reference: &contract.Reference{
+									Uuid:      SourceUUID,
+									Namespace: ConsumerNamespace,
+									Name:      SourceName,
+								},
+								FeatureFlags: defaultContractFeatureFlags,
+							}},
+							Auth:                nil,
+							CloudEventOverrides: nil,
+							Reference: &contract.Reference{
+								Uuid:      SourceUUID,
+								Namespace: ConsumerNamespace,
+								Name:      SourceName,
+							},
+						},
+					},
+				},
+					DispatcherPodAsOwnerReference("p1"),
+				),
+				{Object: NewDispatcherPod("p1",
+					PodAnnotations(map[string]string{
+						base.VolumeGenerationAnnotationKey: "2",
+					}),
+				)},
+			},
+		},
 	}
 
 	table.Test(t, NewFactory(nil, func(ctx context.Context, listers *Listers, env *config.Env, row *TableRow) controller.Reconciler {
diff --git a/control-plane/pkg/reconciler/testing/objects_consumer.go b/control-plane/pkg/reconciler/testing/objects_consumer.go
index 3842aba0f..041b43920 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -345,3 +345,9 @@ func ConsumerGroupAsOwnerRef() metav1.OwnerReference {
 		BlockOwnerDeletion: pointer.Bool(true),
 	}
 }
+
+func ConsumerDeletedTimeStamp() ConsumerOption {
+	return func(c *kafkainternals.Consumer) {
+		WithDeletedTimeStamp(c)
+	}
+}
diff --git a/test/e2e_new/broker_test.go b/test/e2e_new/broker_test.go
index 4f9a0ac66..334d5df25 100644
--- a/test/e2e_new/broker_test.go
+++ b/test/e2e_new/broker_test.go
@@ -100,6 +100,38 @@ func TestBrokerAuthSecretDoesNotExist(t *testing.T) {
 	env.Test(ctx, t, features.BrokerAuthSecretDoesNotExist())
 }
 
+func TestBrokerExternalTopicDoesNotExist(t *testing.T) {
+	// this test is observed to flake more when it is parallel
+	// t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.WithPollTimings(PollInterval, PollTimeout),
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.BrokerExternalTopicDoesNotExist())
+}
+
+func TestBrokerExternalTopicAuthSecretDoesNotExist(t *testing.T) {
+	// this test is observed to flake more when it is parallel
+	// t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.WithPollTimings(PollInterval, PollTimeout),
+		environment.Managed(t),
+	)
+
+	env.Test(ctx, t, features.BrokerExternalTopicAuthSecretDoesNotExist())
+}
+
 func TestTriggerLatestOffset(t *testing.T) {
 	// this test is observed to flake more when it is parallel
 	// t.Parallel()
diff --git a/test/e2e_new/kafka_source_test.go b/test/e2e_new/kafka_source_test.go
index fe040a6f7..01b81d533 100644
--- a/test/e2e_new/kafka_source_test.go
+++ b/test/e2e_new/kafka_source_test.go
@@ -44,3 +44,21 @@ func TestKafkaSourceCreateSecretsAfterKafkaSource(t *testing.T) {
 
 	env.Test(ctx, t, features.CreateSecretsAfterKafkaSource())
 }
+
+func TestKafkaSourceDeletedFromContractConfigMaps(t *testing.T) {
+
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+		environment.Managed(t),
+	)
+	t.Cleanup(env.Finish)
+
+	env.Test(ctx, t, features.SetupKafkaSources("permanent-kafka-source-", 21))
+	env.Test(ctx, t, features.SetupAndCleanupKafkaSources("x-kafka-source-", 42))
+	env.Test(ctx, t, features.KafkaSourcesAreNotPresentInContractConfigMaps("x-kafka-source-"))
+}
diff --git a/test/rekt/features/broker_deleted_recreated.go b/test/rekt/features/broker_deleted_recreated.go
index 774150b29..01e1ab486 100644
--- a/test/rekt/features/broker_deleted_recreated.go
+++ b/test/rekt/features/broker_deleted_recreated.go
@@ -131,6 +131,63 @@ func BrokerAuthSecretDoesNotExist() *feature.Feature {
 	return f
 }
 
+// BrokerExternalTopicDoesNotExist tests that a broker can be deleted without the Topic existing.
+func BrokerExternalTopicDoesNotExist() *feature.Feature {
+	f := feature.NewFeatureNamed("delete broker with non existing Topic")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	configName := feature.MakeRandomK8sName("config")
+	topicName := feature.MakeRandomK8sName("topic-does-not-exist") // A k8s name is also a valid topic name.
+
+	f.Setup("create broker config", brokerconfigmap.Install(
+		configName,
+		brokerconfigmap.WithBootstrapServer(testpkg.BootstrapServersPlaintext),
+	))
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		append(
+			broker.WithEnvConfig(),
+			broker.WithConfig(configName),
+			broker.WithAnnotations(
+				map[string]interface{}{
+					kafkabroker.ExternalTopicAnnotation: topicName,
+				}))...))
+
+	f.Assert("delete broker", featuressteps.DeleteBroker(brokerName))
+
+	return f
+}
+
+// BrokerExternalTopicAuthSecretDoesNotExist tests that a broker can be deleted without the Secret and Topic existing.
+func BrokerExternalTopicAuthSecretDoesNotExist() *feature.Feature {
+	f := feature.NewFeatureNamed("delete broker with non existing Secret or Topic")
+
+	brokerName := feature.MakeRandomK8sName("broker")
+	configName := feature.MakeRandomK8sName("config")
+	topicName := feature.MakeRandomK8sName("topic-does-not-exist") // A k8s name is also a valid topic name.
+
+	f.Setup("create broker config", brokerconfigmap.Install(
+		configName,
+		brokerconfigmap.WithBootstrapServer(testpkg.BootstrapServersPlaintext),
+		brokerconfigmap.WithAuthSecret("does-not-exist"),
+	))
+
+	f.Setup("install broker", broker.Install(
+		brokerName,
+		append(
+			broker.WithEnvConfig(),
+			broker.WithConfig(configName),
+			broker.WithAnnotations(
+				map[string]interface{}{
+					kafkabroker.ExternalTopicAnnotation: topicName,
+				}))...))
+
+	f.Assert("delete broker", featuressteps.DeleteBroker(brokerName))
+
+	return f
+}
+
 // BrokerCannotReachKafkaCluster tests that a broker can be deleted even when KafkaCluster is unreachable.
 func BrokerCannotReachKafkaCluster() *feature.Feature {
 	f := feature.NewFeatureNamed("delete broker with unreachable Kafka cluster")
diff --git a/test/rekt/features/kafka_source.go b/test/rekt/features/kafka_source.go
new file mode 100644
index 000000000..3712fa374
--- /dev/null
+++ b/test/rekt/features/kafka_source.go
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2023 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package features
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kafkaclient "knative.dev/eventing-kafka/pkg/client/injection/client"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/environment"
+	"knative.dev/reconciler-test/pkg/feature"
+	"knative.dev/reconciler-test/pkg/knative"
+	"knative.dev/reconciler-test/resources/svc"
+
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkasource"
+	"knative.dev/eventing-kafka-broker/test/rekt/resources/kafkatopic"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	testingpkg "knative.dev/eventing-kafka-broker/test/pkg"
+)
+
+func SetupAndCleanupKafkaSources(prefix string, n int) *feature.Feature {
+	f := SetupKafkaSources(prefix, n)
+	f.Teardown("cleanup resources", f.DeleteResources)
+	return f
+}
+
+func SetupKafkaSources(prefix string, n int) *feature.Feature {
+	sink := "sink"
+	f := feature.NewFeatureNamed("KafkaSources")
+
+	f.Setup("install a sink", svc.Install(sink, "app", "rekt"))
+
+	for i := 0; i < n; i++ {
+		topicName := feature.MakeRandomK8sName("topic") // A k8s name is also a valid topic name.
+		name := fmt.Sprintf("%s%d", prefix, i)
+
+		f.Setup("install kafka topic", kafkatopic.Install(topicName))
+		f.Setup(fmt.Sprintf("install kafkasource %s", name), kafkasource.Install(
+			name,
+			kafkasource.WithBootstrapServers(testingpkg.BootstrapServersPlaintextArr),
+			kafkasource.WithTopics([]string{topicName}),
+			kafkasource.WithSink(&duckv1.KReference{Kind: "Service", Name: sink, APIVersion: "v1"}, ""),
+		))
+
+		f.Assert(fmt.Sprintf("kafkasource %s is ready", name), kafkasource.IsReady(name))
+	}
+
+	return f
+}
+
+func KafkaSourcesAreNotPresentInContractConfigMaps(prefix string) *feature.Feature {
+	f := feature.NewFeatureNamed("KafkaSources are not present in Contract configmaps")
+
+	f.Assert("deleted KafkaSources are not present in Contract CMs", deletedKafkaSourcesAreNotPresentInContractConfigMaps(prefix))
+
+	return f
+}
+
+func deletedKafkaSourcesAreNotPresentInContractConfigMaps(prefix string) feature.StepFn {
+	return func(ctx context.Context, t feature.T) {
+
+		namespace := environment.FromContext(ctx).Namespace()
+
+		kss, err := kafkaclient.Get(ctx).SourcesV1beta1().
+			KafkaSources(namespace).
+			List(ctx, metav1.ListOptions{
+				Limit: 2000,
+			})
+		if err != nil {
+			t.Fatal("Failed to list KafkaSources")
+		}
+
+		systemNamespaceCMs, err := kubeclient.Get(ctx).CoreV1().
+			ConfigMaps(knative.KnativeNamespaceFromContext(ctx)).
+			List(ctx, metav1.ListOptions{
+				Limit: 100,
+			})
+		if err != nil {
+			t.Fatal(err)
+		}
+		for _, cm := range systemNamespaceCMs.Items {
+			if !strings.HasPrefix(cm.Name, "kafka-source-dispatcher") {
+				continue
+			}
+
+			ct, err := base.GetDataPlaneConfigMapData(zap.NewNop(), &cm, base.Json)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			for _, r := range ct.Resources {
+				if r.Reference.Namespace != namespace {
+					continue
+				}
+				if !strings.HasPrefix(r.Reference.Name, prefix) {
+					continue
+				}
+
+				found := false
+				for _, s := range kss.Items {
+					if r.Reference.Namespace == s.Namespace && r.Reference.Name == s.Name {
+						found = true
+					}
+				}
+				if !found {
+					t.Errorf("%v", ResourceError{
+						Resource: r,
+						Sources:  kss,
+					}.Error())
+				}
+			}
+		}
+	}
+}
+
+type ResourceError struct {
+	Resource *contract.Resource       `json:"resource"`
+	Sources  *sources.KafkaSourceList `json:"sources"`
+}
+
+func (r ResourceError) Error() string {
+	bytes, _ := json.Marshal(r)
+	return string(bytes)
+}
diff --git a/test/rekt/features/namespace.go b/test/rekt/features/namespace.go
new file mode 100644
index 000000000..3ad2798b7
--- /dev/null
+++ b/test/rekt/features/namespace.go
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2023 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package features
+
+import (
+	"context"
+	"fmt"
+
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	"knative.dev/reconciler-test/pkg/feature"
+)
+
+func SetupNamespace(name string) *feature.Feature {
+	f := feature.NewFeatureNamed("setup namespace")
+
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+
+	f.Setup(fmt.Sprintf("install namespace %s", name), func(ctx context.Context, t feature.T) {
+		_, err := kubeclient.Get(ctx).CoreV1().Namespaces().Create(ctx, ns, metav1.CreateOptions{})
+		if err != nil && !apierrors.IsAlreadyExists(err) {
+			t.Fatal(err)
+		}
+
+	})
+
+	return f
+}
+
+func CleanupNamespace(name string) *feature.Feature {
+	f := feature.NewFeatureNamed("delete namespace")
+
+	f.Setup(fmt.Sprintf("install namespace %s", name), func(ctx context.Context, t feature.T) {
+		err := kubeclient.Get(ctx).CoreV1().Namespaces().Delete(ctx, name, metav1.DeleteOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			t.Fatal(err)
+		}
+
+	})
+
+	return f
+}