Repository: knative-sandbox/eventing-kafka-broker
Pull request #2627 (2022-09-08 22:08:58)

[release-1.6] Set TerminationMessagePolicy=FallbackToLogsOnError in a test image
This is an automated cherry-pick of #2622

```release-note

```

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/2627#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign odacremolbap for approval by writing `/assign @odacremolbap` 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.6/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":["odacremolbap"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2627](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (4c24f8e) into [release-1.6](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/ba8b04000765dd45bdec7c47f1b83cea4950b926?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ba8b040) will **increase** coverage by `22.03%`.
> The diff coverage is `n/a`.

```diff
@@                Coverage Diff                 @@
##             release-1.6    #2627       +/-   ##
==================================================
+ Coverage          59.85%   81.88%   +22.03%     
- Complexity             0      724      +724     
==================================================
  Files                 69       72        +3     
  Lines               6997     2501     -4496     
  Branches               0      228      +228     
==================================================
- Hits                4188     2048     -2140     
+ Misses              2512      328     -2184     
+ Partials             297      125      -172     
```

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

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/2627?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/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [...ol-plane/pkg/reconciler/channel/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | | |
| [...als/kafka/eventing/v1alpha1/consumer\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/core/config/egress.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvZWdyZXNzLmdv) | | |
| [...trol-plane/pkg/reconciler/consumergroup/evictor.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2V2aWN0b3IuZ28=) | | |
| [control-plane/pkg/receiver/address.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjZWl2ZXIvYWRkcmVzcy5nbw==) | | |
| [control-plane/pkg/kafka/broker\_class.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cva2Fma2EvYnJva2VyX2NsYXNzLmdv) | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | |
| [control-plane/pkg/reconciler/consumergroup/auth.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2F1dGguZ28=) | | |
| [control-plane/pkg/reconciler/broker/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvY29udHJvbGxlci5nbw==) | | |
| ... and [131 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2627/diff?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 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
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_release-1.6 | 4c24f8e5e4f06a68f1b3a56fdd9a17033d00736f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2627/channel-integration-tests-ssl_eventing-kafka-broker_release-1.6/1567998462685876224) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_release-1.6`

[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
--- | --- | --- | --- | ---
channel-integration-tests-ssl_eventing-kafka-broker_release-1.6 | 4c24f8e5e4f06a68f1b3a56fdd9a17033d00736f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2627/channel-integration-tests-ssl_eventing-kafka-broker_release-1.6/1567998462685876224) | true | `/test channel-integration-tests-ssl_eventing-kafka-broker_release-1.6`
integration-tests_eventing-kafka-broker_release-1.6 | 4c24f8e5e4f06a68f1b3a56fdd9a17033d00736f | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2627/integration-tests_eventing-kafka-broker_release-1.6/1567998462492938240) | true | `/test integration-tests_eventing-kafka-broker_release-1.6`

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

<details>

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

>/cherry-pick release-1.5


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>
aliok: /retest-required

Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 817f85797..fe73316c1 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -32,7 +32,7 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
-	eventingcorev1beta1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 
@@ -49,10 +49,7 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"):    &eventingv1alpha1.KafkaSink{},
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):    &sourcesv1beta1.KafkaSource{},
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
-}
-
-var validationCallbacks = map[schema.GroupVersionKind]validation.Callback{
-	eventingcorev1beta1.SchemeGroupVersion.WithKind("Broker"): eventingv1.BrokerValidationCallback(),
+	eventingcorev1.SchemeGroupVersion.WithKind("Broker"):         &eventingv1.BrokerStub{},
 }
 
 var defaultingCallbacks = map[schema.GroupVersionKind]defaulting.Callback{
@@ -131,9 +128,6 @@ func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher)
 
 		// Whether to disallow unknown fields.
 		true,
-
-		// Extra validating callbacks to be applied to resources.
-		validationCallbacks,
 	)
 }
 
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation.go b/control-plane/pkg/apis/eventing/v1/broker_validation.go
index fcfd551b8..88465108a 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation.go
@@ -21,25 +21,39 @@ import (
 	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/webhook"
-	"knative.dev/pkg/webhook/resourcesemantics/validation"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/pkg/webhook/resourcesemantics"
 )
 
-type BrokerPartial struct {
+type BrokerStub struct {
+	metav1.TypeMeta `json:",inline"`
+
 	// +optional
 	metav1.ObjectMeta `json:"metadata,omitempty"`
-	// Spec defines the desired state of the Broker.
-	Spec BrokerPartialSpec `json:"spec,omitempty"`
+
+	Spec eventing.BrokerSpec `json:"spec,omitempty"`
+
+	// +optional
+	Status eventing.BrokerStatus `json:"status,omitempty"`
 }
 
-type BrokerPartialSpec struct {
+var (
+	// Check that Broker is resourcesemantics.GenericCRD.
+	// We do this to make sure the validating/defaulting webhooks can operate on our stub.
+	// If we don't do this stubbing, we will end up doing the defaulting/validating using the upstream Broker type.
+	// However, we want to let the upstream webhook to do its job and not us. We only want to default/validate
+	// the parts relevant to us.
+	_ resourcesemantics.GenericCRD = (*BrokerStub)(nil)
+)
+
+type BrokerStubSpec struct {
 	// Config is a KReference to the configuration that specifies
 	// configuration options for this Broker. For example, this could be
 	// a pointer to a ConfigMap.
@@ -47,27 +61,55 @@ type BrokerPartialSpec struct {
 	Config *duckv1.KReference `json:"config,omitempty"`
 }
 
-func (b BrokerPartialSpec) Validate(context.Context) error {
-	if b.Config == nil {
-		return apis.ErrMissingField("config").ViaField("spec")
+func (b *BrokerStub) Validate(context.Context) *apis.FieldError {
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass && b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.NamespacedBrokerClass {
+	if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass {
+		// validation for the broker of other classes is done by the other webhooks
+		return nil
+	}
+
+	if b.Spec.Config == nil {
+		// validation of the Config requiredness is done by the upstream webhook and we want to let it do its job.
+		// otherwise API server will say that our Kafka admission controller refused it, but it is a generic validation.
+		return nil
 	}
-	if strings.ToLower(b.Config.Kind) != "configmap" {
-		return apis.ErrInvalidValue(b.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
+
+	// we specifically expect our broker class to use ConfigMap as the config type
+	if strings.ToLower(b.Spec.Config.Kind) != "configmap" {
+		return apis.ErrInvalidValue(b.Spec.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
 	}
+
+	// we specifically expect for our broker classes, that there's a namespace in the config
+	if b.Spec.Config.Namespace == "" {
+		return apis.ErrMissingField(b.Spec.Config.Namespace, "namespace").ViaField("config").ViaField("spec")
+	}
+
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// for the namespaced broker, we expect the config to be in the same namespace as the broker
+	//if b.Annotations[eventing.BrokerClassAnnotationKey] == kafka.NamespacedBrokerClass && b.Spec.Config.Namespace != b.Namespace {
+	//	return apis.ErrInvalidValue(b.Spec.Config.Namespace, "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec")
+	//}
+
 	return nil
 }
 
-func validateBrokerFromUnstructured(ctx context.Context, unstructured *unstructured.Unstructured) error {
-	broker := BrokerPartial{}
-	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &broker); err != nil {
-		return err
-	}
-	if class, ok := broker.Annotations[eventing.BrokerClassAnnotationKey]; !ok || class != kafka.BrokerClass {
+func (b *BrokerStub) SetDefaults(ctx context.Context) {
+	// do nothing, we let the upstream webhook do its job
+}
+
+func (b *BrokerStub) DeepCopyObject() runtime.Object {
+	if b == nil {
 		return nil
 	}
-	return broker.Spec.Validate(ctx)
-}
 
-func BrokerValidationCallback() validation.Callback {
-	return validation.NewCallback(validateBrokerFromUnstructured, webhook.Create, webhook.Update)
+	out := &BrokerStub{
+		TypeMeta: b.TypeMeta, //simple struct
+	}
+
+	b.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	b.Spec.DeepCopyInto(&out.Spec)
+	b.Status.DeepCopyInto(&out.Status)
+
+	return out
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
index f5254cb07..6d51db9e4 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/metrics/Metrics.java
@@ -15,15 +15,18 @@
  */
 package dev.knative.eventing.kafka.broker.core.metrics;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.Gauge;
+import io.micrometer.core.instrument.Meter;
 import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
+import io.micrometer.core.instrument.search.Search;
 import io.micrometer.prometheus.PrometheusConfig;
 import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.Future;
@@ -42,6 +45,7 @@ import org.apache.kafka.clients.producer.Producer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
@@ -293,6 +297,18 @@ public class Metrics {
       .baseUnit(Metrics.Units.DIMENSIONLESS);
   }
 
+  public static Collection<Meter> searchResourceMeters(final MeterRegistry registry, final DataPlaneContract.Reference ref) {
+    return Search.in(registry)
+      .tags(resourceRefTags(ref))
+      .meters();
+  }
+
+  public static Collection<Meter> searchEgressMeters(final MeterRegistry registry, final DataPlaneContract.Reference ref) {
+    return Search.in(registry)
+      .tags(egressRefTags(ref))
+      .meters();
+  }
+
   public static DistributionSummary.Builder eventDispatchLatency(final io.micrometer.core.instrument.Tags tags) {
     return DistributionSummary
       .builder(EVENT_DISPATCH_LATENCY)
@@ -337,4 +353,18 @@ public class Metrics {
       .tags(tags)
       .baseUnit(Metrics.Units.DIMENSIONLESS);
   }
+
+  public static io.micrometer.core.instrument.Tags resourceRefTags(final DataPlaneContract.Reference ref) {
+    return io.micrometer.core.instrument.Tags.of(
+      Tag.of(Metrics.Tags.RESOURCE_NAME, ref.getName()),
+      Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, ref.getNamespace())
+    );
+  }
+
+  public static io.micrometer.core.instrument.Tags egressRefTags(final DataPlaneContract.Reference ref) {
+    return io.micrometer.core.instrument.Tags.of(
+      Tag.of(Metrics.Tags.CONSUMER_NAME, ref.getName()),
+      Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, ref.getNamespace())
+    );
+  }
 }
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/IngressReconcilerListener.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/IngressReconcilerListener.java
index 58f30784b..fbf78238d 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/IngressReconcilerListener.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/IngressReconcilerListener.java
@@ -16,8 +16,13 @@
 package dev.knative.eventing.kafka.broker.core.reconciler;
 
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
+import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
 public interface IngressReconcilerListener {
 
   Future<Void> onNewIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress);
@@ -26,4 +31,44 @@ public interface IngressReconcilerListener {
 
   Future<Void> onDeleteIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress);
 
+  static IngressReconcilerListener all(final IngressReconcilerListener... listeners) {
+    return new CompositeIngressReconcilerLister(listeners);
+  }
+}
+
+class CompositeIngressReconcilerLister implements IngressReconcilerListener {
+
+  private IngressReconcilerListener[] listeners;
+
+  public CompositeIngressReconcilerLister(final IngressReconcilerListener... listeners) {
+    Objects.requireNonNull(listeners);
+    this.listeners = listeners;
+  }
+
+  @Override
+  public Future<Void> onNewIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    return CompositeFuture.all(
+      Arrays.stream(listeners)
+        .map(l -> l.onNewIngress(resource, ingress))
+        .collect(Collectors.toList())
+    ).mapEmpty();
+  }
+
+  @Override
+  public Future<Void> onUpdateIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    return CompositeFuture.all(
+      Arrays.stream(listeners)
+        .map(l -> l.onUpdateIngress(resource, ingress))
+        .collect(Collectors.toList())
+    ).mapEmpty();
+  }
+
+  @Override
+  public Future<Void> onDeleteIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    return CompositeFuture.all(
+      Arrays.stream(listeners)
+        .map(l -> l.onDeleteIngress(resource, ingress))
+        .collect(Collectors.toList())
+    ).mapEmpty();
+  }
 }
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
index aa4acc35f..e3fa0b59e 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/RecordDispatcherImpl.java
@@ -504,6 +504,10 @@ public class RecordDispatcherImpl implements RecordDispatcher {
   public Future<Void> close() {
     this.closed.set(true);
 
+    Metrics
+      .searchEgressMeters(meterRegistry, resourceContext.getEgress().getReference())
+      .forEach(meterRegistry::remove);
+
     if (inFlightEvents.get() == 0) {
       closePromise.tryComplete();
     }
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressRequestHandler.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressRequestHandler.java
index 9fab7b06c..c2a2877c0 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressRequestHandler.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressRequestHandler.java
@@ -15,10 +15,12 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
+import dev.knative.eventing.kafka.broker.core.reconciler.IngressReconcilerListener;
+
 /**
  * This class handles incoming ingress requests.
  */
-public interface IngressRequestHandler {
+public interface IngressRequestHandler extends IngressReconcilerListener {
 
   void handle(RequestContext request, IngressProducer producer);
 
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
index f817b775d..82bc7033d 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticle.java
@@ -15,6 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.receiver.impl;
 
+import dev.knative.eventing.kafka.broker.core.reconciler.IngressReconcilerListener;
 import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import dev.knative.eventing.kafka.broker.receiver.IngressProducer;
 import dev.knative.eventing.kafka.broker.receiver.IngressRequestHandler;
@@ -87,7 +88,7 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     this.ingressProducerStore = this.ingressProducerStoreFactory.apply(vertx);
     this.messageConsumer = ResourcesReconciler
       .builder()
-      .watchIngress(this.ingressProducerStore)
+      .watchIngress(IngressReconcilerListener.all(this.ingressProducerStore, this.ingressRequestHandler))
       .buildAndListen(vertx);
 
     this.server = vertx.createHttpServer(httpServerOptions);
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
index fd0448664..6f85672fe 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImpl.java
@@ -15,7 +15,9 @@
  */
 package dev.knative.eventing.kafka.broker.receiver.impl.handler;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
+import dev.knative.eventing.kafka.broker.core.reconciler.IngressReconcilerListener;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingConfig;
 import dev.knative.eventing.kafka.broker.core.tracing.TracingSpan;
 import dev.knative.eventing.kafka.broker.receiver.IngressProducer;
@@ -44,7 +46,7 @@ import static io.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAVAILABLE
  * <p>
  * Instances of this class can be shared among verticles, given the {@code requestToRecordMapper} provided is shareable among verticles.
  */
-public class IngressRequestHandlerImpl implements IngressRequestHandler {
+public class IngressRequestHandlerImpl implements IngressRequestHandler, IngressReconcilerListener {
 
   static final Tag UNKNOWN_EVENT_TYPE_TAG = Tag.of(Metrics.Tags.EVENT_TYPE, "unknown");
 
@@ -81,10 +83,7 @@ public class IngressRequestHandlerImpl implements IngressRequestHandler {
   @Override
   public void handle(final RequestContext requestContext, final IngressProducer producer) {
 
-    final var resourceTags = Tags.of(
-      Tag.of(Metrics.Tags.RESOURCE_NAME, producer.getReference().getName()),
-      Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, producer.getReference().getNamespace())
-    );
+    final Tags resourceTags = Metrics.resourceRefTags(producer.getReference());
 
     requestToRecordMapper
       .requestToRecord(requestContext.getRequest(), producer.getTopic())
@@ -171,4 +170,19 @@ public class IngressRequestHandlerImpl implements IngressRequestHandler {
       });
   }
 
+  @Override
+  public Future<Void> onNewIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    return Future.succeededFuture();
+  }
+
+  @Override
+  public Future<Void> onUpdateIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    return Future.succeededFuture();
+  }
+
+  @Override
+  public Future<Void> onDeleteIngress(DataPlaneContract.Resource resource, DataPlaneContract.Ingress ingress) {
+    Metrics.searchResourceMeters(meterRegistry, resource.getReference()).forEach(meterRegistry::remove);
+    return Future.succeededFuture();
+  }
 }


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
index 0c08ba763..207dabf26 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
@@ -20,88 +20,124 @@ import (
 	"context"
 	"testing"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
-func TestValidateBrokerFromUnstructured(t *testing.T) {
+func TestValidate(t *testing.T) {
 	tests := []struct {
-		name         string
-		ctx          context.Context
-		unstructured *unstructured.Unstructured
-		wantErr      bool
-	}{
-		{
-			name: "no kafka broker",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{"spec": map[string]interface{}{}},
+		name string
+		b    BrokerStub
+		want *apis.FieldError
+	}{{
+		name: "missing annotation",
+		b:    BrokerStub{},
+	}, {
+		name: "empty annotation",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": ""},
 			},
-			wantErr: false,
 		},
-		{
-			name: "missing config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{},
+	}, {
+		name: "other broker class",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Foo"},
+			},
+		},
+	}, {
+		name: "no spec.config",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+		},
+	}, {
+		name: "no spec.config.namespace",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind: "ConfigMap",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "unknown kind",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "eventing.knative.dev/v1",
-							"kind":       "Broker",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrMissingField("", "namespace").ViaField("config").ViaField("spec"),
+	}, {
+		name: "spec.config is not configmap",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind:      "Service",
+					Namespace: "foo",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "ConfigMap in config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "v1",
-							"kind":       "ConfigMap",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrInvalidValue("Service", "kind", "Expected ConfigMap").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "spec.config.namespace is different",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Kind:      "ConfigMap",
+		//				Namespace: "foo",
+		//			},
+		//		},
+		//	},
+		//	want: apis.ErrInvalidValue("foo", "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "valid config - namespaced broker",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Namespace:  "my-namespace",
+		//				Name:       "name",
+		//				Kind:       "ConfigMap",
+		//				APIVersion: "v1",
+		//			},
+		//		},
+		//	},
+	}, {
+		name: "valid config - regular broker",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:   "my-namespace",
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Name:       "name",
+					Namespace:  "my-namespace",
+					Kind:       "ConfigMap",
+					APIVersion: "v1",
 				},
 			},
-			wantErr: false,
 		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := validateBrokerFromUnstructured(tt.ctx, tt.unstructured); (err != nil) != tt.wantErr {
-				t.Errorf("validateBrokerFromUnstructured() error = %v, wantErr %v", err, tt.wantErr)
+	}}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.b.Validate(context.Background())
+			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Error("Broker.Validate (-want, +got) =", diff)
 			}
 		})
 	}
diff --git a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/metrics/MetricsTest.java b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/metrics/MetricsTest.java
index 7ce223e64..943fa3d0c 100644
--- a/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/metrics/MetricsTest.java
+++ b/data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/metrics/MetricsTest.java
@@ -17,7 +17,10 @@ package dev.knative.eventing.kafka.broker.core.metrics;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.utils.BaseEnv;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
@@ -26,9 +29,13 @@ import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.clients.producer.MockProducer;
+import org.assertj.core.api.AssertionInfo;
+import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import java.util.stream.Collectors;
+
 @ExtendWith(VertxExtension.class)
 public class MetricsTest {
 
@@ -55,4 +62,80 @@ public class MetricsTest {
     final var meterBinder = Metrics.register(new MockConsumer<>(OffsetResetStrategy.LATEST));
     meterBinder.close();
   }
+
+  @Test
+  public void searchShouldReturnMetersWithAdditionalTags() {
+
+    final var registry = Metrics.getRegistry();
+
+    Metrics.eventCount(Tags.of(
+        Tag.of(Metrics.Tags.EVENT_TYPE, "type"),
+        Tag.of(Metrics.Tags.RESOURCE_NAME, "name"),
+        Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, "namespace")
+      ))
+      .register(registry)
+      .increment();
+
+    Metrics.eventDispatchLatency(Tags.of(
+        Tag.of(Metrics.Tags.EVENT_TYPE, "type"),
+        Tag.of(Metrics.Tags.RESOURCE_NAME, "name"),
+        Tag.of(Metrics.Tags.CONSUMER_NAME, "cname"),
+        Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, "namespace")
+      ))
+      .register(registry)
+      .record(10);
+
+    Metrics.eventProcessingLatency(Tags.of(
+        Tag.of(Metrics.Tags.EVENT_TYPE, "type"),
+        Tag.of(Metrics.Tags.RESOURCE_NAME, "name"),
+        Tag.of(Metrics.Tags.CONSUMER_NAME, "cname"),
+        Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, "namespace")
+      ))
+      .register(registry)
+      .record(10);
+
+    Metrics.eventDispatchLatency(Tags.of(
+        Tag.of(Metrics.Tags.EVENT_TYPE, "type"),
+        Tag.of(Metrics.Tags.RESOURCE_NAME, "name"),
+        Tag.of(Metrics.Tags.CONSUMER_NAME, "cname"),
+        Tag.of(Metrics.Tags.RESOURCE_NAMESPACE, "other-namespace")
+      ))
+      .register(registry)
+      .record(10);
+
+    final var expectedResourceMetersCount = 3;
+    final var expectedEgressMetersCount = 2;
+
+    final var resourceMeters = Metrics.searchResourceMeters(registry, DataPlaneContract.Reference.newBuilder()
+      .setName("name")
+      .setNamespace("namespace")
+      .build()
+    );
+
+    final var egressMeters = Metrics.searchEgressMeters(registry, DataPlaneContract.Reference.newBuilder()
+      .setName("cname")
+      .setNamespace("namespace")
+      .build()
+    );
+
+    Assertions.assertThat(resourceMeters).hasSize(expectedResourceMetersCount);
+    Assertions.assertThat(egressMeters).hasSize(expectedEgressMetersCount);
+
+    Assertions.assertThat(
+      resourceMeters.stream().filter(m -> m.getId().getName().equals(Metrics.EVENTS_COUNT)).collect(Collectors.toList())
+    ).hasSize(1);
+    Assertions.assertThat(
+      resourceMeters.stream().filter(m -> m.getId().getName().equals(Metrics.EVENT_DISPATCH_LATENCY)).collect(Collectors.toList())
+    ).hasSize(1);
+    Assertions.assertThat(
+      resourceMeters.stream().filter(m -> m.getId().getName().equals(Metrics.EVENT_PROCESSING_LATENCY)).collect(Collectors.toList())
+    ).hasSize(1);
+
+    Assertions.assertThat(
+      egressMeters.stream().filter(m -> m.getId().getName().equals(Metrics.EVENT_DISPATCH_LATENCY)).collect(Collectors.toList())
+    ).hasSize(1);
+    Assertions.assertThat(
+      egressMeters.stream().filter(m -> m.getId().getName().equals(Metrics.EVENT_PROCESSING_LATENCY)).collect(Collectors.toList())
+    ).hasSize(1);
+  }
 }
diff --git a/test/pkg/kafka/verify_num_partitions_replication.go b/test/pkg/kafka/verify_num_partitions_replication.go
index d6198482c..0e5e03ff7 100644
--- a/test/pkg/kafka/verify_num_partitions_replication.go
+++ b/test/pkg/kafka/verify_num_partitions_replication.go
@@ -66,9 +66,10 @@ func VerifyNumPartitionAndReplicationFactor(
 				Spec: corev1.PodSpec{
 					Containers: []corev1.Container{
 						{
-							Name:            name,
-							Image:           pkgtest.ImagePath(partitionReplicationVerifierImage),
-							ImagePullPolicy: corev1.PullIfNotPresent,
+							Name:                     name,
+							Image:                    pkgtest.ImagePath(partitionReplicationVerifierImage),
+							ImagePullPolicy:          corev1.PullIfNotPresent,
+							TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError,
 							Env: []corev1.EnvVar{
 								{
 									Name:  "BOOTSTRAP_SERVERS",