Repository: knative-sandbox/eventing-kafka-broker
Pull request #1990 (2022-03-09 21:32:49)

Handle host headers in dataplane for identification of channel instances
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Handle host headers in dataplane for identification of channel instances
- Needed to change the proto definition, but it is backwards compatible. See https://github.com/aliok/proto-experiments?
- Logic is like this:
  - If path is defined for the ingress, use that as the identification information and ignore the hostname
  - If hostname is defined for the ingress, use that as the identification information unless there's a path defined

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note
- :gift: Handle host headers in dataplane for identification of channel instances
```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

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

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":[]} -->
aliok: Created instead of https://github.com/knative-sandbox/eventing-kafka-broker/pull/1985
aliok: cc @pierDipi 
aliok: /retest

```
Creating cluster e2e-cls-qoxagyct in us-east1...
{"component":"entrypoint","file":"k8s.io/test-infra/prow/entrypoint/run.go:169","func":"k8s.io/test-infra/prow/entrypoint.Options.ExecuteProcess","level":"error","msg":"Entrypoint received interrupt: terminated","severity":"error","time":"2022-03-09T21:33:26Z"}
```
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#1990](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (2f404f0) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/a4f7a2c4aaa8e74283b34615dbf5d7f3a86f8d3a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a4f7a2c) will **increase** coverage by `0.00%`.
> The diff coverage is `64.51%`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/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/1990?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    #1990   +/-   ##
=========================================
  Coverage     67.06%   67.07%           
- Complexity      633      640    +7     
=========================================
  Files           140      140           
  Lines          6567     6572    +5     
  Branches        179      185    +6     
=========================================
+ Hits           4404     4408    +4     
+ Misses         1794     1789    -5     
- Partials        369      375    +6     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `82.29% <70.83%> (-0.18%)` | :arrow_down: |

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/1990?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/contract/contract.pb.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvY29udHJhY3QucGIuZ28=) | `8.41% <0.00%> (-0.19%)` | :arrow_down: |
| [control-plane/pkg/contract/log\_marshaller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvbG9nX21hcnNoYWxsZXIuZ28=) | `0.00% <0.00%> (ø)` | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | `70.04% <ø> (-0.89%)` | :arrow_down: |
| [...eceiver/impl/IngressProducerReconcilableStore.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL0luZ3Jlc3NQcm9kdWNlclJlY29uY2lsYWJsZVN0b3JlLmphdmE=) | `83.67% <68.18%> (-2.58%)` | :arrow_down: |
| [control-plane/pkg/reconciler/channel/channel.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL2NoYW5uZWwuZ28=) | `71.08% <100.00%> (-0.21%)` | :arrow_down: |
| [...ntrol-plane/pkg/reconciler/channel/v2/channelv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NoYW5uZWx2Mi5nbw==) | `71.77% <100.00%> (-0.18%)` | :arrow_down: |
| [control-plane/pkg/reconciler/sink/kafka\_sink.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2thZmthX3NpbmsuZ28=) | `72.78% <100.00%> (ø)` | |
| [.../core/reconciler/impl/ResourcesReconcilerImpl.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9jb3JlL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2NvcmUvcmVjb25jaWxlci9pbXBsL1Jlc291cmNlc1JlY29uY2lsZXJJbXBsLmphdmE=) | `90.20% <100.00%> (+0.06%)` | :arrow_up: |
| [...g/kafka/broker/receiver/impl/ReceiverVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9yZWNlaXZlci9zcmMvbWFpbi9qYXZhL2Rldi9rbmF0aXZlL2V2ZW50aW5nL2thZmthL2Jyb2tlci9yZWNlaXZlci9pbXBsL1JlY2VpdmVyVmVydGljbGUuamF2YQ==) | `97.72% <100.00%> (ø)` | |
| [...tcher/impl/consumer/UnorderedConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9Vbm9yZGVyZWRDb25zdW1lclZlcnRpY2xlLmphdmE=) | `91.30% <0.00%> (-8.70%)` | :arrow_down: |
| ... and [11 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990?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/1990?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [a4f7a2c...2f404f0](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1990?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).

knative-prow-robot: @aliok: 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
--- | --- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-broker-unit-tests | 2f404f0948e323e36931eab783341aa5d91fb894 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/1990/pull-knative-sandbox-eventing-kafka-broker-unit-tests/1501672878385401856) | true | `/test pull-knative-sandbox-eventing-kafka-broker-unit-tests`

[Full PR test history](https://gubernator.knative.dev/pr/knative-sandbox_eventing-kafka-broker/1990). [Your PR dashboard](https://gubernator.knative.dev/pr/aliok).

<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 -->
devguyio: I am gonna take the liberty of holding this until #1922 goes in, there's a possibility of a conflict here because of contract changes and we've been working for a while to have #1922 in. 

/hold
aliok: /unhold

https://github.com/knative-sandbox/eventing-kafka-broker/pull/1922 is merged

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

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1990#" title="Author self-approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/1990#pullrequestreview-905515297" title="Approved">pierDipi</a>*

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

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

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

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

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

Patch:
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index abdafb042..b22878227 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -1172,6 +1172,14 @@ func (*Egress_ReplyToOriginalTopic) isEgress_ReplyStrategy() {}
 
 func (*Egress_DiscardReply) isEgress_ReplyStrategy() {}
 
+// Ingress is the definition for HTTP ingress that is receiving the events
+// into the Knative Kafka component.
+//
+// path and host fields are used for identifying the targets. They are exclusive.
+// When a request comes with "/some-path", hostname will not be checked.
+// When a request comes with "/", only hostname matching will be done.
+// It is allowed to specify both path and host in ingress contract
+// to support both modes.
 type Ingress struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -1179,13 +1187,10 @@ type Ingress struct {
 
 	// Optional content mode to use when pushing messages to Kafka
 	ContentMode ContentMode `protobuf:"varint,1,opt,name=contentMode,proto3,enum=ContentMode" json:"contentMode,omitempty"`
-	// Ingress can both listen on a specific HTTP path
-	// or listen to the / path but match the Host header
-	//
-	// Types that are assignable to IngressType:
-	//	*Ingress_Path
-	//	*Ingress_Host
-	IngressType isIngress_IngressType `protobuf_oneof:"ingressType"`
+	// path to listen for incoming events.
+	Path string `protobuf:"bytes,2,opt,name=path,proto3" json:"path,omitempty"`
+	// host header to match
+	Host string `protobuf:"bytes,3,opt,name=host,proto3" json:"host,omitempty"`
 }
 
 func (x *Ingress) Reset() {
@@ -1227,45 +1232,20 @@ func (x *Ingress) GetContentMode() ContentMode {
 	return ContentMode_BINARY
 }
 
-func (m *Ingress) GetIngressType() isIngress_IngressType {
-	if m != nil {
-		return m.IngressType
-	}
-	return nil
-}
-
 func (x *Ingress) GetPath() string {
-	if x, ok := x.GetIngressType().(*Ingress_Path); ok {
+	if x != nil {
 		return x.Path
 	}
 	return ""
 }
 
 func (x *Ingress) GetHost() string {
-	if x, ok := x.GetIngressType().(*Ingress_Host); ok {
+	if x != nil {
 		return x.Host
 	}
 	return ""
 }
 
-type isIngress_IngressType interface {
-	isIngress_IngressType()
-}
-
-type Ingress_Path struct {
-	// path to listen for incoming events.
-	Path string `protobuf:"bytes,2,opt,name=path,proto3,oneof"`
-}
-
-type Ingress_Host struct {
-	// host header to match
-	Host string `protobuf:"bytes,3,opt,name=host,proto3,oneof"`
-}
-
-func (*Ingress_Path) isIngress_IngressType() {}
-
-func (*Ingress_Host) isIngress_IngressType() {}
-
 // Kubernetes resource reference.
 type Reference struct {
 	state         protoimpl.MessageState
@@ -1935,117 +1915,116 @@ var file_contract_proto_rawDesc = []byte{
 	0x10, 0x2e, 0x44, 0x69, 0x61, 0x6c, 0x65, 0x63, 0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74, 0x65,
 	0x72, 0x52, 0x0f, 0x64, 0x69, 0x61, 0x6c, 0x65, 0x63, 0x74, 0x65, 0x64, 0x46, 0x69, 0x6c, 0x74,
 	0x65, 0x72, 0x42, 0x0f, 0x0a, 0x0d, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x53, 0x74, 0x72, 0x61, 0x74,
-	0x65, 0x67, 0x79, 0x22, 0x74, 0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2e,
+	0x65, 0x67, 0x79, 0x22, 0x61, 0x0a, 0x07, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x12, 0x2e,
 	0x0a, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20,
 	0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64,
-	0x65, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x14,
-	0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x04,
-	0x70, 0x61, 0x74, 0x68, 0x12, 0x14, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x09, 0x48, 0x00, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x42, 0x0d, 0x0a, 0x0b, 0x69, 0x6e,
-	0x67, 0x72, 0x65, 0x73, 0x73, 0x54, 0x79, 0x70, 0x65, 0x22, 0x6b, 0x0a, 0x09, 0x52, 0x65, 0x66,
-	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x75, 0x69, 0x64, 0x18, 0x01,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x75, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61,
-	0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e,
-	0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65,
-	0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07,
-	0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76,
-	0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74,
-	0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66,
-	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x12, 0x42, 0x0a, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32,
-	0x12, 0x2e, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x52, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66,
-	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79, 0x46, 0x69,
-	0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c, 0x0a, 0x09,
-	0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x09, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x22, 0x0a, 0x05, 0x66, 0x69,
-	0x65, 0x6c, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65, 0x63, 0x72,
-	0x65, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x6f,
-	0x0a, 0x14, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66,
-	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63,
-	0x6f, 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f,
-	0x63, 0x6f, 0x6c, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x30, 0x0a,
-	0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28,
-	0x0b, 0x32, 0x10, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x52, 0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22,
-	0x9a, 0x01, 0x0a, 0x13, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76,
-	0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e,
-	0x73, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x43, 0x6c,
-	0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65,
-	0x73, 0x2e, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72,
-	0x79, 0x52, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3d, 0x0a,
-	0x0f, 0x45, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79,
-	0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b,
-	0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1, 0x03, 0x0a,
-	0x08, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64,
-	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74,
-	0x6f, 0x70, 0x69, 0x63, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70,
-	0x69, 0x63, 0x73, 0x12, 0x2a, 0x0a, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70,
-	0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62,
-	0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12,
-	0x22, 0x0a, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b,
-	0x32, 0x08, 0x2e, 0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x69, 0x6e, 0x67, 0x72,
-	0x65, 0x73, 0x73, 0x12, 0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e,
-	0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67, 0x72, 0x65,
-	0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73,
-	0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x23, 0x0a, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73,
-	0x65, 0x73, 0x18, 0x06, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73,
-	0x73, 0x52, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0a, 0x61,
-	0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32,
-	0x06, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e,
-	0x74, 0x41, 0x75, 0x74, 0x68, 0x12, 0x2c, 0x0a, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63,
-	0x72, 0x65, 0x74, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65,
-	0x72, 0x65, 0x6e, 0x63, 0x65, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63,
-	0x72, 0x65, 0x74, 0x12, 0x41, 0x0a, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68,
-	0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x4d,
+	0x65, 0x52, 0x0b, 0x63, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x12,
+	0x0a, 0x04, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x70, 0x61,
+	0x74, 0x68, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09,
+	0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x22, 0x6b, 0x0a, 0x09, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
+	0x6e, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x04, 0x75, 0x75, 0x69, 0x64, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73,
+	0x70, 0x61, 0x63, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65,
+	0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x76, 0x65, 0x72,
+	0x73, 0x69, 0x6f, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x76, 0x65, 0x72, 0x73,
+	0x69, 0x6f, 0x6e, 0x22, 0x7f, 0x0a, 0x0f, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66,
+	0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x28, 0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65,
+	0x6e, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65,
+	0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x12, 0x42, 0x0a, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65,
+	0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x4b,
+	0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x52, 0x12, 0x6b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
+	0x6e, 0x63, 0x65, 0x73, 0x22, 0x55, 0x0a, 0x11, 0x4b, 0x65, 0x79, 0x46, 0x69, 0x65, 0x6c, 0x64,
+	0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x73, 0x65, 0x63,
+	0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x73, 0x65,
+	0x63, 0x72, 0x65, 0x74, 0x4b, 0x65, 0x79, 0x12, 0x22, 0x0a, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64,
+	0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x0c, 0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x46,
+	0x69, 0x65, 0x6c, 0x64, 0x52, 0x05, 0x66, 0x69, 0x65, 0x6c, 0x64, 0x22, 0x6f, 0x0a, 0x14, 0x4d,
 	0x75, 0x6c, 0x74, 0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65,
-	0x6e, 0x63, 0x65, 0x48, 0x00, 0x52, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68,
-	0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x12, 0x46, 0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45,
-	0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x18, 0x0a, 0x20,
-	0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74,
-	0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64,
-	0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x28,
-	0x0a, 0x09, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28,
-	0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72,
-	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x42, 0x06, 0x0a, 0x04, 0x41, 0x75, 0x74, 0x68,
-	0x22, 0x53, 0x0a, 0x08, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x12, 0x1e, 0x0a, 0x0a,
-	0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04,
-	0x52, 0x0a, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x09,
-	0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32,
-	0x09, 0x2e, 0x52, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f,
-	0x75, 0x72, 0x63, 0x65, 0x73, 0x2a, 0x2c, 0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66,
-	0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65,
-	0x6e, 0x74, 0x69, 0x61, 0x6c, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61,
-	0x72, 0x10, 0x01, 0x2a, 0x2b, 0x0a, 0x0d, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f,
-	0x72, 0x64, 0x65, 0x72, 0x12, 0x0d, 0x0a, 0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45,
-	0x44, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01,
-	0x2a, 0x3d, 0x0a, 0x07, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x53,
-	0x74, 0x72, 0x69, 0x6e, 0x67, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74, 0x65, 0x67,
-	0x65, 0x72, 0x10, 0x01, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x10, 0x02,
-	0x12, 0x0d, 0x0a, 0x09, 0x42, 0x79, 0x74, 0x65, 0x41, 0x72, 0x72, 0x61, 0x79, 0x10, 0x03, 0x2a,
-	0x29, 0x0a, 0x0b, 0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a,
-	0x0a, 0x06, 0x42, 0x49, 0x4e, 0x41, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54,
-	0x52, 0x55, 0x43, 0x54, 0x55, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b, 0x53, 0x65,
-	0x63, 0x72, 0x65, 0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53,
-	0x4c, 0x5f, 0x4d, 0x45, 0x43, 0x48, 0x41, 0x4e, 0x49, 0x53, 0x4d, 0x10, 0x00, 0x12, 0x0a, 0x0a,
-	0x06, 0x43, 0x41, 0x5f, 0x43, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45,
-	0x52, 0x5f, 0x43, 0x52, 0x54, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f,
-	0x4b, 0x45, 0x59, 0x10, 0x03, 0x12, 0x08, 0x0a, 0x04, 0x55, 0x53, 0x45, 0x52, 0x10, 0x04, 0x12,
-	0x0c, 0x0a, 0x08, 0x50, 0x41, 0x53, 0x53, 0x57, 0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a, 0x44, 0x0a,
-	0x08, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50, 0x4c, 0x41,
-	0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c,
-	0x5f, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x01, 0x12, 0x07, 0x0a, 0x03,
-	0x53, 0x53, 0x4c, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x53, 0x53,
-	0x4c, 0x10, 0x03, 0x42, 0x5b, 0x0a, 0x2a, 0x64, 0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61, 0x74, 0x69,
-	0x76, 0x65, 0x2e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61, 0x66, 0x6b,
-	0x61, 0x2e, 0x62, 0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63,
-	0x74, 0x42, 0x11, 0x44, 0x61, 0x74, 0x61, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f, 0x6e, 0x74,
-	0x72, 0x61, 0x63, 0x74, 0x5a, 0x1a, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d, 0x70, 0x6c,
-	0x61, 0x6e, 0x65, 0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74,
-	0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
+	0x6e, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x18,
+	0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x09, 0x2e, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c,
+	0x52, 0x08, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x30, 0x0a, 0x0a, 0x72, 0x65,
+	0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x10,
+	0x2e, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x52, 0x0a, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x73, 0x22, 0x9a, 0x01, 0x0a,
+	0x13, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72,
+	0x69, 0x64, 0x65, 0x73, 0x12, 0x44, 0x0a, 0x0a, 0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f,
+	0x6e, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x24, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64,
+	0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x2e, 0x45,
+	0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x0a,
+	0x65, 0x78, 0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x1a, 0x3d, 0x0a, 0x0f, 0x45, 0x78,
+	0x74, 0x65, 0x6e, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a,
+	0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12,
+	0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05,
+	0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xf1, 0x03, 0x0a, 0x08, 0x52, 0x65,
+	0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x01, 0x20,
+	0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x6f, 0x70, 0x69,
+	0x63, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x6f, 0x70, 0x69, 0x63, 0x73,
+	0x12, 0x2a, 0x0a, 0x10, 0x62, 0x6f, 0x6f, 0x74, 0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72,
+	0x76, 0x65, 0x72, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x10, 0x62, 0x6f, 0x6f, 0x74,
+	0x73, 0x74, 0x72, 0x61, 0x70, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x73, 0x12, 0x22, 0x0a, 0x07,
+	0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x08, 0x2e,
+	0x49, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x07, 0x69, 0x6e, 0x67, 0x72, 0x65, 0x73, 0x73,
+	0x12, 0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67,
+	0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0d, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43,
+	0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e,
+	0x66, 0x69, 0x67, 0x12, 0x23, 0x0a, 0x08, 0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x18,
+	0x06, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x52, 0x08,
+	0x65, 0x67, 0x72, 0x65, 0x73, 0x73, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x0a, 0x61, 0x62, 0x73, 0x65,
+	0x6e, 0x74, 0x41, 0x75, 0x74, 0x68, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45,
+	0x6d, 0x70, 0x74, 0x79, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x62, 0x73, 0x65, 0x6e, 0x74, 0x41, 0x75,
+	0x74, 0x68, 0x12, 0x2c, 0x0a, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74,
+	0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a, 0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e,
+	0x63, 0x65, 0x48, 0x00, 0x52, 0x0a, 0x61, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74,
+	0x12, 0x41, 0x0a, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63,
+	0x72, 0x65, 0x74, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, 0x4d, 0x75, 0x6c, 0x74,
+	0x69, 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65,
+	0x48, 0x00, 0x52, 0x0f, 0x6d, 0x75, 0x6c, 0x74, 0x69, 0x41, 0x75, 0x74, 0x68, 0x53, 0x65, 0x63,
+	0x72, 0x65, 0x74, 0x12, 0x46, 0x0a, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e,
+	0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b,
+	0x32, 0x14, 0x2e, 0x43, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x4f, 0x76, 0x65,
+	0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x52, 0x13, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x45, 0x76, 0x65,
+	0x6e, 0x74, 0x4f, 0x76, 0x65, 0x72, 0x72, 0x69, 0x64, 0x65, 0x73, 0x12, 0x28, 0x0a, 0x09, 0x72,
+	0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x0a,
+	0x2e, 0x52, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6e, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x66, 0x65,
+	0x72, 0x65, 0x6e, 0x63, 0x65, 0x42, 0x06, 0x0a, 0x04, 0x41, 0x75, 0x74, 0x68, 0x22, 0x53, 0x0a,
+	0x08, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x12, 0x1e, 0x0a, 0x0a, 0x67, 0x65, 0x6e,
+	0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x04, 0x52, 0x0a, 0x67,
+	0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x27, 0x0a, 0x09, 0x72, 0x65, 0x73,
+	0x6f, 0x75, 0x72, 0x63, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x09, 0x2e, 0x52,
+	0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x52, 0x09, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63,
+	0x65, 0x73, 0x2a, 0x2c, 0x0a, 0x0d, 0x42, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c,
+	0x69, 0x63, 0x79, 0x12, 0x0f, 0x0a, 0x0b, 0x45, 0x78, 0x70, 0x6f, 0x6e, 0x65, 0x6e, 0x74, 0x69,
+	0x61, 0x6c, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x4c, 0x69, 0x6e, 0x65, 0x61, 0x72, 0x10, 0x01,
+	0x2a, 0x2b, 0x0a, 0x0d, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65,
+	0x72, 0x12, 0x0d, 0x0a, 0x09, 0x55, 0x4e, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x00,
+	0x12, 0x0b, 0x0a, 0x07, 0x4f, 0x52, 0x44, 0x45, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x3d, 0x0a,
+	0x07, 0x4b, 0x65, 0x79, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x53, 0x74, 0x72, 0x69,
+	0x6e, 0x67, 0x10, 0x00, 0x12, 0x0b, 0x0a, 0x07, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x65, 0x72, 0x10,
+	0x01, 0x12, 0x0a, 0x0a, 0x06, 0x44, 0x6f, 0x75, 0x62, 0x6c, 0x65, 0x10, 0x02, 0x12, 0x0d, 0x0a,
+	0x09, 0x42, 0x79, 0x74, 0x65, 0x41, 0x72, 0x72, 0x61, 0x79, 0x10, 0x03, 0x2a, 0x29, 0x0a, 0x0b,
+	0x43, 0x6f, 0x6e, 0x74, 0x65, 0x6e, 0x74, 0x4d, 0x6f, 0x64, 0x65, 0x12, 0x0a, 0x0a, 0x06, 0x42,
+	0x49, 0x4e, 0x41, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x53, 0x54, 0x52, 0x55, 0x43,
+	0x54, 0x55, 0x52, 0x45, 0x44, 0x10, 0x01, 0x2a, 0x61, 0x0a, 0x0b, 0x53, 0x65, 0x63, 0x72, 0x65,
+	0x74, 0x46, 0x69, 0x65, 0x6c, 0x64, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x4d,
+	0x45, 0x43, 0x48, 0x41, 0x4e, 0x49, 0x53, 0x4d, 0x10, 0x00, 0x12, 0x0a, 0x0a, 0x06, 0x43, 0x41,
+	0x5f, 0x43, 0x52, 0x54, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x43,
+	0x52, 0x54, 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x55, 0x53, 0x45, 0x52, 0x5f, 0x4b, 0x45, 0x59,
+	0x10, 0x03, 0x12, 0x08, 0x0a, 0x04, 0x55, 0x53, 0x45, 0x52, 0x10, 0x04, 0x12, 0x0c, 0x0a, 0x08,
+	0x50, 0x41, 0x53, 0x53, 0x57, 0x4f, 0x52, 0x44, 0x10, 0x05, 0x2a, 0x44, 0x0a, 0x08, 0x50, 0x72,
+	0x6f, 0x74, 0x6f, 0x63, 0x6f, 0x6c, 0x12, 0x0d, 0x0a, 0x09, 0x50, 0x4c, 0x41, 0x49, 0x4e, 0x54,
+	0x45, 0x58, 0x54, 0x10, 0x00, 0x12, 0x12, 0x0a, 0x0e, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x50, 0x4c,
+	0x41, 0x49, 0x4e, 0x54, 0x45, 0x58, 0x54, 0x10, 0x01, 0x12, 0x07, 0x0a, 0x03, 0x53, 0x53, 0x4c,
+	0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x41, 0x53, 0x4c, 0x5f, 0x53, 0x53, 0x4c, 0x10, 0x03,
+	0x42, 0x5b, 0x0a, 0x2a, 0x64, 0x65, 0x76, 0x2e, 0x6b, 0x6e, 0x61, 0x74, 0x69, 0x76, 0x65, 0x2e,
+	0x65, 0x76, 0x65, 0x6e, 0x74, 0x69, 0x6e, 0x67, 0x2e, 0x6b, 0x61, 0x66, 0x6b, 0x61, 0x2e, 0x62,
+	0x72, 0x6f, 0x6b, 0x65, 0x72, 0x2e, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x42, 0x11,
+	0x44, 0x61, 0x74, 0x61, 0x50, 0x6c, 0x61, 0x6e, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63,
+	0x74, 0x5a, 0x1a, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x6f, 0x6c, 0x2d, 0x70, 0x6c, 0x61, 0x6e, 0x65,
+	0x2f, 0x70, 0x6b, 0x67, 0x2f, 0x63, 0x6f, 0x6e, 0x74, 0x72, 0x61, 0x63, 0x74, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -2403,10 +2382,6 @@ func file_contract_proto_init() {
 		(*Egress_ReplyToOriginalTopic)(nil),
 		(*Egress_DiscardReply)(nil),
 	}
-	file_contract_proto_msgTypes[12].OneofWrappers = []interface{}{
-		(*Ingress_Path)(nil),
-		(*Ingress_Host)(nil),
-	}
 	file_contract_proto_msgTypes[18].OneofWrappers = []interface{}{
 		(*Resource_AbsentAuth)(nil),
 		(*Resource_AuthSecret)(nil),
diff --git a/control-plane/pkg/contract/log_marshaller.go b/control-plane/pkg/contract/log_marshaller.go
index 6289e7453..568e3b8ab 100644
--- a/control-plane/pkg/contract/log_marshaller.go
+++ b/control-plane/pkg/contract/log_marshaller.go
@@ -67,13 +67,8 @@ func (x *Resource) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
 }
 
 func (x *Ingress) MarshalLogObject(encoder zapcore.ObjectEncoder) error {
-	switch it := x.IngressType.(type) {
-	case *Ingress_Path:
-		encoder.AddString("ingress_path", it.Path)
-	case *Ingress_Host:
-		encoder.AddString("ingress_host", it.Host)
-	}
-
+	encoder.AddString("ingress_path", x.Path)
+	encoder.AddString("ingress_host", x.Host)
 	encoder.AddString("contentMode", x.ContentMode.String())
 	return nil
 }
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index ca69b2c57..671d390f5 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -449,9 +449,7 @@ func (r *Reconciler) reconcilerBrokerResource(ctx context.Context, topic string,
 		Uid:    string(broker.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			IngressType: &contract.Ingress_Path{
-				Path: receiver.PathFromObject(broker),
-			},
+			Path: receiver.PathFromObject(broker),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/channel/channel.go b/control-plane/pkg/reconciler/channel/channel.go
index da58834d6..fb5251fda 100644
--- a/control-plane/pkg/reconciler/channel/channel.go
+++ b/control-plane/pkg/reconciler/channel/channel.go
@@ -557,9 +557,7 @@ func (r *Reconciler) getChannelContractResource(ctx context.Context, topic strin
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			IngressType: &contract.Ingress_Path{
-				Path: receiver.PathFromObject(channel),
-			},
+			Path: receiver.PathFromObject(channel),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2.go b/control-plane/pkg/reconciler/channel/v2/channelv2.go
index 417a2a698..8a12625b6 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2.go
@@ -564,9 +564,7 @@ func (r *Reconciler) getChannelContractResource(topic string, channel *messaging
 		Uid:    string(channel.UID),
 		Topics: []string{topic},
 		Ingress: &contract.Ingress{
-			IngressType: &contract.Ingress_Path{
-				Path: receiver.PathFromObject(channel),
-			},
+			Path: receiver.PathFromObject(channel),
 		},
 		BootstrapServers: config.GetBootstrapServers(),
 		Reference: &contract.Reference{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 8b7845faf..2d68595ab 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -173,7 +173,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 		Topics: []string{ks.Spec.Topic},
 		Ingress: &contract.Ingress{
 			ContentMode: coreconfig.ContentModeFromString(*ks.Spec.ContentMode),
-			IngressType: &contract.Ingress_Path{Path: receiver.PathFromObject(ks)},
+			Path:        receiver.PathFromObject(ks),
 		},
 		BootstrapServers: kafka.BootstrapServersCommaSeparated(ks.Spec.BootstrapServers),
 		Reference: &contract.Reference{
diff --git a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
index 30eb88b9e..330eed3a5 100644
--- a/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
+++ b/data-plane/contract/src/main/java/dev/knative/eventing/kafka/broker/contract/DataPlaneContract.java
@@ -13705,15 +13705,6 @@ public final class DataPlaneContract {
      */
     dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ContentMode getContentMode();
 
-    /**
-     * <pre>
-     * path to listen for incoming events.
-     * </pre>
-     *
-     * <code>string path = 2;</code>
-     * @return Whether the path field is set.
-     */
-    boolean hasPath();
     /**
      * <pre>
      * path to listen for incoming events.
@@ -13734,15 +13725,6 @@ public final class DataPlaneContract {
     com.google.protobuf.ByteString
         getPathBytes();
 
-    /**
-     * <pre>
-     * host header to match
-     * </pre>
-     *
-     * <code>string host = 3;</code>
-     * @return Whether the host field is set.
-     */
-    boolean hasHost();
     /**
      * <pre>
      * host header to match
@@ -13762,10 +13744,18 @@ public final class DataPlaneContract {
      */
     com.google.protobuf.ByteString
         getHostBytes();
-
-    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress.IngressTypeCase getIngressTypeCase();
   }
   /**
+   * <pre>
+   * Ingress is the definition for HTTP ingress that is receiving the events
+   * into the Knative Kafka component.
+   * path and host fields are used for identifying the targets. They are exclusive.
+   * When a request comes with "/some-path", hostname will not be checked.
+   * When a request comes with "/", only hostname matching will be done.
+   * It is allowed to specify both path and host in ingress contract
+   * to support both modes.
+   * </pre>
+   *
    * Protobuf type {@code Ingress}
    */
   public static final class Ingress extends
@@ -13779,6 +13769,8 @@ public final class DataPlaneContract {
     }
     private Ingress() {
       contentMode_ = 0;
+      path_ = "";
+      host_ = "";
     }
 
     @java.lang.Override
@@ -13819,14 +13811,14 @@ public final class DataPlaneContract {
             }
             case 18: {
               java.lang.String s = input.readStringRequireUtf8();
-              ingressTypeCase_ = 2;
-              ingressType_ = s;
+
+              path_ = s;
               break;
             }
             case 26: {
               java.lang.String s = input.readStringRequireUtf8();
-              ingressTypeCase_ = 3;
-              ingressType_ = s;
+
+              host_ = s;
               break;
             }
             default: {
@@ -13861,47 +13853,6 @@ public final class DataPlaneContract {
               dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress.class, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress.Builder.class);
     }
 
-    private int ingressTypeCase_ = 0;
-    private java.lang.Object ingressType_;
-    public enum IngressTypeCase
-        implements com.google.protobuf.Internal.EnumLite,
-            com.google.protobuf.AbstractMessage.InternalOneOfEnum {
-      PATH(2),
-      HOST(3),
-      INGRESSTYPE_NOT_SET(0);
-      private final int value;
-      private IngressTypeCase(int value) {
-        this.value = value;
-      }
-      /**
-       * @param value The number of the enum to look for.
-       * @return The enum associated with the given number.
-       * @deprecated Use {@link #forNumber(int)} instead.
-       */
-      @java.lang.Deprecated
-      public static IngressTypeCase valueOf(int value) {
-        return forNumber(value);
-      }
-
-      public static IngressTypeCase forNumber(int value) {
-        switch (value) {
-          case 2: return PATH;
-          case 3: return HOST;
-          case 0: return INGRESSTYPE_NOT_SET;
-          default: return null;
-        }
-      }
-      public int getNumber() {
-        return this.value;
-      }
-    };
-
-    public IngressTypeCase
-    getIngressTypeCase() {
-      return IngressTypeCase.forNumber(
-          ingressTypeCase_);
-    }
-
     public static final int CONTENTMODE_FIELD_NUMBER = 1;
     private int contentMode_;
     /**
@@ -13930,17 +13881,7 @@ public final class DataPlaneContract {
     }
 
     public static final int PATH_FIELD_NUMBER = 2;
-    /**
-     * <pre>
-     * path to listen for incoming events.
-     * </pre>
-     *
-     * <code>string path = 2;</code>
-     * @return Whether the path field is set.
-     */
-    public boolean hasPath() {
-      return ingressTypeCase_ == 2;
-    }
+    private volatile java.lang.Object path_;
     /**
      * <pre>
      * path to listen for incoming events.
@@ -13949,20 +13890,16 @@ public final class DataPlaneContract {
      * <code>string path = 2;</code>
      * @return The path.
      */
+    @java.lang.Override
     public java.lang.String getPath() {
-      java.lang.Object ref = "";
-      if (ingressTypeCase_ == 2) {
-        ref = ingressType_;
-      }
+      java.lang.Object ref = path_;
       if (ref instanceof java.lang.String) {
         return (java.lang.String) ref;
       } else {
         com.google.protobuf.ByteString bs = 
             (com.google.protobuf.ByteString) ref;
         java.lang.String s = bs.toStringUtf8();
-        if (ingressTypeCase_ == 2) {
-          ingressType_ = s;
-        }
+        path_ = s;
         return s;
       }
     }
@@ -13974,19 +13911,15 @@ public final class DataPlaneContract {
      * <code>string path = 2;</code>
      * @return The bytes for path.
      */
+    @java.lang.Override
     public com.google.protobuf.ByteString
         getPathBytes() {
-      java.lang.Object ref = "";
-      if (ingressTypeCase_ == 2) {
-        ref = ingressType_;
-      }
+      java.lang.Object ref = path_;
       if (ref instanceof java.lang.String) {
         com.google.protobuf.ByteString b = 
             com.google.protobuf.ByteString.copyFromUtf8(
                 (java.lang.String) ref);
-        if (ingressTypeCase_ == 2) {
-          ingressType_ = b;
-        }
+        path_ = b;
         return b;
       } else {
         return (com.google.protobuf.ByteString) ref;
@@ -13994,17 +13927,7 @@ public final class DataPlaneContract {
     }
 
     public static final int HOST_FIELD_NUMBER = 3;
-    /**
-     * <pre>
-     * host header to match
-     * </pre>
-     *
-     * <code>string host = 3;</code>
-     * @return Whether the host field is set.
-     */
-    public boolean hasHost() {
-      return ingressTypeCase_ == 3;
-    }
+    private volatile java.lang.Object host_;
     /**
      * <pre>
      * host header to match
@@ -14013,20 +13936,16 @@ public final class DataPlaneContract {
      * <code>string host = 3;</code>
      * @return The host.
      */
+    @java.lang.Override
     public java.lang.String getHost() {
-      java.lang.Object ref = "";
-      if (ingressTypeCase_ == 3) {
-        ref = ingressType_;
-      }
+      java.lang.Object ref = host_;
       if (ref instanceof java.lang.String) {
         return (java.lang.String) ref;
       } else {
         com.google.protobuf.ByteString bs = 
             (com.google.protobuf.ByteString) ref;
         java.lang.String s = bs.toStringUtf8();
-        if (ingressTypeCase_ == 3) {
-          ingressType_ = s;
-        }
+        host_ = s;
         return s;
       }
     }
@@ -14038,19 +13957,15 @@ public final class DataPlaneContract {
      * <code>string host = 3;</code>
      * @return The bytes for host.
      */
+    @java.lang.Override
     public com.google.protobuf.ByteString
         getHostBytes() {
-      java.lang.Object ref = "";
-      if (ingressTypeCase_ == 3) {
-        ref = ingressType_;
-      }
+      java.lang.Object ref = host_;
       if (ref instanceof java.lang.String) {
         com.google.protobuf.ByteString b = 
             com.google.protobuf.ByteString.copyFromUtf8(
                 (java.lang.String) ref);
-        if (ingressTypeCase_ == 3) {
-          ingressType_ = b;
-        }
+        host_ = b;
         return b;
       } else {
         return (com.google.protobuf.ByteString) ref;
@@ -14074,11 +13989,11 @@ public final class DataPlaneContract {
       if (contentMode_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.ContentMode.BINARY.getNumber()) {
         output.writeEnum(1, contentMode_);
       }
-      if (ingressTypeCase_ == 2) {
-        com.google.protobuf.GeneratedMessageV3.writeString(output, 2, ingressType_);
+      if (!getPathBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 2, path_);
       }
-      if (ingressTypeCase_ == 3) {
-        com.google.protobuf.GeneratedMessageV3.writeString(output, 3, ingressType_);
+      if (!getHostBytes().isEmpty()) {
+        com.google.protobuf.GeneratedMessageV3.writeString(output, 3, host_);
       }
       unknownFields.writeTo(output);
     }
@@ -14093,11 +14008,11 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(1, contentMode_);
       }
-      if (ingressTypeCase_ == 2) {
-        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, ingressType_);
+      if (!getPathBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, path_);
       }
-      if (ingressTypeCase_ == 3) {
-        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(3, ingressType_);
+      if (!getHostBytes().isEmpty()) {
+        size += com.google.protobuf.GeneratedMessageV3.computeStringSize(3, host_);
       }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
@@ -14115,19 +14030,10 @@ public final class DataPlaneContract {
       dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress other = (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress) obj;
 
       if (contentMode_ != other.contentMode_) return false;
-      if (!getIngressTypeCase().equals(other.getIngressTypeCase())) return false;
-      switch (ingressTypeCase_) {
-        case 2:
-          if (!getPath()
-              .equals(other.getPath())) return false;
-          break;
-        case 3:
-          if (!getHost()
-              .equals(other.getHost())) return false;
-          break;
-        case 0:
-        default:
-      }
+      if (!getPath()
+          .equals(other.getPath())) return false;
+      if (!getHost()
+          .equals(other.getHost())) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -14141,18 +14047,10 @@ public final class DataPlaneContract {
       hash = (19 * hash) + getDescriptor().hashCode();
       hash = (37 * hash) + CONTENTMODE_FIELD_NUMBER;
       hash = (53 * hash) + contentMode_;
-      switch (ingressTypeCase_) {
-        case 2:
-          hash = (37 * hash) + PATH_FIELD_NUMBER;
-          hash = (53 * hash) + getPath().hashCode();
-          break;
-        case 3:
-          hash = (37 * hash) + HOST_FIELD_NUMBER;
-          hash = (53 * hash) + getHost().hashCode();
-          break;
-        case 0:
-        default:
-      }
+      hash = (37 * hash) + PATH_FIELD_NUMBER;
+      hash = (53 * hash) + getPath().hashCode();
+      hash = (37 * hash) + HOST_FIELD_NUMBER;
+      hash = (53 * hash) + getHost().hashCode();
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -14249,6 +14147,16 @@ public final class DataPlaneContract {
       return builder;
     }
     /**
+     * <pre>
+     * Ingress is the definition for HTTP ingress that is receiving the events
+     * into the Knative Kafka component.
+     * path and host fields are used for identifying the targets. They are exclusive.
+     * When a request comes with "/some-path", hostname will not be checked.
+     * When a request comes with "/", only hostname matching will be done.
+     * It is allowed to specify both path and host in ingress contract
+     * to support both modes.
+     * </pre>
+     *
      * Protobuf type {@code Ingress}
      */
     public static final class Builder extends
@@ -14288,8 +14196,10 @@ public final class DataPlaneContract {
         super.clear();
         contentMode_ = 0;
 
-        ingressTypeCase_ = 0;
-        ingressType_ = null;
+        path_ = "";
+
+        host_ = "";
+
         return this;
       }
 
@@ -14317,13 +14227,8 @@ public final class DataPlaneContract {
       public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress buildPartial() {
         dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress result = new dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Ingress(this);
         result.contentMode_ = contentMode_;
-        if (ingressTypeCase_ == 2) {
-          result.ingressType_ = ingressType_;
-        }
-        if (ingressTypeCase_ == 3) {
-          result.ingressType_ = ingressType_;
-        }
-        result.ingressTypeCase_ = ingressTypeCase_;
+        result.path_ = path_;
+        result.host_ = host_;
         onBuilt();
         return result;
       }
@@ -14375,22 +14280,13 @@ public final class DataPlaneContract {
         if (other.contentMode_ != 0) {
           setContentModeValue(other.getContentModeValue());
         }
-        switch (other.getIngressTypeCase()) {
-          case PATH: {
-            ingressTypeCase_ = 2;
-            ingressType_ = other.ingressType_;
-            onChanged();
-            break;
-          }
-          case HOST: {
-            ingressTypeCase_ = 3;
-            ingressType_ = other.ingressType_;
-            onChanged();
-            break;
-          }
-          case INGRESSTYPE_NOT_SET: {
-            break;
-          }
+        if (!other.getPath().isEmpty()) {
+          path_ = other.path_;
+          onChanged();
+        }
+        if (!other.getHost().isEmpty()) {
+          host_ = other.host_;
+          onChanged();
         }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
@@ -14420,21 +14316,6 @@ public final class DataPlaneContract {
         }
         return this;
       }
-      private int ingressTypeCase_ = 0;
-      private java.lang.Object ingressType_;
-      public IngressTypeCase
-          getIngressTypeCase() {
-        return IngressTypeCase.forNumber(
-            ingressTypeCase_);
-      }
-
-      public Builder clearIngressType() {
-        ingressTypeCase_ = 0;
-        ingressType_ = null;
-        onChanged();
-        return this;
-      }
-
 
       private int contentMode_ = 0;
       /**
@@ -14510,18 +14391,7 @@ public final class DataPlaneContract {
         return this;
       }
 
-      /**
-       * <pre>
-       * path to listen for incoming events.
-       * </pre>
-       *
-       * <code>string path = 2;</code>
-       * @return Whether the path field is set.
-       */
-      @java.lang.Override
-      public boolean hasPath() {
-        return ingressTypeCase_ == 2;
-      }
+      private java.lang.Object path_ = "";
       /**
        * <pre>
        * path to listen for incoming events.
@@ -14530,19 +14400,13 @@ public final class DataPlaneContract {
        * <code>string path = 2;</code>
        * @return The path.
        */
-      @java.lang.Override
       public java.lang.String getPath() {
-        java.lang.Object ref = "";
-        if (ingressTypeCase_ == 2) {
-          ref = ingressType_;
-        }
+        java.lang.Object ref = path_;
         if (!(ref instanceof java.lang.String)) {
           com.google.protobuf.ByteString bs =
               (com.google.protobuf.ByteString) ref;
           java.lang.String s = bs.toStringUtf8();
-          if (ingressTypeCase_ == 2) {
-            ingressType_ = s;
-          }
+          path_ = s;
           return s;
         } else {
           return (java.lang.String) ref;
@@ -14556,20 +14420,14 @@ public final class DataPlaneContract {
        * <code>string path = 2;</code>
        * @return The bytes for path.
        */
-      @java.lang.Override
       public com.google.protobuf.ByteString
           getPathBytes() {
-        java.lang.Object ref = "";
-        if (ingressTypeCase_ == 2) {
-          ref = ingressType_;
-        }
+        java.lang.Object ref = path_;
         if (ref instanceof String) {
           com.google.protobuf.ByteString b = 
               com.google.protobuf.ByteString.copyFromUtf8(
                   (java.lang.String) ref);
-          if (ingressTypeCase_ == 2) {
-            ingressType_ = b;
-          }
+          path_ = b;
           return b;
         } else {
           return (com.google.protobuf.ByteString) ref;
@@ -14589,8 +14447,8 @@ public final class DataPlaneContract {
         if (value == null) {
     throw new NullPointerException();
   }
-  ingressTypeCase_ = 2;
-        ingressType_ = value;
+  
+        path_ = value;
         onChanged();
         return this;
       }
@@ -14603,11 +14461,9 @@ public final class DataPlaneContract {
        * @return This builder for chaining.
        */
       public Builder clearPath() {
-        if (ingressTypeCase_ == 2) {
-          ingressTypeCase_ = 0;
-          ingressType_ = null;
-          onChanged();
-        }
+        
+        path_ = getDefaultInstance().getPath();
+        onChanged();
         return this;
       }
       /**
@@ -14625,24 +14481,13 @@ public final class DataPlaneContract {
     throw new NullPointerException();
   }
   checkByteStringIsUtf8(value);
-        ingressTypeCase_ = 2;
-        ingressType_ = value;
+        
+        path_ = value;
         onChanged();
         return this;
       }
 
-      /**
-       * <pre>
-       * host header to match
-       * </pre>
-       *
-       * <code>string host = 3;</code>
-       * @return Whether the host field is set.
-       */
-      @java.lang.Override
-      public boolean hasHost() {
-        return ingressTypeCase_ == 3;
-      }
+      private java.lang.Object host_ = "";
       /**
        * <pre>
        * host header to match
@@ -14651,19 +14496,13 @@ public final class DataPlaneContract {
        * <code>string host = 3;</code>
        * @return The host.
        */
-      @java.lang.Override
       public java.lang.String getHost() {
-        java.lang.Object ref = "";
-        if (ingressTypeCase_ == 3) {
-          ref = ingressType_;
-        }
+        java.lang.Object ref = host_;
         if (!(ref instanceof java.lang.String)) {
           com.google.protobuf.ByteString bs =
               (com.google.protobuf.ByteString) ref;
           java.lang.String s = bs.toStringUtf8();
-          if (ingressTypeCase_ == 3) {
-            ingressType_ = s;
-          }
+          host_ = s;
           return s;
         } else {
           return (java.lang.String) ref;
@@ -14677,20 +14516,14 @@ public final class DataPlaneContract {
        * <code>string host = 3;</code>
        * @return The bytes for host.
        */
-      @java.lang.Override
       public com.google.protobuf.ByteString
           getHostBytes() {
-        java.lang.Object ref = "";
-        if (ingressTypeCase_ == 3) {
-          ref = ingressType_;
-        }
+        java.lang.Object ref = host_;
         if (ref instanceof String) {
           com.google.protobuf.ByteString b = 
               com.google.protobuf.ByteString.copyFromUtf8(
                   (java.lang.String) ref);
-          if (ingressTypeCase_ == 3) {
-            ingressType_ = b;
-          }
+          host_ = b;
           return b;
         } else {
           return (com.google.protobuf.ByteString) ref;
@@ -14710,8 +14543,8 @@ public final class DataPlaneContract {
         if (value == null) {
     throw new NullPointerException();
   }
-  ingressTypeCase_ = 3;
-        ingressType_ = value;
+  
+        host_ = value;
         onChanged();
         return this;
       }
@@ -14724,11 +14557,9 @@ public final class DataPlaneContract {
        * @return This builder for chaining.
        */
       public Builder clearHost() {
-        if (ingressTypeCase_ == 3) {
-          ingressTypeCase_ = 0;
-          ingressType_ = null;
-          onChanged();
-        }
+        
+        host_ = getDefaultInstance().getHost();
+        onChanged();
         return this;
       }
       /**
@@ -14746,8 +14577,8 @@ public final class DataPlaneContract {
     throw new NullPointerException();
   }
   checkByteStringIsUtf8(value);
-        ingressTypeCase_ = 3;
-        ingressType_ = value;
+        
+        host_ = value;
         onChanged();
         return this;
       }
@@ -24762,43 +24593,43 @@ public final class DataPlaneContract {
       "yOrder\030\010 \001(\0162\016.DeliveryOrder\022\031\n\007keyType\030" +
       "\n \001(\0162\010.KeyType\022\035\n\treference\030\013 \001(\0132\n.Ref" +
       "erence\022)\n\017dialectedFilter\030\014 \003(\0132\020.Dialec" +
-      "tedFilterB\017\n\rreplyStrategy\"[\n\007Ingress\022!\n" +
-      "\013contentMode\030\001 \001(\0162\014.ContentMode\022\016\n\004path" +
-      "\030\002 \001(\tH\000\022\016\n\004host\030\003 \001(\tH\000B\r\n\013ingressType\"" +
-      "K\n\tReference\022\014\n\004uuid\030\001 \001(\t\022\021\n\tnamespace\030" +
-      "\002 \001(\t\022\014\n\004name\030\003 \001(\t\022\017\n\007version\030\004 \001(\t\"`\n\017" +
-      "SecretReference\022\035\n\treference\030\001 \001(\0132\n.Ref" +
-      "erence\022.\n\022keyFieldReferences\030\002 \003(\0132\022.Key" +
-      "FieldReference\"C\n\021KeyFieldReference\022\021\n\ts" +
-      "ecretKey\030\002 \001(\t\022\033\n\005field\030\003 \001(\0162\014.SecretFi" +
-      "eld\"Y\n\024MultiSecretReference\022\033\n\010protocol\030" +
-      "\001 \001(\0162\t.Protocol\022$\n\nreferences\030\002 \003(\0132\020.S" +
-      "ecretReference\"\202\001\n\023CloudEventOverrides\0228" +
-      "\n\nextensions\030\001 \003(\0132$.CloudEventOverrides" +
-      ".ExtensionsEntry\0321\n\017ExtensionsEntry\022\013\n\003k" +
-      "ey\030\001 \001(\t\022\r\n\005value\030\002 \001(\t:\0028\001\"\350\002\n\010Resource" +
-      "\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022\030\n\020bootstr" +
-      "apServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingre" +
-      "ss\022#\n\014egressConfig\030\005 \001(\0132\r.EgressConfig\022" +
-      "\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\nabsentAuth" +
-      "\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret\030\010 \001(\0132\n.R" +
-      "eferenceH\000\0220\n\017multiAuthSecret\030\t \001(\0132\025.Mu" +
-      "ltiSecretReferenceH\000\0221\n\023cloudEventOverri" +
-      "des\030\n \001(\0132\024.CloudEventOverrides\022\035\n\trefer" +
-      "ence\030\013 \001(\0132\n.ReferenceB\006\n\004Auth\"<\n\010Contra" +
-      "ct\022\022\n\ngeneration\030\001 \001(\004\022\034\n\tresources\030\002 \003(" +
-      "\0132\t.Resource*,\n\rBackoffPolicy\022\017\n\013Exponen" +
-      "tial\020\000\022\n\n\006Linear\020\001*+\n\rDeliveryOrder\022\r\n\tU" +
-      "NORDERED\020\000\022\013\n\007ORDERED\020\001*=\n\007KeyType\022\n\n\006St" +
-      "ring\020\000\022\013\n\007Integer\020\001\022\n\n\006Double\020\002\022\r\n\tByteA" +
-      "rray\020\003*)\n\013ContentMode\022\n\n\006BINARY\020\000\022\016\n\nSTR" +
-      "UCTURED\020\001*a\n\013SecretField\022\022\n\016SASL_MECHANI" +
-      "SM\020\000\022\n\n\006CA_CRT\020\001\022\014\n\010USER_CRT\020\002\022\014\n\010USER_K" +
-      "EY\020\003\022\010\n\004USER\020\004\022\014\n\010PASSWORD\020\005*D\n\010Protocol" +
-      "\022\r\n\tPLAINTEXT\020\000\022\022\n\016SASL_PLAINTEXT\020\001\022\007\n\003S" +
-      "SL\020\002\022\014\n\010SASL_SSL\020\003B[\n*dev.knative.eventi" +
-      "ng.kafka.broker.contractB\021DataPlaneContr" +
-      "actZ\032control-plane/pkg/contractb\006proto3"
+      "tedFilterB\017\n\rreplyStrategy\"H\n\007Ingress\022!\n" +
+      "\013contentMode\030\001 \001(\0162\014.ContentMode\022\014\n\004path" +
+      "\030\002 \001(\t\022\014\n\004host\030\003 \001(\t\"K\n\tReference\022\014\n\004uui" +
+      "d\030\001 \001(\t\022\021\n\tnamespace\030\002 \001(\t\022\014\n\004name\030\003 \001(\t" +
+      "\022\017\n\007version\030\004 \001(\t\"`\n\017SecretReference\022\035\n\t" +
+      "reference\030\001 \001(\0132\n.Reference\022.\n\022keyFieldR" +
+      "eferences\030\002 \003(\0132\022.KeyFieldReference\"C\n\021K" +
+      "eyFieldReference\022\021\n\tsecretKey\030\002 \001(\t\022\033\n\005f" +
+      "ield\030\003 \001(\0162\014.SecretField\"Y\n\024MultiSecretR" +
+      "eference\022\033\n\010protocol\030\001 \001(\0162\t.Protocol\022$\n" +
+      "\nreferences\030\002 \003(\0132\020.SecretReference\"\202\001\n\023" +
+      "CloudEventOverrides\0228\n\nextensions\030\001 \003(\0132" +
+      "$.CloudEventOverrides.ExtensionsEntry\0321\n" +
+      "\017ExtensionsEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005value\030\002" +
+      " \001(\t:\0028\001\"\350\002\n\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006to" +
+      "pics\030\002 \003(\t\022\030\n\020bootstrapServers\030\003 \001(\t\022\031\n\007" +
+      "ingress\030\004 \001(\0132\010.Ingress\022#\n\014egressConfig\030" +
+      "\005 \001(\0132\r.EgressConfig\022\031\n\010egresses\030\006 \003(\0132\007" +
+      ".Egress\022\034\n\nabsentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n" +
+      "\nauthSecret\030\010 \001(\0132\n.ReferenceH\000\0220\n\017multi" +
+      "AuthSecret\030\t \001(\0132\025.MultiSecretReferenceH" +
+      "\000\0221\n\023cloudEventOverrides\030\n \001(\0132\024.CloudEv" +
+      "entOverrides\022\035\n\treference\030\013 \001(\0132\n.Refere" +
+      "nceB\006\n\004Auth\"<\n\010Contract\022\022\n\ngeneration\030\001 " +
+      "\001(\004\022\034\n\tresources\030\002 \003(\0132\t.Resource*,\n\rBac" +
+      "koffPolicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*" +
+      "+\n\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDER" +
+      "ED\020\001*=\n\007KeyType\022\n\n\006String\020\000\022\013\n\007Integer\020\001" +
+      "\022\n\n\006Double\020\002\022\r\n\tByteArray\020\003*)\n\013ContentMo" +
+      "de\022\n\n\006BINARY\020\000\022\016\n\nSTRUCTURED\020\001*a\n\013Secret" +
+      "Field\022\022\n\016SASL_MECHANISM\020\000\022\n\n\006CA_CRT\020\001\022\014\n" +
+      "\010USER_CRT\020\002\022\014\n\010USER_KEY\020\003\022\010\n\004USER\020\004\022\014\n\010P" +
+      "ASSWORD\020\005*D\n\010Protocol\022\r\n\tPLAINTEXT\020\000\022\022\n\016" +
+      "SASL_PLAINTEXT\020\001\022\007\n\003SSL\020\002\022\014\n\010SASL_SSL\020\003B" +
+      "[\n*dev.knative.eventing.kafka.broker.con" +
+      "tractB\021DataPlaneContractZ\032control-plane/" +
+      "pkg/contractb\006proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -24905,7 +24736,7 @@ public final class DataPlaneContract {
     internal_static_Ingress_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Ingress_descriptor,
