Repository: knative-sandbox/eventing-kafka-broker
Pull request #2888 (2023-01-12 17:12:35)

[release-1.8] Fix Consumer resource finalization
This is an automated cherry-pick of #2887

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2888#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign evankanderson for approval by writing `/assign @evankanderson` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

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/release-1.8/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":["evankanderson"]} -->
knative-prow[bot]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/unit-tests_eventing-kafka-broker_release-1.8/1613584748318298112) | true | `/test unit-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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/unit-tests_eventing-kafka-broker_release-1.8/1613584748318298112) | true | `/test unit-tests`
build-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/build-tests_eventing-kafka-broker_release-1.8/1613584748259577856) | 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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2888](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (705f965) into [release-1.8](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/95ba3a6448cc02111c7edba582d400a1d50fc21a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (95ba3a6) will **increase** coverage by `22.94%`.
> The diff coverage is `n/a`.

```diff
@@                Coverage Diff                 @@
##             release-1.8    #2888       +/-   ##
==================================================
+ Coverage          58.04%   80.98%   +22.94%     
- Complexity             0      747      +747     
==================================================
  Files                 75       77        +2     
  Lines               8075     2619     -5456     
  Branches               0      231      +231     
==================================================
- Hits                4687     2121     -2566     
+ Misses              3031      357     -2674     
+ Partials             357      141      -216     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `80.98% <ø> (?)` | |

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/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/reconciler/consumer/consumer.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lci9jb25zdW1lci5nbw==) | | |
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | | |
| [control-plane/pkg/contract/subscriptionsapi.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3Qvc3Vic2NyaXB0aW9uc2FwaS5nbw==) | | |
| [control-plane/pkg/receiver/address.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvYWRkcmVzcy5nbw==) | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | |
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [...ane/pkg/apis/eventing/v1alpha1/kafka\_sink\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3R5cGVzLmdv) | | |
| [...ne/pkg/reconciler/trigger/namespaced\_controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL25hbWVzcGFjZWRfY29udHJvbGxlci5nbw==) | | |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | | |
| ... and [142 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2888?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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/unit-tests_eventing-kafka-broker_release-1.8/1613584748318298112) | true | `/test unit-tests`
build-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/build-tests_eventing-kafka-broker_release-1.8/1613584748259577856) | true | `/test build-tests`
reconciler-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/reconciler-tests_eventing-kafka-broker_release-1.8/1613584748632870912) | true | `/test reconciler-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]: @knative-prow-robot: 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
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/unit-tests_eventing-kafka-broker_release-1.8/1613584748318298112) | true | `/test unit-tests`
build-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/build-tests_eventing-kafka-broker_release-1.8/1613584748259577856) | true | `/test build-tests`
reconciler-tests_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/reconciler-tests_eventing-kafka-broker_release-1.8/1613584748632870912) | true | `/test reconciler-tests`
reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/reconciler-tests-namespaced-broker_eventing-kafka-broker_release-1.8/1613584748695785472) | true | `/test reconciler-tests-namespaced-broker`
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8 | 705f965ee9cce8d2f624ac0f8caa9cf839d8d4fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2888/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8/1613584748884529152) | 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 -->
pierDipi: /cherry-pick release-1.7
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.7 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2888#issuecomment-1381449892):

>/cherry-pick release-1.7


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>
pierDipi: /cherry-pick release-1.6
knative-prow-robot: @pierDipi: once the present PR merges, I will cherry-pick it on top of release-1.6 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2888#issuecomment-1381449982):

>/cherry-pick release-1.6


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>
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/2888#" title="Author self-approved">knative-prow-robot</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2888#pullrequestreview-1247217905" title="Approved">pierDipi</a>*

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.8/OWNERS)~~ [pierDipi]

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

Patch:
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
index c103cd4c1..bf1121fa2 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,
@@ -404,3 +404,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 e254b3723..8514f3e4a 100644
--- a/control-plane/pkg/reconciler/testing/objects_consumer.go
+++ b/control-plane/pkg/reconciler/testing/objects_consumer.go
@@ -344,3 +344,9 @@ func ConsumerGroupAsOwnerRef() metav1.OwnerReference {
 		BlockOwnerDeletion: pointer.BoolPtr(true),
 	}
 }
+
+func ConsumerDeletedTimeStamp() ConsumerOption {
+	return func(c *kafkainternals.Consumer) {
+		WithDeletedTimeStamp(c)
+	}
+}
diff --git a/test/e2e_new/features/kafka_source.go b/test/e2e_new/features/kafka_source.go
new file mode 100644
index 000000000..d8142bfc3
--- /dev/null
+++ b/test/e2e_new/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/test/rekt/resources/kafkasource"
+	"knative.dev/eventing-kafka/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/e2e_new/features/namespace.go b/test/e2e_new/features/namespace.go
new file mode 100644
index 000000000..3ad2798b7
--- /dev/null
+++ b/test/e2e_new/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
+}
diff --git a/test/e2e_new/kafka_source_test.go b/test/e2e_new/kafka_source_test.go
index bc5dc7f20..4fad559a4 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-"))
+}