Repository: knative-sandbox/eventing-kafka-broker
Pull request #1733 (2022-01-12 20:02:11)

[data-plane]: Bump maven-compiler-plugin from 3.8.1 to 3.9.0 in /data-plane
Bumps [maven-compiler-plugin](https://github.com/apache/maven-compiler-plugin) from 3.8.1 to 3.9.0.
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/aeb15b6cf38ee44391febf23310a4f2b432ab444"><code>aeb15b6</code></a> [maven-release-plugin] prepare release maven-compiler-plugin-3.9.0</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/6335382f97aad2e69b9f9a0a4af3cf801cd29273"><code>6335382</code></a> Shared GitHub Acton v2</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/8d5d3cd68db3cefb947e1e1bf5fd83ac231ef00d"><code>8d5d3cd</code></a> Fix site build</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/ce4eb1e353d176f8573f7a3ba893fc3ec42ec21b"><code>ce4eb1e</code></a> Bump plexus-component-metadata from 2.1.0 to 2.1.1</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/f875750ee9bcf07bb25b470c5ceb73d7dccfbe12"><code>f875750</code></a> Bump mockito-core from 4.1.0 to 4.2.0</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/5463357ebc1c65a4e25bfd8ea4c453a618b66bf1"><code>5463357</code></a> fix CI site goal</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/859c9039e4b73367a0d9486cc63b98ec1cc1fa6f"><code>859c903</code></a> Update plugins</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/b0de9bc0fbe9416e81b6a13cb9e67fdf891192af"><code>b0de9bc</code></a> Bump mockito-core from 4.0.0 to 4.1.0</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/f95dd463d904276564ac47ee822c37ac8751f4e9"><code>f95dd46</code></a> Bump plexusCompilerVersion from 2.8.8 to 2.9.0</li>
<li><a href="https://github.com/apache/maven-compiler-plugin/commit/26900cfe598981a516b66e01843e79f1c486a147"><code>26900cf</code></a> Bump mockito-core from 2.28.2 to 4.0.0</li>
<li>Additional commits viewable in <a href="https://github.com/apache/maven-compiler-plugin/compare/maven-compiler-plugin-3.8.1...maven-compiler-plugin-3.9.0">compare view</a></li>
</ul>
</details>
<br />


[![Dependabot compatibility score](https://dependabot-badges.githubapp.com/badges/compatibility_score?dependency-name=org.apache.maven.plugins:maven-compiler-plugin&package-manager=maven&previous-version=3.8.1&new-version=3.9.0)](https://docs.github.com/en/github/managing-security-vulnerabilities/about-dependabot-security-updates#about-compatibility-scores)

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

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

---

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

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


</details>

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

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

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

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

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/main/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":["psschwei"]} -->
knative-prow-robot: Hi @dependabot[bot]. Thanks for your PR.

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

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

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

<details>

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

codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1733](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (26376cb) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/e8208b204f6ab9ee99320b419c0d66697c8bf9aa?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (e8208b2) will **increase** coverage by `0.04%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733/graphs/tree.svg?width=650&height=150&src=pr&token=xyPIgGOiwy&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@             Coverage Diff              @@
##               main    #1733      +/-   ##
============================================
+ Coverage     72.67%   72.71%   +0.04%     
  Complexity      604      604              
============================================
  Files           124      124              
  Lines          4604     4604              
  Branches        176      176              
============================================
+ Hits           3346     3348       +2     
+ Misses          983      982       -1     
+ Partials        275      274       -1     
```

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

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/1733?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...lane/pkg/reconciler/consumergroup/consumergroup.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnN1bWVyZ3JvdXAuZ28=) | `79.26% <0.00%> (+2.43%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [e8208b2...26376cb](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1733?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
new file mode 100644
index 000000000..bbc185a6e
--- /dev/null
+++ b/control-plane/pkg/apis/internals/kafka/eventing/data_plane.go
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021 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 eventing
+
+const (
+	// ConfigMapVolumeName is the volume name of the data plane ConfigMap
+	ConfigMapVolumeName = "kafka-resources"
+)
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
index 6e5441add..8e0210fe3 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation.go
@@ -42,5 +42,9 @@ func (cgs *ConsumerGroupSpec) Validate(ctx context.Context) *apis.FieldError {
 }
 
 func (cts *ConsumerTemplateSpec) Validate(ctx context.Context) *apis.FieldError {
-	return cts.Spec.Validate(ctx).ViaField("spec")
+	specCtx := ctx
+	if apis.IsInUpdate(ctx) {
+		specCtx = apis.WithinUpdate(ctx, apis.GetBaseline(ctx).(*ConsumerGroup).Spec.Template.Spec)
+	}
+	return cts.Spec.Validate(specCtx).ViaField("spec")
 }
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_lifecycle.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_lifecycle.go
index b3086a368..c8a1d344a 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_lifecycle.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_lifecycle.go
@@ -17,9 +17,48 @@
 package v1alpha1
 
 import (
+	"fmt"
+
 	"knative.dev/pkg/apis"
+	"knative.dev/pkg/reconciler"
+)
+
+const (
+	ConsumerConditionContract = "Contract"
+	ConsumerConditionBind     = "Bind"
+)
+
+var (
+	consumerConditionSet = apis.NewLivingConditionSet(
+		ConsumerConditionContract,
+		ConsumerConditionBind,
+	)
 )
 
 func (c *Consumer) GetConditionSet() apis.ConditionSet {
-	return apis.NewLivingConditionSet()
+	return consumerConditionSet
+}
+
+func (c *Consumer) MarkReconcileContractFailed(err error) reconciler.Event {
+	err = fmt.Errorf("failed to reconcile contract: %w", err)
+	c.GetConditionSet().Manage(c.GetStatus()).MarkFalse(ConsumerConditionContract, "ReconcileContract", err.Error())
+	return err
+}
+
+func (c *Consumer) MarkReconcileContractSucceeded() {
+	c.GetConditionSet().Manage(c.GetStatus()).MarkTrue(ConsumerConditionContract)
+}
+
+func (c *Consumer) MarkBindFailed(err error) reconciler.Event {
+	err = fmt.Errorf("failed to bind resource to pod: %w", err)
+	c.GetConditionSet().Manage(c.GetStatus()).MarkFalse(ConsumerConditionBind, "ConsumerBinding", err.Error())
+	return err
+}
+
+func (c *Consumer) MarkBindInProgress() {
+	c.GetConditionSet().Manage(c.GetStatus()).MarkFalse(ConsumerConditionBind, "BindInProgress", "")
+}
+
+func (c *Consumer) MarkBindSucceeded() {
+	c.GetConditionSet().Manage(c.GetStatus()).MarkTrue(ConsumerConditionBind)
 }
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
index 7d5aa4d0f..6bac11ace 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_types.go
@@ -75,6 +75,10 @@ type ConsumerSpec struct {
 	// +optional
 	Delivery *DeliverySpec `json:"delivery,omitempty"`
 
+	// Reply is the strategy to handle event replies.
+	// +optional
+	Reply *ReplyStrategy `json:"reply,omitempty"`
+
 	// Filters is a set of filters.
 	// +optional
 	Filters *Filters `json:"filters,omitempty"`
@@ -94,6 +98,19 @@ type ConsumerSpec struct {
 	PodBind *PodBind `json:"podBind"`
 }
 
+type ReplyStrategy struct {
+	TopicReply *TopicReply
+	URLReply   *DestinationReply
+	NoReply    *struct{}
+}
+
+type TopicReply struct {
+}
+
+type DestinationReply struct {
+	Destination duckv1.Destination
+}
+
 type Auth struct {
 	NetSpec  *bindings.KafkaNetSpec
 	AuthSpec *eventingv1alpha1.Auth
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
index a074c6464..f8dd77591 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_validation.go
@@ -22,17 +22,31 @@ import (
 	"knative.dev/pkg/apis"
 )
 
+func (c *Consumer) Validate(ctx context.Context) *apis.FieldError {
+	specCtx := ctx
+	if apis.IsInUpdate(ctx) {
+		specCtx = apis.WithinUpdate(ctx, apis.GetBaseline(ctx).(*Consumer).Spec)
+	}
+	return c.Spec.Validate(specCtx).ViaField("spec")
+}
+
 func (cs *ConsumerSpec) Validate(ctx context.Context) *apis.FieldError {
+	if cs == nil {
+		return nil
+	}
 	var err *apis.FieldError
 	if len(cs.Topics) == 0 {
 		return apis.ErrMissingField("topics")
 	}
-	return err.Also(
+	err = err.Also(
 		cs.Delivery.Validate(ctx).ViaField("delivery"),
 		cs.Configs.Validate(ctx).ViaField("configs"),
 		cs.Filters.Validate(ctx).ViaField("filters"),
 		cs.Subscriber.Validate(ctx).ViaField("subscriber"),
+		cs.PodBind.Validate(ctx).ViaField("podBind"),
+		cs.CloudEventOverrides.Validate(ctx).ViaField("ceOverrides"),
 	)
+	return err
 }
 
 func (d *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
@@ -43,8 +57,14 @@ func (d *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 }
 
 func (cc *ConsumerConfigs) Validate(ctx context.Context) *apis.FieldError {
-	if v, ok := cc.Configs["group.id"]; !ok || v == "" {
-		return apis.ErrMissingField("group.id")
+	expected := []string{
+		"group.id",
+		"bootstrap.servers",
+	}
+	for _, key := range expected {
+		if v, ok := cc.Configs[key]; !ok || v == "" {
+			return apis.ErrMissingField(key)
+		}
 	}
 	return nil
 }
@@ -52,3 +72,37 @@ func (cc *ConsumerConfigs) Validate(ctx context.Context) *apis.FieldError {
 func (f *Filters) Validate(ctx context.Context) *apis.FieldError {
 	return nil
 }
+
+func (p *PodBind) Validate(ctx context.Context) *apis.FieldError {
+	if p == nil {
+		return apis.ErrMissingField("")
+	}
+	if len(p.PodName) == 0 {
+		return apis.ErrMissingField("podName")
+	}
+	if len(p.PodNamespace) == 0 {
+		return apis.ErrMissingField("podNamespace")
+	}
+	if apis.IsInUpdate(ctx) {
+		return p.CheckImmutableFields(ctx, apis.GetBaseline(ctx).(ConsumerSpec).PodBind)
+	}
+	return nil
+}
+
+func (p PodBind) CheckImmutableFields(ctx context.Context, original *PodBind) *apis.FieldError {
+	if p.PodName != original.PodName || p.PodNamespace != original.PodNamespace {
+		return ErrImmutableField("podBind",
+			"Moving a consumer to a different pod is unsupported, to move a consumer to another pod, "+
+				"remove this one and create a new consumer with the same spec",
+		)
+	}
+	return nil
+}
+
+func ErrImmutableField(field, details string) *apis.FieldError {
+	return &apis.FieldError{
+		Message: "Immutable field updated",
+		Paths:   []string{field},
+		Details: details,
+	}
+}
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
index a4f823522..b9e527115 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/zz_generated.deepcopy.go
@@ -279,6 +279,11 @@ func (in *ConsumerSpec) DeepCopyInto(out *ConsumerSpec) {
 		*out = new(DeliverySpec)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Reply != nil {
+		in, out := &in.Reply, &out.Reply
+		*out = new(ReplyStrategy)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Filters != nil {
 		in, out := &in.Filters, &out.Filters
 		*out = new(Filters)
@@ -375,6 +380,23 @@ func (in *DeliverySpec) DeepCopy() *DeliverySpec {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *DestinationReply) DeepCopyInto(out *DestinationReply) {
+	*out = *in
+	in.Destination.DeepCopyInto(&out.Destination)
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DestinationReply.
+func (in *DestinationReply) DeepCopy() *DestinationReply {
+	if in == nil {
+		return nil
+	}
+	out := new(DestinationReply)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Filters) DeepCopyInto(out *Filters) {
 	*out = *in
@@ -418,3 +440,50 @@ func (in *PodBind) DeepCopy() *PodBind {
 	in.DeepCopyInto(out)
 	return out
 }
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ReplyStrategy) DeepCopyInto(out *ReplyStrategy) {
+	*out = *in
+	if in.TopicReply != nil {
+		in, out := &in.TopicReply, &out.TopicReply
+		*out = new(TopicReply)
+		**out = **in
+	}
+	if in.URLReply != nil {
+		in, out := &in.URLReply, &out.URLReply
+		*out = new(DestinationReply)
+		(*in).DeepCopyInto(*out)
+	}
+	if in.NoReply != nil {
+		in, out := &in.NoReply, &out.NoReply
+		*out = new(struct{})
+		**out = **in
+	}
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReplyStrategy.
+func (in *ReplyStrategy) DeepCopy() *ReplyStrategy {
+	if in == nil {
+		return nil
+	}
+	out := new(ReplyStrategy)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *TopicReply) DeepCopyInto(out *TopicReply) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TopicReply.
+func (in *TopicReply) DeepCopy() *TopicReply {
+	if in == nil {
+		return nil
+	}
+	out := new(TopicReply)
+	in.DeepCopyInto(out)
+	return out
+}
diff --git a/control-plane/pkg/contract/extensions.go b/control-plane/pkg/contract/extensions.go
new file mode 100644
index 000000000..e5b1e1a29
--- /dev/null
+++ b/control-plane/pkg/contract/extensions.go
@@ -0,0 +1,22 @@
+/*
+ * Copyright 2021 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 contract
+
+// IncrementGeneration increments Generation.
+func (x *Contract) IncrementGeneration() {
+	x.Generation++
+}
diff --git a/control-plane/pkg/contract/serde.go b/control-plane/pkg/contract/serde.go
new file mode 100644
index 000000000..2c244720c
--- /dev/null
+++ b/control-plane/pkg/contract/serde.go
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2020 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 contract
+
+import (
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+)
+
+type format string
+
+const (
+	Protobuf format = "protobuf"
+	Json     format = "json"
+)
+
+type Serializer interface {
+	Serialize(ct *Contract) ([]byte, error)
+}
+
+type Deserializer interface {
+	Deserialize([]byte) (*Contract, error)
+}
+
+type FormatSerDe struct {
+	Format format
+}
+
+func (s *FormatSerDe) Serialize(ct *Contract) ([]byte, error) {
+	switch s.Format {
+	case Protobuf:
+		return proto.Marshal(ct)
+	default:
+		return protojson.Marshal(ct)
+	}
+}
+
+func (s *FormatSerDe) Deserialize(data []byte) (*Contract, error) {
+	ct := &Contract{}
+	switch s.Format {
+	case Protobuf:
+		err := proto.Unmarshal(data, ct)
+		return ct, err
+	default:
+		err := protojson.Unmarshal(data, ct)
+		return ct, err
+	}
+}
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index fea7d0dbd..5b4b8f154 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -216,7 +216,7 @@ func (r *Reconciler) UpdateDispatcherPodsAnnotation(ctx context.Context, logger
 	if errors != nil {
 		return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
 	}
-	return r.updatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
+	return r.UpdatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
 }
 
 func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
@@ -224,10 +224,10 @@ func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *z
 	if errors != nil {
 		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
 	}
-	return r.updatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
+	return r.UpdatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
 }
 
-func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logger, component string, volumeGeneration uint64, pods []*corev1.Pod) error {
+func (r *Reconciler) UpdatePodsAnnotation(ctx context.Context, logger *zap.Logger, component string, volumeGeneration uint64, pods []*corev1.Pod) error {
 
 	var errors error
 
diff --git a/control-plane/pkg/reconciler/consumer/consumer.go b/control-plane/pkg/reconciler/consumer/consumer.go
new file mode 100644
index 000000000..40fef589c
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/consumer.go
@@ -0,0 +1,412 @@
+/*
+ * Copyright 2021 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 consumer
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	"knative.dev/pkg/apis"
+	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+	"knative.dev/pkg/tracker"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing"
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/injection/reconciler/eventing/v1alpha1/consumer"
+	kafkainternalslisters "knative.dev/eventing-kafka-broker/control-plane/pkg/client/internals/kafka/listers/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
+)
+
+type Reconciler struct {
+	SerDe contract.FormatSerDe
+
+	Enqueue func(c *kafkainternals.Consumer)
+
+	Resolver            *resolver.URIResolver
+	Tracker             tracker.Interface
+	ConsumerGroupLister kafkainternalslisters.ConsumerGroupLister
+	SecretLister        corelisters.SecretLister
+	PodLister           corelisters.PodLister
+	KubeClient          kubernetes.Interface
+}
+
+var (
+	_ consumer.Interface = Reconciler{}
+	_ consumer.Finalizer = Reconciler{}
+)
+
+func (r Reconciler) ReconcileKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
+	logger := logging.FromContext(ctx).Desugar()
+
+	resourceCt, err := r.reconcileContractResource(ctx, c)
+	if err != nil {
+		return c.MarkReconcileContractFailed(err)
+	}
+	c.MarkReconcileContractSucceeded()
+
+	if resourceCt == nil {
+		return nil // Resource will get queued once we have all resources to build the contract.
+	}
+
+	bound, err := r.schedule(ctx, logger, c, addResource(resourceCt))
+	if err != nil {
+		return c.MarkBindFailed(err)
+	}
+	if !bound {
+		// Resource will get queued once we have all resources to schedule the Consumer.
+		c.MarkBindInProgress()
+		return nil
+	}
+	c.MarkBindSucceeded()
+
+	return nil
+}
+
+func (r Reconciler) FinalizeKind(ctx context.Context, c *kafkainternals.Consumer) reconciler.Event {
+
+	logger := logging.FromContext(ctx).Desugar()
+
+	if _, err := r.schedule(ctx, logger, c, removeResource); err != nil {
+		return c.MarkBindFailed(err)
+	}
+
+	return nil
+}
+
+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)
+	}
+
+	userFacingResourceRef, err := r.reconcileUserFacingResourceRef(c)
+	if err != nil {
+		return nil, fmt.Errorf("failed to reconcile user facing resource reference: %w", err)
+	}
+	if userFacingResourceRef == nil {
+		// We don't have yet the user-facing resource in the lister cache.
+		return nil, nil
+	}
+
+	resource := &contract.Resource{
+		Uid:                 string(c.UID),
+		Topics:              c.Spec.Topics,
+		BootstrapServers:    c.Spec.Configs.Configs["bootstrap.servers"],
+		Egresses:            []*contract.Egress{egress},
+		Auth:                nil, // Auth will be added by reconcileAuth
+		CloudEventOverrides: reconcileCEOverrides(c),
+		Reference:           userFacingResourceRef,
+	}
+
+	if err := r.reconcileAuth(ctx, c, resource); err != nil {
+		return nil, fmt.Errorf("failed to reconcile auth: %w", err)
+	}
+
+	return resource, nil
+}
+
+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 subcriber: %w", err)
+	}
+	c.Status.SubscriberURI = destination
+
+	egressConfig := &contract.EgressConfig{}
+	if c.Spec.Delivery != nil {
+		egressConfig, err = coreconfig.EgressConfigFromDelivery(ctx, r.Resolver, c, c.Spec.Delivery.DeliverySpec, 200)
+		if err != nil {
+			return nil, err
+		}
+	}
+	c.Status.DeliveryStatus.DeadLetterSinkURI, _ = apis.ParseURL(egressConfig.DeadLetter)
+
+	egress := &contract.Egress{
+		ConsumerGroup: c.Spec.Configs.Configs["group.id"],
+		Destination:   destination.String(),
+		ReplyStrategy: nil, // Reply will be added by reconcileReplyStrategy
+		Filter:        reconcileFilters(c),
+		Uid:           string(c.UID),
+		EgressConfig:  egressConfig,
+		DeliveryOrder: reconcileDeliveryOrder(c),
+
+		KeyType: 0, // TODO handle key type
+	}
+
+	if err := r.reconcileReplyStrategy(ctx, c, egress); err != nil {
+		return nil, fmt.Errorf("failed to reconcile reply strategy: %w", err)
+	}
+
+	return egress, nil
+}
+
+func (r Reconciler) reconcileAuth(ctx context.Context, c *kafkainternals.Consumer, resource *contract.Resource) error {
+
+	if c.Spec.Auth == nil {
+		return nil
+	}
+	if c.Spec.Auth.NetSpec != nil {
+		authContext, err := security.ResolveAuthContextFromNetSpec(r.SecretLister, c.GetNamespace(), *c.Spec.Auth.NetSpec)
+		if err != nil {
+			return fmt.Errorf("failed to resolve auth context: %w", err)
+		}
+		resource.Auth = &contract.Resource_MultiAuthSecret{MultiAuthSecret: authContext.MultiSecretReference}
+		return nil
+	}
+	if c.Spec.Auth.AuthSpec != nil {
+		secret, err := security.Secret(ctx, &SecretLocator{Consumer: c}, r.SecretProviderFunc())
+		if err != nil {
+			return fmt.Errorf("failed to get secret: %w", err)
+		}
+		resource.Auth = &contract.Resource_AuthSecret{
+			AuthSecret: &contract.Reference{
+				Uuid:      string(secret.UID),
+				Namespace: secret.Namespace,
+				Name:      secret.Name,
+				Version:   secret.ResourceVersion,
+			},
+		}
+		ref := tracker.Reference{
+			APIVersion: c.APIVersion,
+			Kind:       c.Kind,
+			Namespace:  c.GetNamespace(),
+			Name:       c.GetName(),
+		}
+		if err := r.Tracker.TrackReference(ref, c); err != nil {
+			return fmt.Errorf("failed to track reference %v: %w", ref, err)
+		}
+		return nil
+	}
+
+	return nil
+}
+
+func (r Reconciler) SecretProviderFunc() security.SecretProviderFunc {
+	return security.DefaultSecretProviderFunc(r.SecretLister, r.KubeClient)
+}
+
+func reconcileCEOverrides(c *kafkainternals.Consumer) *contract.CloudEventOverrides {
+	if c.Spec.CloudEventOverrides == nil {
+		return nil
+	}
+	return &contract.CloudEventOverrides{Extensions: c.Spec.CloudEventOverrides.Extensions}
+}
+
+func (r Reconciler) reconcileUserFacingResourceRef(c *kafkainternals.Consumer) (*contract.Reference, error) {
+
+	// Get ConsumerGroup from the owner references list.
+	var cOr *metav1.OwnerReference
+	for i, or := range c.OwnerReferences {
+		if strings.ToLower(or.Kind) == kafkainternals.ConsumerGroupGroupVersionKind.Kind {
+			cOr = &c.OwnerReferences[i]
+			break
+		}
+	}
+	if cOr == nil {
+		return nil, fmt.Errorf("failed to find %s in owner references", kafkainternals.ConsumerGroupGroupVersionKind.Kind)
+	}
+
+	cg, err := r.ConsumerGroupLister.ConsumerGroups(c.GetNamespace()).Get(cOr.Name)
+	if err != nil && !apierrors.IsNotFound(err) {
+		return nil, fmt.Errorf("failed to get %s: %w", kafkainternals.ConsumerGroupGroupVersionKind.Kind, err)
+	}
+	if apierrors.IsNotFound(err) {
+		return nil, nil
+	}
+
+	// Get user facing resource from the owner references list.
+	var cgOr *metav1.OwnerReference
+	for i, or := range cg.OwnerReferences {
+		k := strings.ToLower(or.Kind)
+		// TODO hardcoded resource kinds.
+		if k == "trigger" || k == "kafkasource" || k == "kafkachannel" {
+			cgOr = &cg.OwnerReferences[i]
+			break
+		}
+	}
+
+	userFacingResource := cgOr
+	ref := &contract.Reference{
+		Uuid:      string(userFacingResource.UID),
+		Namespace: c.GetNamespace(),
+		Name:      userFacingResource.Name,
+	}
+	return ref, nil
+}
+
+func reconcileDeliveryOrder(c *kafkainternals.Consumer) contract.DeliveryOrder {
+	if c.Spec.Delivery == nil {
+		return contract.DeliveryOrder_UNORDERED
+	}
+	switch c.Spec.Delivery.Ordering {
+	case eventing.Ordered:
+		return contract.DeliveryOrder_ORDERED
+	case eventing.Unordered:
+		return contract.DeliveryOrder_UNORDERED
+	}
+	return contract.DeliveryOrder_UNORDERED
+}
+
+func (r Reconciler) reconcileReplyStrategy(ctx context.Context, c *kafkainternals.Consumer, egress *contract.Egress) error {
+	if c.Spec.Reply == nil {
+		return nil
+	}
+	if c.Spec.Reply.NoReply != nil {
+		egress.ReplyStrategy = &contract.Egress_DiscardReply{}
+		return nil
+	}
+	if c.Spec.Reply.URLReply != nil {
+		destination, err := r.Resolver.URIFromDestinationV1(ctx, c.Spec.Reply.URLReply.Destination, c)
+		if err != nil {
+			return fmt.Errorf("failed to resolve reply destination: %w", err)
+		}
+		egress.ReplyStrategy = &contract.Egress_ReplyUrl{
+			ReplyUrl: destination.String(),
+		}
+		return nil
+	}
+	if c.Spec.Reply.TopicReply != nil {
+		egress.ReplyStrategy = &contract.Egress_ReplyToOriginalTopic{}
+		return nil
+	}
+
+	return nil
+}
+
+func reconcileFilters(c *kafkainternals.Consumer) *contract.Filter {
+	if c.Spec.Filters == nil {
+		return nil
+	}
+	if c.Spec.Filters.Filter != nil {
+		return &contract.Filter{Attributes: c.Spec.Filters.Filter.Attributes}
+	}
+
+	return nil
+}
+
+type contractMutatorFunc func(logger *zap.Logger, ct *contract.Contract, c *kafkainternals.Consumer) int
+
+func addResource(resource *contract.Resource) contractMutatorFunc {
+	return func(logger *zap.Logger, ct *contract.Contract, c *kafkainternals.Consumer) int {
+		return coreconfig.AddOrUpdateResourceConfig(ct, resource, coreconfig.FindResource(ct, c.GetUID()), logger)
+	}
+}
+
+func removeResource(_ *zap.Logger, ct *contract.Contract, c *kafkainternals.Consumer) int {
+	idx := coreconfig.FindResource(ct, c.GetUID())
+	if idx == coreconfig.NoResource {
+		return coreconfig.NoResource
+	}
+	coreconfig.DeleteResource(ct, idx)
+	return idx
+}
+
+// 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) {
+	// 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 err != nil {
+		return false, fmt.Errorf("failed to get pod %s/%s: %w", c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName, err)
+	}
+	if apierrors.IsNotFound(err) {
+		// Pod not found, return no error since the Consumer
+		// will get re-queued when the pod is added.
+		return false, nil
+	}
+	if err != nil {
+		return false, err
+	}
+
+	// Check if the pod is running.
+	if p.Status.Phase != corev1.PodRunning {
+		return false, nil
+	}
+
+	// Get contract associated with the pod.
+	cmName, err := cmNameFromPod(p, c)
+	if err != nil {
+		return false, err
+	}
+
+	b := r.commonReconciler(p, cmName)
+
+	cm, err := b.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return false, fmt.Errorf("failed to get or create data plane ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
+	}
+
+	ct, err := b.GetDataPlaneConfigMapData(logger, cm)
+	if err != nil {
+		return false, fmt.Errorf("failed to get contract from ConfigMap %s/%s: %w", p.GetNamespace(), cmName, err)
+	}
+
+	if changed := mutatorFunc(logger, ct, c); changed == coreconfig.ResourceChanged {
+		logger.Debug("Contract changed", zap.Int("changed", changed))
+
+		ct.IncrementGeneration()
+
+		if err := b.UpdateDataPlaneConfigMap(ctx, ct, cm); err != nil {
+			return false, err
+		}
+	}
+
+	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 {
+	return base.Reconciler{
+		KubeClient:                  r.KubeClient,
+		PodLister:                   r.PodLister,
+		SecretLister:                r.SecretLister,
+		SecretTracker:               r.Tracker,
+		ConfigMapTracker:            r.Tracker,
+		DataPlaneConfigMapNamespace: p.GetNamespace(),
+		DataPlaneConfigMapName:      cmName,
+		DataPlaneConfigFormat:       string(r.SerDe.Format),
+		SystemNamespace:             p.GetNamespace(),
+		DispatcherLabel:             "",
+		ReceiverLabel:               "",
+	}
+}
+
+func cmNameFromPod(p *corev1.Pod, c *kafkainternals.Consumer) (string, error) {
+	var vDp *corev1.Volume
+	for i, v := range p.Spec.Volumes {
+		if v.Name == eventing.ConfigMapVolumeName && v.ConfigMap != nil && v.ConfigMap.Name != "" {
+			vDp = &p.Spec.Volumes[i]
+			break
+		}
+	}
+	if vDp == nil {
+		return "", fmt.Errorf("failed to get data plane volume %s in pod %s/%s", eventing.ConfigMapVolumeName, c.Spec.PodBind.PodNamespace, c.Spec.PodBind.PodName)
+	}
+	return vDp.ConfigMap.Name, nil
+}
diff --git a/control-plane/pkg/reconciler/consumer/secret_locator.go b/control-plane/pkg/reconciler/consumer/secret_locator.go
new file mode 100644
index 000000000..b83a3873a
--- /dev/null
+++ b/control-plane/pkg/reconciler/consumer/secret_locator.go
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2021 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 consumer
+
+import (
+	kafkainternals "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1"
+)
+
+type SecretLocator struct {
+	*kafkainternals.Consumer
+}
+
+func (s *SecretLocator) SecretName() (string, bool) {
+	if s.hasNoAuth() {
+		return "", false
+	}
+	return s.Consumer.Spec.Auth.AuthSpec.Secret.Ref.Name, true
+}
+
+func (s *SecretLocator) hasNoAuth() bool {
+	return s.Consumer.Spec.Auth == nil ||
+		s.Consumer.Spec.Auth.AuthSpec == nil ||
+		s.Consumer.Spec.Auth.AuthSpec.Secret == nil ||
+		s.Consumer.Spec.Auth.AuthSpec.Secret.Ref == nil ||
+		s.Consumer.Spec.Auth.AuthSpec.Secret.Ref.Name == ""
+}
+
+func (s *SecretLocator) SecretNamespace() (string, bool) {
+	return s.GetNamespace(), !s.hasNoAuth()
+}
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 cb89d7d0c..bef025d53 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
@@ -16,8 +16,14 @@
 package dev.knative.eventing.kafka.broker.core.metrics;
 
 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.MeterRegistry;
+import io.micrometer.core.instrument.binder.BaseUnits;
 import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
+import io.micrometer.core.instrument.config.NamingConvention;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.core.metrics.MetricsOptions;
 import io.vertx.core.net.PemKeyCertOptions;
@@ -50,19 +56,38 @@ public class Metrics {
   // are disallowed by the monitoring system.
 
   /**
-   * In prometheus format --> http_requests_malformed_total
+   * In prometheus format --> http_events_sent_total
    */
-  public static final String HTTP_REQUESTS_MALFORMED_COUNT = "http.requests.malformed";
+  public static final String HTTP_EVENTS_SENT_COUNT = "http.events.sent";
 
   /**
-   * In prometheus format --> http_requests_produce_total
+   * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
+   * @see Metrics#eventCount(io.micrometer.core.instrument.Tags)
    */
-  public static final String HTTP_REQUESTS_PRODUCE_COUNT = "http.requests.produce";
+  public static final String EVENTS_COUNT = "event_count";
 
   /**
-   * In prometheus format --> http_events_sent_total
+   * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
+   * @see Metrics#eventDispatchLatency(io.micrometer.core.instrument.Tags)
    */
-  public static final String HTTP_EVENTS_SENT_COUNT = "http.events.sent";
+  public static final String EVENT_DISPATCH_LATENCY = "event_dispatch_latencies";
+
+  /**
+   * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
+   */
+  public static class Tags {
+    public static final String RESPONSE_CODE = "response_code";
+    public static final String RESPONSE_CODE_CLASS = "response_code_class";
+    public static final String EVENT_TYPE = "event_type";
+  }
+
+  /**
+   * @link https://knative.dev/docs/eventing/observability/metrics/eventing-metrics/
+   */
+  public static class Units {
+    // Unified Code for Units of Measure: http://unitsofmeasure.org/ucum.html
+    public static final String DIMENSIONLESS = "1";
+  }
 
   /**
    * Get metrics options from the given metrics configurations.
@@ -71,6 +96,9 @@ public class Metrics {
    * @return Metrics options.
    */
   public static MetricsOptions getOptions(final BaseEnv metricsConfigs) {
+    final var registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+    registry.config().namingConvention(NamingConvention.identity);
+
     return new MicrometerMetricsOptions()
       .setEnabled(true)
       .addDisabledMetricsCategory(MetricsDomain.EVENT_BUS)
@@ -81,6 +109,7 @@ public class Metrics {
       .setMetricsNaming(MetricsNaming.v4Names())
       .setRegistryName(METRICS_REGISTRY_NAME)
       .setJvmMetricsEnabled(metricsConfigs.isMetricsJvmEnabled())
+      .setMicrometerRegistry(registry)
       .setPrometheusOptions(new VertxPrometheusOptions()
         .setEmbeddedServerOptions(new HttpServerOptions()
           .setPort(metricsConfigs.getMetricsPort())
@@ -149,4 +178,21 @@ public class Metrics {
     }
     return host;
   }
+
+  public static Counter.Builder eventCount(final io.micrometer.core.instrument.Tags tags) {
+    return Counter
+      .builder(EVENTS_COUNT)
+      .description("Number of events received")
+      .tags(tags)
+      .baseUnit(Metrics.Units.DIMENSIONLESS);
+  }
+
+  public static DistributionSummary.Builder eventDispatchLatency(final io.micrometer.core.instrument.Tags tags) {
+    return DistributionSummary
+      .builder(EVENT_DISPATCH_LATENCY)
+      .description("The time spent dispatching an event to Kafka")
+      .tags(tags)
+      .baseUnit(BaseUnits.MILLISECONDS)
+      .serviceLevelObjectives(1, 2, 5, 10, 20, 50, 100, 500, 1000, 5000, 10000);
+  }
 }
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 8fb9aa685..d80a8446a 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -31,7 +31,7 @@
     <java.version>15</java.version>
 
     <!-- Maven plugins -->
-    <maven.compiler.plugin.version>3.8.1</maven.compiler.plugin.version>
+    <maven.compiler.plugin.version>3.9.0</maven.compiler.plugin.version>
     <maven.surefire.plugin.version>2.22.2</maven.surefire.plugin.version>
     <maven.shade.plugin.version>3.2.4</maven.shade.plugin.version>
     <maven.editorconfig.plugin.version>0.1.1</maven.editorconfig.plugin.version>
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressProducer.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressProducer.java
index b42131983..5f0b09d81 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressProducer.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/IngressProducer.java
@@ -15,6 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
+import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import io.cloudevents.CloudEvent;
 import io.vertx.core.Future;
 import io.vertx.kafka.client.producer.KafkaProducer;
@@ -45,4 +46,9 @@ public interface IngressProducer {
    */
   String getTopic();
 
+  /**
+   * @return the resource associated with this producer.
+   */
+  DataPlaneContract.Reference getReference();
+
 }
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 7c7edaef1..9fab7b06c 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,13 +15,11 @@
  */
 package dev.knative.eventing.kafka.broker.receiver;
 
-import io.vertx.core.http.HttpServerRequest;
-
 /**
  * This class handles incoming ingress requests.
  */
 public interface IngressRequestHandler {
 
-  void handle(HttpServerRequest request, IngressProducer producer);
+  void handle(RequestContext request, IngressProducer producer);
 
 }
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestContext.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestContext.java
new file mode 100644
index 000000000..16fe784fc
--- /dev/null
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/RequestContext.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.receiver;
+
+import io.vertx.core.http.HttpServerRequest;
+
+public class RequestContext {
+
+  private final HttpServerRequest request;
+  private final long receivedAtMs;
+
+  public RequestContext(final HttpServerRequest request) {
+    this.request = request;
+    this.receivedAtMs = System.currentTimeMillis();
+  }
+
+  public long performLatency() {
+    return System.currentTimeMillis() - receivedAtMs;
+  }
+
+  public HttpServerRequest getRequest() {
+    return request;
+  }
+}
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
index 98693b048..0f7983acf 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java
@@ -120,7 +120,7 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
 
       final var ingressInfo = new IngressProducerImpl(
         rc.getValue().getProducer(),
-        resource.getTopics(0),
+        resource,
         ingress.getPath(),
         producerProps
       );
@@ -223,11 +223,15 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     private final String topic;
     private final String path;
     private final Properties producerProperties;
+    private final DataPlaneContract.Reference reference;
 
-    IngressProducerImpl(final KafkaProducer<String, CloudEvent> producer, final String topic, final String path,
+    IngressProducerImpl(final KafkaProducer<String, CloudEvent> producer,
+                        final DataPlaneContract.Resource resource,
+                        final String path,
                         final Properties producerProperties) {
       this.producer = producer;
-      this.topic = topic;
+      this.topic = resource.getTopics(0);
+      this.reference = resource.getReference();
       this.path = path;
       this.producerProperties = producerProperties;
     }
@@ -242,6 +246,11 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
       return topic;
     }
 
+    @Override
+    public DataPlaneContract.Reference getReference() {
+      return reference;
+    }
+
     String getPath() {
       return path;
     }
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 fa4c84b82..2e4410646 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
@@ -18,6 +18,7 @@ package dev.knative.eventing.kafka.broker.receiver.impl;
 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;
+import dev.knative.eventing.kafka.broker.receiver.RequestContext;
 import dev.knative.eventing.kafka.broker.receiver.impl.handler.MethodNotAllowedHandler;
 import dev.knative.eventing.kafka.broker.receiver.impl.handler.ProbeHandler;
 import dev.knative.eventing.kafka.broker.receiver.main.ReceiverEnv;
@@ -117,6 +118,8 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
   @Override
   public void handle(HttpServerRequest request) {
 
+    final var requestContext = new RequestContext(request);
+
     // Look up for the ingress producer
     IngressProducer producer = this.ingressProducerStore.resolve(request.path());
     if (producer == null) {
@@ -133,6 +136,6 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     }
 
     // Invoke the ingress request handler
-    this.ingressRequestHandler.handle(request, producer);
+    this.ingressRequestHandler.handle(requestContext, producer);
   }
 }
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 d87474682..c4d35b474 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,17 +15,20 @@
  */
 package dev.knative.eventing.kafka.broker.receiver.impl.handler;
 
+import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 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;
 import dev.knative.eventing.kafka.broker.receiver.IngressRequestHandler;
+import dev.knative.eventing.kafka.broker.receiver.RequestContext;
 import dev.knative.eventing.kafka.broker.receiver.RequestToRecordMapper;
 import io.cloudevents.CloudEvent;
-import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Tag;
+import io.micrometer.core.instrument.Tags;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.context.Context;
 import io.vertx.core.Future;
-import io.vertx.core.http.HttpServerRequest;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 import io.vertx.kafka.client.producer.RecordMetadata;
 import org.slf4j.Logger;
@@ -43,36 +46,58 @@ import static io.netty.handler.codec.http.HttpResponseStatus.SERVICE_UNAVAILABLE
  */
 public class IngressRequestHandlerImpl implements IngressRequestHandler {
 
+  static final Tag UNKNOWN_EVENT_TYPE_TAG = Tag.of(Metrics.Tags.EVENT_TYPE, "unknown");
+
   static final int MAPPER_FAILED = BAD_REQUEST.code();
+  static final Tags MAPPER_FAILED_COMMON_TAGS = Tags.of(
+    Tag.of(Metrics.Tags.RESPONSE_CODE_CLASS, "4xx"),
+    Tag.of(Metrics.Tags.RESPONSE_CODE, Integer.toString(MAPPER_FAILED)),
+    UNKNOWN_EVENT_TYPE_TAG
+  );
+
   static final int RECORD_PRODUCED = ACCEPTED.code();
+  static final Tags RECORD_PRODUCED_COMMON_TAGS = Tags.of(
+    Tag.of(Metrics.Tags.RESPONSE_CODE, Integer.toString(RECORD_PRODUCED)),
+    Tag.of(Metrics.Tags.RESPONSE_CODE_CLASS, "2xx")
+  );
+
   static final int FAILED_TO_PRODUCE = SERVICE_UNAVAILABLE.code();
+  static final Tags FAILED_TO_PRODUCE_COMMON_TAGS = Tags.of(
+    Tag.of(Metrics.Tags.RESPONSE_CODE, Integer.toString(FAILED_TO_PRODUCE)),
+    Tag.of(Metrics.Tags.RESPONSE_CODE_CLASS, "5xx")
+  );
 
   private static final Logger logger = LoggerFactory.getLogger(IngressRequestHandlerImpl.class);
 
   private final RequestToRecordMapper requestToRecordMapper;
-  private final Counter badRequestCounter;
-  private final Counter produceEventsCounter;
+  private final MeterRegistry meterRegistry;
 
-  public IngressRequestHandlerImpl(
-    RequestToRecordMapper requestToRecordMapper,
-    Counter badRequestCounter,
-    Counter produceEventsCounter) {
+  public IngressRequestHandlerImpl(final RequestToRecordMapper requestToRecordMapper,
+                                   final MeterRegistry meterRegistry) {
     this.requestToRecordMapper = requestToRecordMapper;
-    this.badRequestCounter = badRequestCounter;
-    this.produceEventsCounter = produceEventsCounter;
+    this.meterRegistry = meterRegistry;
   }
 
   @Override
-  public void handle(HttpServerRequest request, IngressProducer producer) {
+  public void handle(final RequestContext requestContext, final IngressProducer producer) {
+
+    final var resourceTags = Tags.of(
+      Tag.of("name", producer.getReference().getName()),
+      Tag.of("namespace_name", producer.getReference().getNamespace())
+    );
+
     requestToRecordMapper
-      .requestToRecord(request, producer.getTopic())
+      .requestToRecord(requestContext.getRequest(), producer.getTopic())
       .onFailure(cause -> {
         // Conversion to record failed
-        request.response().setStatusCode(MAPPER_FAILED).end();
-        badRequestCounter.increment();
+        requestContext.getRequest().response().setStatusCode(MAPPER_FAILED).end();
+
+        final var tags = MAPPER_FAILED_COMMON_TAGS.and(resourceTags);
+        Metrics.eventDispatchLatency(tags).register(meterRegistry).record(requestContext.performLatency());
+        Metrics.eventCount(tags).register(meterRegistry).increment();
 
         logger.warn("Failed to convert request to record {}",
-          keyValue("path", request.path()),
+          keyValue("path", requestContext.getRequest().path()),
           cause
         );
       })
@@ -93,28 +118,41 @@ public class IngressRequestHandlerImpl implements IngressRequestHandler {
         // Decorate the span with event specific attributed
         TracingSpan.decorateCurrentWithEvent(record.value());
 
-        // Publish the record
-        return publishRecord(producer, record).onComplete(ar -> {
-          // Write the response back
-          if (ar.succeeded()) {
-            request.response()
-              .setStatusCode(RECORD_PRODUCED)
-              .end();
-          } else {
-            request.response()
-              .setStatusCode(FAILED_TO_PRODUCE)
-              .end();
-          }
-        });
+        final var eventTypeTag = Tag.of(Metrics.Tags.EVENT_TYPE, record.value().getType());
+
+        return publishRecord(producer, record)
+          .onSuccess(m -> {
+            requestContext.getRequest().response().setStatusCode(RECORD_PRODUCED).end();
+
+            final var tags = RECORD_PRODUCED_COMMON_TAGS
+              .and(resourceTags)
+              .and(eventTypeTag);
+            Metrics.eventDispatchLatency(tags).register(meterRegistry).record(requestContext.performLatency());
+            Metrics.eventCount(tags).register(meterRegistry).increment();
+
+          })
+          .onFailure(cause -> {
+            requestContext.getRequest().response().setStatusCode(FAILED_TO_PRODUCE).end();
+
+            final var tags = FAILED_TO_PRODUCE_COMMON_TAGS
+              .and(resourceTags)
+              .and(eventTypeTag);
+            Metrics.eventDispatchLatency(tags).register(meterRegistry).record(requestContext.performLatency());
+            Metrics.eventCount(tags).register(meterRegistry).increment();
+
+            logger.warn("Failed to produce record {}",
+              keyValue("path", requestContext.getRequest().path()),
+              cause
+            );
+          });
       });
   }
 
-  private Future<RecordMetadata> publishRecord(IngressProducer ingress,
-                                               KafkaProducerRecord<String, CloudEvent> record) {
+  private static Future<RecordMetadata> publishRecord(final IngressProducer ingress,
+                                                      final KafkaProducerRecord<String, CloudEvent> record) {
     return ingress.send(record)
       .onComplete(ar -> {
         if (ar.succeeded()) {
-          produceEventsCounter.increment();
           if (logger.isDebugEnabled()) {
             logger.debug("Record produced {} {} {} {} {}",
               keyValue("topic", record.topic()),
diff --git a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java
index 4e5f5adc5..f8fcfcbf7 100644
--- a/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java
+++ b/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/main/ReceiverVerticleFactory.java
@@ -36,7 +36,7 @@ import java.util.function.Supplier;
 
 class ReceiverVerticleFactory implements Supplier<Verticle> {
 
-  private ReceiverEnv env;
+  private final ReceiverEnv env;
   private final Properties producerConfigs;
   private final HttpServerOptions httpServerOptions;
 
@@ -51,8 +51,7 @@ class ReceiverVerticleFactory implements Supplier<Verticle> {
     this.httpServerOptions = httpServerOptions;
     this.ingressRequestHandler = new IngressRequestHandlerImpl(
       StrictRequestToRecordMapper.getInstance(),
-      metricsRegistry.counter(Metrics.HTTP_REQUESTS_MALFORMED_COUNT),
-      metricsRegistry.counter(Metrics.HTTP_REQUESTS_PRODUCE_COUNT)
+      metricsRegistry
     );
   }
 


Test patch:
diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
index d42e2b32f..3e00b64c7 100644
--- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
+++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go
@@ -95,7 +95,8 @@ func TestConsumerGroup_Validate(t *testing.T) {
 							Topics: []string{"t1"},
 							Configs: ConsumerConfigs{
 								Configs: map[string]string{
-									"group.id": "g1",
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
 								},
 							},
 							Delivery: &DeliverySpec{
@@ -107,12 +108,276 @@ func TestConsumerGroup_Validate(t *testing.T) {
 									Host:   "127.0.0.1",
 								},
 							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "invalid no pod name",
+			ctx:  context.Background(),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid no pod namespace",
+			ctx:  context.Background(),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName: "p-0",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "valid - no updates",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
 						},
 					},
 				},
 			},
 			wantErr: false,
 		},
+		{
+			name: "invalid pod name update",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-1",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
+		{
+			name: "invalid pod name update",
+			ctx: apis.WithinUpdate(context.Background(), &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns-1",
+							},
+						},
+					},
+				},
+			}),
+			given: &ConsumerGroup{
+				Spec: ConsumerGroupSpec{
+					Replicas: pointer.Int32Ptr(1),
+					Selector: map[string]string{"app": "app"},
+					Template: ConsumerTemplateSpec{
+						Spec: ConsumerSpec{
+							Topics: []string{"t1"},
+							Configs: ConsumerConfigs{
+								Configs: map[string]string{
+									"group.id":          "g1",
+									"bootstrap.servers": "kafka:9092",
+								},
+							},
+							Delivery: &DeliverySpec{
+								DeliverySpec: &eventingduck.DeliverySpec{},
+							},
+							Subscriber: duckv1.Destination{
+								URI: &apis.URL{
+									Scheme: "http",
+									Host:   "127.0.0.1",
+								},
+							},
+							PodBind: &PodBind{
+								PodName:      "p-0",
+								PodNamespace: "ns-2",
+							},
+						},
+					},
+				},
+			},
+			wantErr: true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/control-plane/pkg/contract/extensions_test.go b/control-plane/pkg/contract/extensions_test.go
new file mode 100644
index 000000000..e005f8ca9
--- /dev/null
+++ b/control-plane/pkg/contract/extensions_test.go
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021 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 contract_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+)
+
+func TestContractIncrementGeneration(t *testing.T) {
+	c := &contract.Contract{Generation: 42}
+	c.IncrementGeneration()
+
+	require.Equal(t, uint64(43), c.Generation)
+}
diff --git a/control-plane/pkg/contract/serde_test.go b/control-plane/pkg/contract/serde_test.go
new file mode 100644
index 000000000..7b24ee162
--- /dev/null
+++ b/control-plane/pkg/contract/serde_test.go
@@ -0,0 +1,60 @@
+/*
+ * Copyright 2020 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 contract_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+)
+
+func TestFormatSerDe(t *testing.T) {
+
+	tt := []struct {
+		serde contract.FormatSerDe
+	}{
+		{
+			serde: contract.FormatSerDe{Format: contract.Protobuf},
+		},
+		{
+			serde: contract.FormatSerDe{Format: contract.Json},
+		},
+	}
+
+	for _, tc := range tt {
+		t.Run(string(tc.serde.Format), func(t *testing.T) {
+			ct := &contract.Contract{
+				Generation: 42,
+				Resources: []*contract.Resource{
+					{Uid: "abc"},
+					{Uid: "abd"},
+				},
+			}
+
+			b, err := tc.serde.Serialize(ct)
+			require.Nil(t, err)
+
+			got, err := tc.serde.Deserialize(b)
+			require.Nil(t, err)
+
+			require.True(t, proto.Equal(ct, got), "%v %v", ct, got)
+		})
+	}
+}
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java
index 78d6927ff..b2888b700 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTest.java
@@ -25,15 +25,17 @@ import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
 import dev.knative.eventing.kafka.broker.core.reconciler.impl.ResourcesReconcilerMessageHandler;
 import dev.knative.eventing.kafka.broker.core.security.AuthProvider;
 import dev.knative.eventing.kafka.broker.core.testing.CloudEventSerializerMock;
-import dev.knative.eventing.kafka.broker.receiver.impl.handler.IngressRequestHandlerImpl;
 import dev.knative.eventing.kafka.broker.receiver.impl.handler.ControlPlaneProbeRequestUtil;
+import dev.knative.eventing.kafka.broker.receiver.impl.handler.IngressRequestHandlerImpl;
 import dev.knative.eventing.kafka.broker.receiver.main.ReceiverEnv;
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.v1.CloudEventBuilder;
 import io.cloudevents.http.vertx.VertxMessageFactory;
 import io.micrometer.core.instrument.Counter;
-import io.micrometer.core.instrument.Meter.Id;
-import io.micrometer.core.instrument.cumulative.CumulativeCounter;
+import io.micrometer.core.instrument.Measurement;
+import io.micrometer.core.instrument.search.MeterNotFoundException;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
@@ -51,6 +53,7 @@ import org.apache.kafka.clients.producer.MockProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
@@ -63,6 +66,8 @@ import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 import static io.netty.handler.codec.http.HttpResponseStatus.ACCEPTED;
 import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
@@ -82,13 +87,15 @@ public class ReceiverVerticleTest {
 
   private static MockProducer<String, CloudEvent> mockProducer;
   private static IngressProducerReconcilableStore store;
-  private static Counter badRequestCount;
-  private static Counter produceRequestCount;
 
   static {
-    BackendRegistries.setupBackend(new MicrometerMetricsOptions().setRegistryName(Metrics.METRICS_REGISTRY_NAME));
+    BackendRegistries.setupBackend(new MicrometerMetricsOptions()
+      .setMicrometerRegistry(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT))
+      .setRegistryName(Metrics.METRICS_REGISTRY_NAME));
   }
 
+  private PrometheusMeterRegistry registry;
+
   @BeforeEach
   public void setUp(final Vertx vertx, final VertxTestContext testContext) {
     ContractMessageCodec.register(vertx.eventBus());
@@ -101,15 +108,14 @@ public class ReceiverVerticleTest {
     );
     KafkaProducer<String, CloudEvent> producer = KafkaProducer.create(vertx, mockProducer);
 
-    badRequestCount = new CumulativeCounter(mock(Id.class));
-    produceRequestCount = new CumulativeCounter(mock(Id.class));
-
     store = new IngressProducerReconcilableStore(
       AuthProvider.noAuth(),
       new Properties(),
       properties -> producer
     );
 
+    registry = new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
+
     final var httpServerOptions = new HttpServerOptions();
     httpServerOptions.setPort(PORT);
     httpServerOptions.setHost("localhost");
@@ -122,8 +128,7 @@ public class ReceiverVerticleTest {
       v -> store,
       new IngressRequestHandlerImpl(
         StrictRequestToRecordMapper.getInstance(),
-        badRequestCount,
-        produceRequestCount
+        registry
       )
     );
     vertx.deployVerticle(verticle, testContext.succeeding(ar -> testContext.completeNow()));
@@ -134,6 +139,11 @@ public class ReceiverVerticleTest {
     mockProducer.clear();
   }
 
+  @AfterEach
+  public void tearDownEach() {
+    registry.close();
+  }
+
   @AfterAll
   public static void tearDown() {
     webClient.close();
@@ -159,8 +169,38 @@ public class ReceiverVerticleTest {
           .as("verify path: " + tc.path)
           .isEqualTo(tc.responseStatusCode);
 
-        assertThat((double) tc.badRequestCount).isEqualTo(badRequestCount.count());
-        assertThat((double) tc.produceEventCount).isEqualTo(produceRequestCount.count());
+        final var expectedCount = (double) tc.badRequestCount + (double) tc.produceEventCount;
+
+        final var defaultCounter = Counter.builder(Metrics.EVENTS_COUNT).register(Metrics.getRegistry());
+        Counter counter = defaultCounter;
+        try {
+          if (expectedCount > 0) {
+            counter = registry.get(Metrics.EVENTS_COUNT).counters()
+              .stream()
+              .reduce((a, b) -> b) // get last element
+              .orElse(defaultCounter);
+          }
+        } catch (MeterNotFoundException ignored1) {
+        }
+
+        assertThat(counter.count())
+          .describedAs("Counter: " + StreamSupport
+            .stream(counter.measure().spliterator(), false)
+            .map(Measurement::toString)
+            .collect(Collectors.joining()))
+          .isEqualTo(expectedCount);
+
+        if (tc.expectedDispatchLatency) {
+          final var eventDispatchLatency = registry.get(Metrics.EVENT_DISPATCH_LATENCY)
+            .summaries()
+            .stream()
+            .reduce((a, b) -> b)
+            .get();
+
+          assertThat(eventDispatchLatency.max()).isGreaterThan(0);
+          assertThat(eventDispatchLatency.totalAmount()).isGreaterThan(0);
+          assertThat(eventDispatchLatency.mean()).isGreaterThan(0);
+        }
 
         checkpoints.flag();
         countDown.countDown();
@@ -384,6 +424,7 @@ public class ReceiverVerticleTest {
     final DataPlaneContract.Resource resource;
     final int badRequestCount;
     final int produceEventCount;
+    final boolean expectedDispatchLatency;
 
     TestCase(
       final ProducerRecord<String, CloudEvent> record,
@@ -407,6 +448,8 @@ public class ReceiverVerticleTest {
         produceEventsCount = 1;
       }
 
+      this.expectedDispatchLatency = responseStatusCode == ACCEPTED.code();
+
       this.badRequestCount = badRequestCount;
       this.produceEventCount = produceEventsCount;
     }
@@ -416,5 +459,13 @@ public class ReceiverVerticleTest {
         .addResources(this.resource)
         .build();
     }
+
+    @Override
+    public String toString() {
+      return "TestCase{" +
+        "path='" + path + '\'' +
+        ", responseStatusCode=" + responseStatusCode +
+        '}';
+    }
   }
 }
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTracingTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTracingTest.java
index 2f5a2453f..58f9090d3 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTracingTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/ReceiverVerticleTracingTest.java
@@ -124,8 +124,7 @@ public class ReceiverVerticleTracingTest {
       v -> store,
       new IngressRequestHandlerImpl(
         StrictRequestToRecordMapper.getInstance(),
-        new CumulativeCounter(mock(Id.class)),
-        new CumulativeCounter(mock(Id.class))
+        Metrics.getRegistry()
       )
     );
 
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImplTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImplTest.java
index 5e85659fa..4df86231f 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImplTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/handler/IngressRequestHandlerImplTest.java
@@ -15,11 +15,15 @@
  */
 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.testing.CoreObjects;
 import dev.knative.eventing.kafka.broker.receiver.IngressProducer;
+import dev.knative.eventing.kafka.broker.receiver.RequestContext;
 import dev.knative.eventing.kafka.broker.receiver.RequestToRecordMapper;
 import io.cloudevents.CloudEvent;
-import io.micrometer.core.instrument.Counter;
+import io.micrometer.prometheus.PrometheusConfig;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.vertx.core.Future;
 import io.vertx.core.http.HttpMethod;
 import io.vertx.core.http.HttpServerRequest;
@@ -28,6 +32,8 @@ import io.vertx.core.http.impl.headers.HeadersMultiMap;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.kafka.client.producer.RecordMetadata;
 import io.vertx.kafka.client.producer.impl.KafkaProducerRecordImpl;
+import io.vertx.micrometer.MicrometerMetricsOptions;
+import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.kafka.clients.producer.MockProducer;
 import org.junit.jupiter.api.Test;
 
@@ -39,6 +45,12 @@ import static org.mockito.Mockito.when;
 
 public class IngressRequestHandlerImplTest {
 
+  static {
+    BackendRegistries.setupBackend(new MicrometerMetricsOptions()
+      .setMicrometerRegistry(new PrometheusMeterRegistry(PrometheusConfig.DEFAULT))
+      .setRegistryName(Metrics.METRICS_REGISTRY_NAME));
+  }
+
   @Test
   public void shouldSendRecordAndTerminateRequestWithRecordProduced() {
     shouldSendRecord(false, IngressRequestHandlerImpl.RECORD_PRODUCED);
@@ -72,11 +84,10 @@ public class IngressRequestHandlerImplTest {
 
     final var handler = new IngressRequestHandlerImpl(
       mapper,
-      mock(Counter.class),
-      mock(Counter.class)
+      Metrics.getRegistry()
     );
 
-    handler.handle(request, new IngressProducer() {
+    handler.handle(new RequestContext(request), new IngressProducer() {
       @Override
       public KafkaProducer<String, CloudEvent> getKafkaProducer() {
         return producer;
@@ -86,6 +97,11 @@ public class IngressRequestHandlerImplTest {
       public String getTopic() {
         return "1-12345";
       }
+
+      @Override
+      public DataPlaneContract.Reference getReference() {
+        return DataPlaneContract.Reference.newBuilder().build();
+      }
     });
 
     verifySetStatusCodeAndTerminateResponse(statusCode, response);
@@ -103,11 +119,10 @@ public class IngressRequestHandlerImplTest {
 
     final var handler = new IngressRequestHandlerImpl(
       mapper,
-      mock(Counter.class),
-      mock(Counter.class)
+      Metrics.getRegistry()
     );
 
-    handler.handle(request, new IngressProducer() {
+    handler.handle(new RequestContext(request), new IngressProducer() {
       @Override
       public KafkaProducer<String, CloudEvent> getKafkaProducer() {
         return producer;
@@ -117,6 +132,11 @@ public class IngressRequestHandlerImplTest {
       public String getTopic() {
         return "1-12345";
       }
+
+      @Override
+      public DataPlaneContract.Reference getReference() {
+        return DataPlaneContract.Reference.newBuilder().build();
+      }
     });
 
     verifySetStatusCodeAndTerminateResponse(IngressRequestHandlerImpl.MAPPER_FAILED, response);
diff --git a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
index 5ef6b3673..095ff8d32 100644
--- a/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
+++ b/data-plane/tests/src/test/java/dev/knative/eventing/kafka/broker/tests/DataPlaneTest.java
@@ -36,7 +36,6 @@ import io.cloudevents.core.v1.CloudEventV1;
 import io.cloudevents.http.vertx.VertxMessageFactory;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.debezium.kafka.KafkaCluster;
-import io.micrometer.core.instrument.Counter;
 import io.vertx.core.Vertx;
 import io.vertx.core.http.HttpServerOptions;
 import io.vertx.ext.web.client.WebClient;
@@ -385,8 +384,7 @@ public class DataPlaneTest {
       ),
       new IngressRequestHandlerImpl(
         StrictRequestToRecordMapper.getInstance(),
-        mock(Counter.class),
-        mock(Counter.class)
+        Metrics.getRegistry()
       )
     );