-        new java.lang.String[] { "ContentMode", "Path", "Host", "IngressType", });
+        new java.lang.String[] { "ContentMode", "Path", "Host", });
     internal_static_Reference_descriptor =
       getDescriptor().getMessageTypes().get(13);
     internal_static_Reference_fieldAccessorTable = new
diff --git a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
index 35af385ae..283aae181 100644
--- a/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
+++ b/data-plane/core/src/main/java/dev/knative/eventing/kafka/broker/core/reconciler/impl/ResourcesReconcilerImpl.java
@@ -22,6 +22,9 @@ import dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler;
 import dev.knative.eventing.kafka.broker.core.utils.CollectionsUtils;
 import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -31,8 +34,6 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import static dev.knative.eventing.kafka.broker.core.utils.Logging.keyValue;
 
@@ -264,6 +265,7 @@ public class ResourcesReconcilerImpl implements ResourcesReconciler {
     logger.error(msg + " {} {} {}",
       keyValue("id", resource.getUid()),
       keyValue("ingress.path", resource.getIngress().getPath()),
+      keyValue("ingress.host", resource.getIngress().getHost()),
       keyValue("topics", resource.getTopicsList()),
       cause);
   }
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 0f7983acf..5e5146c2b 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
@@ -15,6 +15,7 @@
  */
 package dev.knative.eventing.kafka.broker.receiver.impl;
 
+import com.google.common.base.Strings;
 import dev.knative.eventing.kafka.broker.contract.DataPlaneContract;
 import dev.knative.eventing.kafka.broker.core.AsyncCloseable;
 import dev.knative.eventing.kafka.broker.core.metrics.Metrics;
@@ -29,6 +30,9 @@ import io.cloudevents.jackson.JsonFormat;
 import io.cloudevents.kafka.CloudEventSerializer;
 import io.vertx.core.Future;
 import io.vertx.kafka.client.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -36,10 +40,6 @@ import java.util.Objects;
 import java.util.Properties;
 import java.util.function.Function;
 
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 /**
  * This class implements a store of {@link IngressProducer} that can be reconciled
  * using it as {@link IngressReconcilerListener} of a {@link dev.knative.eventing.kafka.broker.core.reconciler.ResourcesReconciler}.
@@ -61,6 +61,9 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
   // path -> IngressInfo
   // We use this map on the hot path to directly resolve the producer from the path
   private final Map<String, IngressProducerImpl> pathMapper;
+  // host -> IngressInfo
+  // We use this map on the root path to directly resolve the producer from the hostname
+  private final Map<String, IngressProducerImpl> hostMapper;
 
   public IngressProducerReconcilableStore(
     final AuthProvider authProvider,
@@ -78,10 +81,18 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     this.ingressInfos = new HashMap<>();
     this.producerReferences = new HashMap<>();
     this.pathMapper = new HashMap<>();
+    this.hostMapper = new HashMap<>();
   }
 
-  public IngressProducer resolve(String path) {
-    return pathMapper.get(path);
+  public IngressProducer resolve(String host, String path) {
+    // Ignore the host when there's a path given in the request.
+    // That means, we support these modes:
+    // - Request coming to "/path" --> path is used for matching
+    // - Request coming to "/" --> hostname is used for matching
+    if (pathMapper.get(path) != null) {
+      return pathMapper.get(path);
+    }
+    return hostMapper.get(host);
   }
 
   @Override
@@ -122,10 +133,20 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
         rc.getValue().getProducer(),
         resource,
         ingress.getPath(),
+        ingress.getHost(),
         producerProps
       );
 
-      this.pathMapper.put(ingress.getPath(), ingressInfo);
+      if(isRootPath(ingress.getPath()) && Strings.isNullOrEmpty(ingress.getHost())){
+        throw new IllegalArgumentException("Ingress path and host is blank. One of them should be defined. Resource UID: " + resource.getUid());
+      }
+
+      if(!isRootPath(ingress.getPath())){
+        this.pathMapper.put(ingress.getPath(), ingressInfo);
+      }
+      if (!Strings.isNullOrEmpty(ingress.getHost())){
+        this.hostMapper.put(ingress.getHost(), ingressInfo);
+      }
       this.ingressInfos.put(resource.getUid(), ingressInfo);
 
       return Future.succeededFuture();
@@ -162,12 +183,22 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
       return rc.getValue().close()
         .onSuccess(r -> {
           // Remove ingress info from the maps
-          this.pathMapper.remove(ingressInfo.getPath());
+          if(!isRootPath(ingressInfo.getPath())){
+            this.pathMapper.remove(ingressInfo.getPath());
+          }
+          if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+            this.hostMapper.remove(ingressInfo.getHost());
+          }
           this.ingressInfos.remove(resource.getUid());
         });
     }
     // Remove ingress info from the maps
-    this.pathMapper.remove(ingressInfo.getPath());
+    if(!isRootPath(ingressInfo.getPath())){
+      this.pathMapper.remove(ingressInfo.getPath());
+    }
+    if(!Strings.isNullOrEmpty(ingressInfo.getHost())){
+      this.hostMapper.remove(ingressInfo.getHost());
+    }
     this.ingressInfos.remove(resource.getUid());
 
     return Future.succeededFuture();
@@ -222,17 +253,20 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
     private final KafkaProducer<String, CloudEvent> producer;
     private final String topic;
     private final String path;
+    private final String host;
     private final Properties producerProperties;
     private final DataPlaneContract.Reference reference;
 
     IngressProducerImpl(final KafkaProducer<String, CloudEvent> producer,
                         final DataPlaneContract.Resource resource,
                         final String path,
+                        final String host,
                         final Properties producerProperties) {
       this.producer = producer;
       this.topic = resource.getTopics(0);
       this.reference = resource.getReference();
       this.path = path;
+      this.host = host;
       this.producerProperties = producerProperties;
     }
 
@@ -255,9 +289,16 @@ public class IngressProducerReconcilableStore implements IngressReconcilerListen
       return path;
     }
 
+    String getHost() {
+      return host;
+    }
+
     Properties getProducerProperties() {
       return producerProperties;
     }
   }
 
+  private boolean isRootPath(String path) {
+    return Strings.isNullOrEmpty(path) || path.equals("/");
+  }
 }
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 2e4410646..a8a056e07 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
@@ -121,7 +121,7 @@ public class ReceiverVerticle extends AbstractVerticle implements Handler<HttpSe
     final var requestContext = new RequestContext(request);
 
     // Look up for the ingress producer
-    IngressProducer producer = this.ingressProducerStore.resolve(request.path());
+    IngressProducer producer = this.ingressProducerStore.resolve(request.host(), request.path());
     if (producer == null) {
       request.response().setStatusCode(NOT_FOUND.code()).end();
       logger.warn("Resource not found {}", keyValue("path", request.path()));
diff --git a/proto/contract.proto b/proto/contract.proto
index e0bed4ff1..b1f69010d 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -156,19 +156,23 @@ enum ContentMode {
   STRUCTURED = 1;
 }
 
+// Ingress is the definition for HTTP ingress that is receiving the events
+// into the Knative Kafka component.
+//
+// path and host fields are used for identifying the targets. They are exclusive.
+// When a request comes with "/some-path", hostname will not be checked.
+// When a request comes with "/", only hostname matching will be done.
+// It is allowed to specify both path and host in ingress contract
+// to support both modes.
 message Ingress {
   // Optional content mode to use when pushing messages to Kafka
   ContentMode contentMode = 1;
 
-  // Ingress can both listen on a specific HTTP path
-  // or listen to the / path but match the Host header
-  oneof ingressType {
-    // path to listen for incoming events.
-    string path = 2;
+  // path to listen for incoming events.
+  string path = 2;
 
-    // host header to match
-    string host = 3;
-  }
+  // host header to match
+  string host = 3;
 }
 
 // Kubernetes resource reference.


Test patch:
diff --git a/control-plane/pkg/core/config/resource_test.go b/control-plane/pkg/core/config/resource_test.go
index 584093179..d9723d309 100644
--- a/control-plane/pkg/core/config/resource_test.go
+++ b/control-plane/pkg/core/config/resource_test.go
@@ -108,9 +108,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 							},
 						},
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 						BootstrapServers: "broker:9092",
@@ -122,9 +120,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 				Uid:    "1",
 				Topics: []string{"topic-name-1"},
 				Ingress: &contract.Ingress{
-					IngressType: &contract.Ingress_Path{
-						Path: "/broker-ns/broker-name",
-					},
+					Path:        "/broker-ns/broker-name",
 					ContentMode: contract.ContentMode_STRUCTURED,
 				},
 				Egresses: []*contract.Egress{
@@ -148,9 +144,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						Uid:    "2",
 						Topics: []string{"topic-name-1"},
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 						Egresses: []*contract.Egress{
@@ -184,9 +178,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						},
 						BootstrapServers: "broker:9092",
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 					},
@@ -215,9 +207,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						},
 						BootstrapServers: "broker:9092",
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 					},
@@ -243,9 +233,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 				},
 				BootstrapServers: "broker:9092,broker-2:9092",
 				Ingress: &contract.Ingress{
-					IngressType: &contract.Ingress_Path{
-						Path: "/broker-ns/broker-name",
-					},
+					Path:        "/broker-ns/broker-name",
 					ContentMode: contract.ContentMode_STRUCTURED,
 				},
 			},
@@ -269,9 +257,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						},
 						BootstrapServers: "broker:9092,broker-2:9092",
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 					},
@@ -300,9 +286,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						},
 						BootstrapServers: "broker:9092",
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 					},
@@ -326,9 +310,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 				},
 				BootstrapServers: "broker:9092",
 				Ingress: &contract.Ingress{
-					IngressType: &contract.Ingress_Path{
-						Path: "/broker-ns/broker-name",
-					},
+					Path:        "/broker-ns/broker-name",
 					ContentMode: contract.ContentMode_STRUCTURED,
 				},
 			},
@@ -353,9 +335,7 @@ func TestAddOrUpdateResourcesConfig(t *testing.T) {
 						},
 						BootstrapServers: "broker:9092",
 						Ingress: &contract.Ingress{
-							IngressType: &contract.Ingress_Path{
-								Path: "/broker-ns/broker-name",
-							},
+							Path:        "/broker-ns/broker-name",
 							ContentMode: contract.ContentMode_STRUCTURED,
 						},
 					},
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 7d3dc8aaa..7acf64112 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -141,7 +141,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -201,7 +201,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -263,7 +263,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -323,7 +323,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 							Reference:        BrokerReference(),
@@ -382,7 +382,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURLFrom(BrokerNamespace, ServiceName)},
 							Reference:        BrokerReference(),
@@ -485,7 +485,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 							Reference:        BrokerReference(),
@@ -542,7 +542,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -584,7 +584,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
@@ -612,7 +612,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
@@ -623,7 +623,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -709,7 +709,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							EgressConfig:     &contract.EgressConfig{DeadLetter: "http://www.my-sink.com/api"},
 							Reference:        BrokerReference(),
@@ -767,13 +767,13 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
 							Uid:          BrokerUUID,
 							Topics:       []string{BrokerTopic()},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 					},
@@ -796,13 +796,13 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -847,13 +847,13 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 1,
@@ -876,13 +876,13 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
 						},
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -931,12 +931,12 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							Uid:          "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:       []string{"my-existing-topic-a"},
 							EgressConfig: &contract.EgressConfig{DeadLetter: "http://www.my-sink.com"},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -1046,7 +1046,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 						},
@@ -1118,7 +1118,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							Auth: &contract.Resource_AuthSecret{
@@ -1296,7 +1296,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 									ConsumerGroup: TriggerUUID + "b",
 								},
 							},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 1,
@@ -1341,7 +1341,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 2,
@@ -1426,7 +1426,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
@@ -1492,7 +1492,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
@@ -1558,7 +1558,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
@@ -1621,7 +1621,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig: &contract.EgressConfig{
@@ -1675,7 +1675,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
@@ -1723,7 +1723,7 @@ func brokerReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              BrokerUUID,
 							Topics:           []string{BrokerTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:          &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        BrokerReference(),
 							EgressConfig:     &contract.EgressConfig{DeadLetter: ServiceURL},
@@ -1816,7 +1816,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 1,
@@ -1845,7 +1845,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 1,
@@ -1874,7 +1874,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 					Generation: 1,
@@ -1906,7 +1906,7 @@ func brokerFinalization(t *testing.T, format string, env config.Env) {
 						{
 							Uid:          BrokerUUID,
 							Topics:       []string{BrokerTopic()},
-							Ingress:      &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress:      &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							EgressConfig: &contract.EgressConfig{DeadLetter: ServiceURL},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/channel/channel_test.go b/control-plane/pkg/reconciler/channel/channel_test.go
index 21f8c9ced..fe3d05572 100644
--- a/control-plane/pkg/reconciler/channel/channel_test.go
+++ b/control-plane/pkg/reconciler/channel/channel_test.go
@@ -170,9 +170,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -239,9 +237,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -310,9 +306,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -382,9 +376,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{},
 						},
@@ -455,9 +447,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -530,9 +520,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -605,9 +593,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -683,9 +669,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -963,9 +947,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1046,9 +1028,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 							Egresses: []*contract.Egress{{
 								ConsumerGroup: "kafka." + ChannelNamespace + "." + ChannelName + "." + Subscription1UUID,
@@ -1120,9 +1100,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1190,9 +1168,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
index be8ce3afb..d4bc7c708 100644
--- a/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
+++ b/control-plane/pkg/reconciler/channel/v2/channelv2_test.go
@@ -153,9 +153,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -197,9 +195,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -256,9 +252,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -309,9 +303,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -381,9 +373,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -449,9 +439,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -517,9 +505,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -574,9 +560,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -673,9 +657,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -745,9 +727,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -831,9 +811,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -990,9 +968,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1072,9 +1048,7 @@ func TestReconcileKind(t *testing.T) {
 								},
 							},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1121,9 +1095,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
@@ -1173,9 +1145,7 @@ func TestReconcileKind(t *testing.T) {
 							BootstrapServers: ChannelBootstrapServers,
 							Reference:        ChannelReference(),
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(ChannelNamespace, ChannelName),
-								},
+								Path: receiver.Path(ChannelNamespace, ChannelName),
 							},
 						},
 					},
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index 4ea2cafc3..7ad5464ea 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -131,7 +131,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -188,7 +188,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Topics:           []string{SinkTopic()},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							Auth: &contract.Resource_AuthSecret{
 								AuthSecret: &contract.Reference{
 									Uuid:      SecretUUID,
@@ -257,9 +257,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 							Reference:        SinkReference(),
 							Ingress: &contract.Ingress{
 								ContentMode: contract.ContentMode_STRUCTURED,
-								IngressType: &contract.Ingress_Path{
-									Path: receiver.Path(SinkNamespace, SinkName),
-								},
+								Path:        receiver.Path(SinkNamespace, SinkName),
 							},
 						},
 					},
@@ -371,7 +369,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{"my-topic-1"},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: "kafka-broker:10000",
 							Reference:        SinkReference(),
 						},
@@ -478,7 +476,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -528,7 +526,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -570,7 +568,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:  []string{"my-existing-topic-a"},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(SinkNamespace, SinkName)},
 						},
 						{
 							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -592,7 +590,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:  []string{"my-existing-topic-a"},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(SinkNamespace, SinkName)},
 						},
 						{
 							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -601,7 +599,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -673,7 +671,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -742,7 +740,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:  []string{"my-existing-topic-a"},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(SinkNamespace, SinkName)},
 						},
 						{
 							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -751,7 +749,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -797,7 +795,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
 							Topics:  []string{"my-existing-topic-a"},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(SinkNamespace, SinkName)},
 						},
 						{
 							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
@@ -806,7 +804,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -866,7 +864,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -920,7 +918,7 @@ func sinkReconciliation(t *testing.T, format string, env config.Env) {
 						{
 							Uid:              SinkUUID,
 							Topics:           []string{SinkTopic()},
-							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, Path: receiver.Path(SinkNamespace, SinkName)},
 							BootstrapServers: bootstrapServers,
 							Reference:        SinkReference(),
 						},
@@ -992,10 +990,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1004,9 +999,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID,
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1023,9 +1016,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1050,9 +1041,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1061,9 +1050,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID,
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1080,9 +1067,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1107,9 +1092,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1118,9 +1101,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID,
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1137,9 +1118,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1164,9 +1143,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1175,9 +1152,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID,
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1194,9 +1169,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1226,9 +1199,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1237,9 +1208,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID,
 							Topics: []string{"topic-2"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
@@ -1265,9 +1234,7 @@ func sinkFinalization(t *testing.T, format string, env config.Env) {
 							Uid:    SinkUUID + "a",
 							Topics: []string{"topic"},
 							Ingress: &contract.Ingress{
-								IngressType: &contract.Ingress_Path{
-									Path: "path",
-								},
+								Path:        "path",
 								ContentMode: contract.ContentMode_STRUCTURED,
 							},
 							BootstrapServers: bootstrapServers,
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index a8471db73..43d5c984d 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -132,7 +132,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -151,7 +151,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -196,7 +196,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -215,7 +215,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -260,7 +260,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -279,7 +279,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -331,7 +331,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -350,7 +350,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -396,7 +396,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -415,7 +415,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -461,7 +461,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 						},
 					},
 				}, &env),
@@ -480,7 +480,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -533,7 +533,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "z",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -546,7 +546,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								EgressBuilder{
 									contract.Egress{
@@ -607,7 +607,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "z",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -620,7 +620,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								EgressBuilder{
 									contract.Egress{
@@ -936,7 +936,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								EgressBuilder{
 									contract.Egress{
@@ -1022,7 +1022,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1078,7 +1078,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1147,7 +1147,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								EgressBuilder{
 									contract.Egress{
@@ -1217,7 +1217,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1326,7 +1326,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1381,7 +1381,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -1445,7 +1445,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1500,7 +1500,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -1612,7 +1612,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1667,7 +1667,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -1736,7 +1736,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1805,7 +1805,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -1860,7 +1860,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -1916,7 +1916,7 @@ func triggerReconciliation(t *testing.T, format string, env config.Env, useNewFi
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2185,7 +2185,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2218,7 +2218,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2265,7 +2265,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/3",
@@ -2293,7 +2293,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2351,7 +2351,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2384,7 +2384,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -2431,7 +2431,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2464,7 +2464,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -2517,7 +2517,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2550,7 +2550,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -2583,7 +2583,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2630,7 +2630,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
@@ -2663,7 +2663,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID,
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   "http://example.com/1",
@@ -2691,7 +2691,7 @@ func triggerFinalizer(t *testing.T, format string, env config.Env) {
 						{
 							Uid:     BrokerUUID + "a",
 							Topics:  []string{BrokerTopic()},
-							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							Ingress: &contract.Ingress{Path: receiver.Path(BrokerNamespace, BrokerName)},
 							Egresses: []*contract.Egress{
 								{
 									Destination:   ServiceURL,
diff --git a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
index 81e2a2581..13d95cd75 100644
--- a/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
+++ b/data-plane/receiver/src/test/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStoreTest.java
@@ -27,15 +27,17 @@ import io.vertx.junit5.VertxTestContext;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
-import org.apache.kafka.clients.producer.MockProducer;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
@@ -321,4 +323,136 @@ public class IngressProducerReconcilableStoreTest {
     when(producer.unwrap()).thenReturn(new MockProducer<>());
     return producer;
   }
+
+  @Test
+  public void shouldResolvePathCorrectly(final Vertx vertx,
+                                         final VertxTestContext context) {
+    final var resource1 = DataPlaneContract.Resource.newBuilder()
+      .setUid("1")
+      .addTopics("topic1")
+      .setBootstrapServers("kafka-1:9092")
+      .setIngress(DataPlaneContract.Ingress.newBuilder().setPath("/hello1")
+        .setContentMode(DataPlaneContract.ContentMode.STRUCTURED).build())
+      .build();
+
+    final var resource2 = DataPlaneContract.Resource.newBuilder()
+      .setUid("2")
+      .addTopics("topic2")
+      .setBootstrapServers("kafka-2:9092")
+      .setIngress(DataPlaneContract.Ingress.newBuilder().setHost("http://host2").setPath("/")
+        .setContentMode(DataPlaneContract.ContentMode.STRUCTURED).build())
+      .build();
+
+    final var resource3 = DataPlaneContract.Resource.newBuilder()
+      .setUid("3")
+      .addTopics("topic3")
+      .setBootstrapServers("kafka-3:9092")
+      .setIngress(DataPlaneContract.Ingress.newBuilder().setHost("http://host3").setPath("/hello3")
+        .setContentMode(DataPlaneContract.ContentMode.STRUCTURED).build())
+      .build();
+
+    // additional resource to make sure path "/" don't mean anything
+    final var resource4 = DataPlaneContract.Resource.newBuilder()
+      .setUid("4")
+      .addTopics("topic4")
+      .setBootstrapServers("kafka-4:9092")
+      .setIngress(DataPlaneContract.Ingress.newBuilder().setHost("http://host4").setPath("/")
+        .setContentMode(DataPlaneContract.ContentMode.STRUCTURED).build())
+      .build();
+
+    KafkaProducer<String, CloudEvent> producer1 = mockProducer();
+    KafkaProducer<String, CloudEvent> producer2 = mockProducer();
+    KafkaProducer<String, CloudEvent> producer3 = mockProducer();
+    KafkaProducer<String, CloudEvent> producer4 = mockProducer();
+
+    Map<String, KafkaProducer<String, CloudEvent>> producerMap = Map.of(
+      "kafka-1:9092", producer1,
+      "kafka-2:9092", producer2,
+      "kafka-3:9092", producer3,
+      "kafka-4:9092", producer4
+    );
+
+    final var store = new IngressProducerReconcilableStore(
+      AuthProvider.noAuth(),
+      new Properties(),
+      properties -> {
+        KafkaProducer<String, CloudEvent> producer = producerMap.get(properties.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
+        if (producer == null) {
+          throw new IllegalStateException("Can't determine what producer to return");
+        }
+        return producer;
+      }
+    );
+
+    final var reconciler = ResourcesReconciler
+      .builder()
+      .watchIngress(store)
+      .build();
+
+    vertx.runOnContext(v -> {
+      Future.succeededFuture()
+        .compose(v1 -> reconciler.reconcile(List.of(resource1, resource2, resource3, resource4)))
+        .onSuccess(event -> {
+          // match by path
+          assertThat(store.resolve("", "/hello1").getKafkaProducer()).isSameAs(producer1);
+          assertThat(store.resolve("", "/hello3").getKafkaProducer()).isSameAs(producer3);
+
+          // match by host
+          assertThat(store.resolve("http://host2", "/").getKafkaProducer()).isSameAs(producer2);
+          assertThat(store.resolve("http://host4", "/").getKafkaProducer()).isSameAs(producer4);
+          assertThat(store.resolve("http://host2", "").getKafkaProducer()).isSameAs(producer2);
+          assertThat(store.resolve("http://host4", "").getKafkaProducer()).isSameAs(producer4);
+
+
+          // only use path when the path is registered
+          assertThat(store.resolve("http://host1", "/hello1").getKafkaProducer()).isSameAs(producer1);
+          assertThat(store.resolve("http://host2", "/hello1").getKafkaProducer()).isSameAs(producer1);
+          assertThat(store.resolve("http://host1", "/hello3").getKafkaProducer()).isSameAs(producer3);
+          assertThat(store.resolve("http://host2", "/hello3").getKafkaProducer()).isSameAs(producer3);
+          assertThat(store.resolve("http://host3", "/hello3").getKafkaProducer()).isSameAs(producer3);
+
+          // match the host when the path is not registered
+          assertThat(store.resolve("http://host2", "/doesntExist").getKafkaProducer()).isSameAs(producer2);
+          assertThat(store.resolve("http://host3", "/doesntExist").getKafkaProducer()).isSameAs(producer3);
+          assertThat(store.resolve("http://host4", "/doesntExist").getKafkaProducer()).isSameAs(producer4);
+
+          // don't return anything when there's nothing matching
+          assertThat(store.resolve("http://unknown", "/")).isNull();
+          assertThat(store.resolve("http://unknown", "")).isNull();
+          assertThat(store.resolve("", "/doesntExist")).isNull();
+          assertThat(store.resolve("http://unknown", "/doesntExist")).isNull();
+
+          context.completeNow();
+        })
+        .onFailure(context::failNow);
+    });
+  }
+
+  @Test
+  public void shouldRejectIngressWithRootPathAndEmptyHost(final Vertx vertx, final VertxTestContext context) {
+    final var resource = DataPlaneContract.Resource.newBuilder()
+      .setUid("1")
+      .addTopics("topic")
+      .setBootstrapServers("kafka-1:9092,kafka-2:9092")
+      .setIngress(DataPlaneContract.Ingress.newBuilder().setHost("").setPath("/").build())
+      .build();
+
+    final var store = new IngressProducerReconcilableStore(
+      AuthProvider.noAuth(),
+      new Properties(),
+      properties -> mockProducer()
+    );
+
+    final var reconciler = ResourcesReconciler
+      .builder()
+      .watchIngress(store)
+      .build();
+
+    vertx.runOnContext(v -> {
+      Future.succeededFuture()
+        .compose(v1 -> reconciler.reconcile(List.of(resource)))
+        .onSuccess(event -> context.failNow("Reconcile should've failed"))
+        .onFailure(event -> context.completeNow());
+    });
+  }
 }