Repository: knative-sandbox/eventing-kafka-broker
Pull request #1301 (2021-10-11 17:23:09)

Bump maven version to the latest release
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

-
-
-

<!--
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

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/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/1301?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    #1301      +/-   ##
============================================
+ Coverage     75.70%   79.98%   +4.28%     
+ Complexity      460      459       -1     
============================================
  Files            87       58      -29     
  Lines          3058     1654    -1404     
  Branches        136      136              
============================================
- Hits           2315     1323     -992     
+ Misses          579      261     -318     
+ Partials        164       70      -94     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `79.98% <ø> (-0.07%)` | :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/1301?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...dispatcher/impl/consumer/BaseConsumerVerticle.java](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-ZGF0YS1wbGFuZS9kaXNwYXRjaGVyL3NyYy9tYWluL2phdmEvZGV2L2tuYXRpdmUvZXZlbnRpbmcva2Fma2EvYnJva2VyL2Rpc3BhdGNoZXIvaW1wbC9jb25zdW1lci9CYXNlQ29uc3VtZXJWZXJ0aWNsZS5qYXZh) | `89.28% <0.00%> (-3.58%)` | :arrow_down: |
| [control-plane/pkg/core/config/egress.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29yZS9jb25maWcvZWdyZXNzLmdv) | | |
| [...ntrol-plane/pkg/reconciler/broker/broker\_config.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyX2NvbmZpZy5nbw==) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| [control-plane/pkg/security/scram.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2NyYW0uZ28=) | | |
| [control-plane/pkg/reconciler/kafka/topic.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9rYWZrYS90b3BpYy5nbw==) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| [...-plane/pkg/reconciler/trigger/trigger\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXJfbGlmZWN5Y2xlLmdv) | | |
| [...lane/pkg/reconciler/base/receiver\_condition\_set.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9iYXNlL3JlY2VpdmVyX2NvbmRpdGlvbl9zZXQuZ28=) | | |
| [...kg/apis/eventing/v1alpha1/kafka\_sink\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX3ZhbGlkYXRpb24uZ28=) | | |
| ... and [20 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301/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/1301?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/1301?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [cbd8c28...4140979](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/1301?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 5392d06f0..b2c244be9 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -40,6 +40,7 @@ const (
 var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"): &eventingv1alpha1.KafkaSink{},
+	// sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"): &sourcesv1beta1.KafkaSource{},
 }
 
 var callbacks = map[schema.GroupVersionKind]validation.Callback{}
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 807da7381..0b3a75095 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -358,6 +358,7 @@ type Egress struct {
 	// Types that are assignable to ReplyStrategy:
 	//	*Egress_ReplyUrl
 	//	*Egress_ReplyToOriginalTopic
+	//	*Egress_DiscardReply
 	ReplyStrategy isEgress_ReplyStrategy `protobuf_oneof:"replyStrategy"`
 	Filter        *Filter                `protobuf:"bytes,5,opt,name=filter,proto3" json:"filter,omitempty"`
 	// Id of the egress
@@ -438,6 +439,13 @@ func (x *Egress) GetReplyToOriginalTopic() *Empty {
 	return nil
 }
 
+func (x *Egress) GetDiscardReply() *Empty {
+	if x, ok := x.GetReplyStrategy().(*Egress_DiscardReply); ok {
+		return x.DiscardReply
+	}
+	return nil
+}
+
 func (x *Egress) GetFilter() *Filter {
 	if x != nil {
 		return x.Filter
@@ -480,10 +488,17 @@ type Egress_ReplyToOriginalTopic struct {
 	ReplyToOriginalTopic *Empty `protobuf:"bytes,4,opt,name=replyToOriginalTopic,proto3,oneof"`
 }
 
+type Egress_DiscardReply struct {
+	// Discard response.
+	DiscardReply *Empty `protobuf:"bytes,9,opt,name=discardReply,proto3,oneof"`
+}
+
 func (*Egress_ReplyUrl) isEgress_ReplyStrategy() {}
 
 func (*Egress_ReplyToOriginalTopic) isEgress_ReplyStrategy() {}
 
+func (*Egress_DiscardReply) isEgress_ReplyStrategy() {}
+
 type Ingress struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -906,7 +921,7 @@ var file_contract_proto_rawDesc = []byte{
 	0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x04, 0x52,
 	0x0c, 0x62, 0x61, 0x63, 0x6b, 0x6f, 0x66, 0x66, 0x44, 0x65, 0x6c, 0x61, 0x79, 0x12, 0x18, 0x0a,
 	0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x05, 0x20, 0x01, 0x28, 0x04, 0x52, 0x07,
-	0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xd9, 0x02, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65,
+	0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0x87, 0x03, 0x0a, 0x06, 0x45, 0x67, 0x72, 0x65,
 	0x73, 0x73, 0x12, 0x24, 0x0a, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x47, 0x72,
 	0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x63, 0x6f, 0x6e, 0x73, 0x75,
 	0x6d, 0x65, 0x72, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x74,
@@ -917,72 +932,75 @@ var file_contract_proto_rawDesc = []byte{
 	0x79, 0x54, 0x6f, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61, 0x6c, 0x54, 0x6f, 0x70, 0x69, 0x63,
 	0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x48, 0x00,
 	0x52, 0x14, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x54, 0x6f, 0x4f, 0x72, 0x69, 0x67, 0x69, 0x6e, 0x61,
-	0x6c, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x1f, 0x0a, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72,
-	0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x52,
-	0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x06,
-	0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72,
-	0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 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, 0x34, 0x0a, 0x0d,
-	0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x18, 0x08, 0x20,
-	0x01, 0x28, 0x0e, 0x32, 0x0e, 0x2e, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72,
-	0x64, 0x65, 0x72, 0x52, 0x0d, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64,
-	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,
-	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, 0xbc, 0x02, 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, 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, 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,
-	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,
+	0x6c, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x12, 0x2c, 0x0a, 0x0c, 0x64, 0x69, 0x73, 0x63, 0x61, 0x72,
+	0x64, 0x52, 0x65, 0x70, 0x6c, 0x79, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x06, 0x2e, 0x45,
+	0x6d, 0x70, 0x74, 0x79, 0x48, 0x00, 0x52, 0x0c, 0x64, 0x69, 0x73, 0x63, 0x61, 0x72, 0x64, 0x52,
+	0x65, 0x70, 0x6c, 0x79, 0x12, 0x1f, 0x0a, 0x06, 0x66, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x18, 0x05,
+	0x20, 0x01, 0x28, 0x0b, 0x32, 0x07, 0x2e, 0x46, 0x69, 0x6c, 0x74, 0x65, 0x72, 0x52, 0x06, 0x66,
+	0x69, 0x6c, 0x74, 0x65, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x69, 0x64, 0x18, 0x06, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x03, 0x75, 0x69, 0x64, 0x12, 0x31, 0x0a, 0x0c, 0x65, 0x67, 0x72, 0x65, 0x73,
+	0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 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, 0x34, 0x0a, 0x0d, 0x64, 0x65,
+	0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65, 0x72, 0x18, 0x08, 0x20, 0x01, 0x28,
+	0x0e, 0x32, 0x0e, 0x2e, 0x44, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 0x65,
+	0x72, 0x52, 0x0d, 0x64, 0x65, 0x6c, 0x69, 0x76, 0x65, 0x72, 0x79, 0x4f, 0x72, 0x64, 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, 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, 0xbc, 0x02, 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, 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, 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, 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 (
@@ -1017,21 +1035,22 @@ var file_contract_proto_depIdxs = []int32{
 	11, // 0: Filter.attributes:type_name -> Filter.AttributesEntry
 	0,  // 1: EgressConfig.backoffPolicy:type_name -> BackoffPolicy
 	3,  // 2: Egress.replyToOriginalTopic:type_name -> Empty
-	4,  // 3: Egress.filter:type_name -> Filter
-	5,  // 4: Egress.egressConfig:type_name -> EgressConfig
-	1,  // 5: Egress.deliveryOrder:type_name -> DeliveryOrder
-	2,  // 6: Ingress.contentMode:type_name -> ContentMode
-	7,  // 7: Resource.ingress:type_name -> Ingress
-	5,  // 8: Resource.egressConfig:type_name -> EgressConfig
-	6,  // 9: Resource.egresses:type_name -> Egress
-	3,  // 10: Resource.absentAuth:type_name -> Empty
-	8,  // 11: Resource.authSecret:type_name -> Reference
-	9,  // 12: Contract.resources:type_name -> Resource
-	13, // [13:13] is the sub-list for method output_type
-	13, // [13:13] is the sub-list for method input_type
-	13, // [13:13] is the sub-list for extension type_name
-	13, // [13:13] is the sub-list for extension extendee
-	0,  // [0:13] is the sub-list for field type_name
+	3,  // 3: Egress.discardReply:type_name -> Empty
+	4,  // 4: Egress.filter:type_name -> Filter
+	5,  // 5: Egress.egressConfig:type_name -> EgressConfig
+	1,  // 6: Egress.deliveryOrder:type_name -> DeliveryOrder
+	2,  // 7: Ingress.contentMode:type_name -> ContentMode
+	7,  // 8: Resource.ingress:type_name -> Ingress
+	5,  // 9: Resource.egressConfig:type_name -> EgressConfig
+	6,  // 10: Resource.egresses:type_name -> Egress
+	3,  // 11: Resource.absentAuth:type_name -> Empty
+	8,  // 12: Resource.authSecret:type_name -> Reference
+	9,  // 13: Contract.resources:type_name -> Resource
+	14, // [14:14] is the sub-list for method output_type
+	14, // [14:14] is the sub-list for method input_type
+	14, // [14:14] is the sub-list for extension type_name
+	14, // [14:14] is the sub-list for extension extendee
+	0,  // [0:14] is the sub-list for field type_name
 }
 
 func init() { file_contract_proto_init() }
@@ -1140,6 +1159,7 @@ func file_contract_proto_init() {
 	file_contract_proto_msgTypes[3].OneofWrappers = []interface{}{
 		(*Egress_ReplyUrl)(nil),
 		(*Egress_ReplyToOriginalTopic)(nil),
+		(*Egress_DiscardReply)(nil),
 	}
 	file_contract_proto_msgTypes[4].OneofWrappers = []interface{}{
 		(*Ingress_Path)(nil),
diff --git a/control-plane/pkg/reconciler/base/receiver_condition_set.go b/control-plane/pkg/reconciler/base/receiver_condition_set.go
index ec4f43372..63f78dc3c 100644
--- a/control-plane/pkg/reconciler/base/receiver_condition_set.go
+++ b/control-plane/pkg/reconciler/base/receiver_condition_set.go
@@ -39,7 +39,7 @@ const (
 	ConditionConfigParsed       apis.ConditionType = "ConfigParsed"
 )
 
-var ConditionSet = apis.NewLivingConditionSet(
+var IngressConditionSet = apis.NewLivingConditionSet(
 	ConditionAddressable,
 	ConditionDataPlaneAvailable,
 	ConditionTopicReady,
@@ -47,6 +47,13 @@ var ConditionSet = apis.NewLivingConditionSet(
 	ConditionConfigParsed,
 )
 
+var EgressConditionSet = apis.NewLivingConditionSet(
+	ConditionDataPlaneAvailable,
+	ConditionTopicReady,
+	ConditionConfigMapUpdated,
+	ConditionConfigParsed,
+)
+
 const (
 	TopicOwnerAnnotation = "eventing.knative.dev/topic.owner"
 
@@ -170,14 +177,16 @@ func (manager *StatusConditionManager) TopicReady(topic string) {
 
 func (manager *StatusConditionManager) Reconciled() reconciler.Event {
 
-	object := manager.Object
+	if manager.SetAddress != nil {
+		object := manager.Object
 
-	manager.SetAddress(&apis.URL{
-		Scheme: "http",
-		Host:   network.GetServiceHostname(manager.Configs.IngressName, manager.Configs.SystemNamespace),
-		Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
-	})
-	object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
+		manager.SetAddress(&apis.URL{
+			Scheme: "http",
+			Host:   network.GetServiceHostname(manager.Configs.IngressName, manager.Configs.SystemNamespace),
+			Path:   fmt.Sprintf("/%s/%s", object.GetNamespace(), object.GetName()),
+		})
+		object.GetConditionSet().Manage(object.GetStatus()).MarkTrue(ConditionAddressable)
+	}
 
 	return nil
 }
diff --git a/control-plane/pkg/reconciler/broker/controller.go b/control-plane/pkg/reconciler/broker/controller.go
index 744ce4043..0ae56b6ae 100644
--- a/control-plane/pkg/reconciler/broker/controller.go
+++ b/control-plane/pkg/reconciler/broker/controller.go
@@ -55,7 +55,7 @@ const (
 
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *Configs) *controller.Impl {
 
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/channel/controller.go b/control-plane/pkg/reconciler/channel/controller.go
index d8e41b1dc..650f93c52 100644
--- a/control-plane/pkg/reconciler/channel/controller.go
+++ b/control-plane/pkg/reconciler/channel/controller.go
@@ -44,7 +44,7 @@ import (
 
 func NewController(ctx context.Context, watcher configmap.Watcher, configs *config.Env) *controller.Impl {
 
-	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.ConditionSet)
+	messagingv1beta.RegisterAlternateKafkaChannelConditionSet(base.IngressConditionSet)
 
 	configmapInformer := configmapinformer.Get(ctx)
 
diff --git a/control-plane/pkg/reconciler/sink/controller.go b/control-plane/pkg/reconciler/sink/controller.go
index aa97b8145..19478b0c5 100644
--- a/control-plane/pkg/reconciler/sink/controller.go
+++ b/control-plane/pkg/reconciler/sink/controller.go
@@ -41,7 +41,7 @@ import (
 
 func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
 
-	eventing.RegisterConditionSet(base.ConditionSet)
+	eventing.RegisterConditionSet(base.IngressConditionSet)
 
 	logger := logging.FromContext(ctx)
 
diff --git a/control-plane/pkg/reconciler/source/auth_provider.go b/control-plane/pkg/reconciler/source/auth_provider.go
new file mode 100644
index 000000000..a677d63c6
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/auth_provider.go
@@ -0,0 +1,50 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+type SecretLocator struct {
+	*sources.KafkaSource
+}
+
+func (ks *SecretLocator) SecretName() (string, bool) {
+	if !hasAuthConfig(ks.KafkaSource) {
+		return "", false
+	}
+	return getName(ks.KafkaSource), hasAuthConfig(ks.KafkaSource)
+}
+
+func (ks *SecretLocator) SecretNamespace() (string, bool) {
+	return ks.Namespace, hasAuthConfig(ks.KafkaSource)
+}
+
+func getName(ks *sources.KafkaSource) string {
+	return ""
+}
+
+func hasAuthConfig(ks *sources.KafkaSource) bool {
+	// TODO handle secrets
+	//return (ks.Spec.KafkaAuthSpec.Net.TLS.Enable || ks.Spec.KafkaAuthSpec.Net.SASL.Enable) &&
+	//	(ks.Spec.KafkaAuthSpec.Net.TLS.Cert.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.User.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.Password.SecretKeyRef != nil ||
+	//		ks.Spec.KafkaAuthSpec.Net.SASL.Type.SecretKeyRef != nil)
+	return false
+}
diff --git a/control-plane/pkg/reconciler/source/controller.go b/control-plane/pkg/reconciler/source/controller.go
new file mode 100644
index 000000000..8f06ce8b8
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/controller.go
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"context"
+
+	"github.com/Shopify/sarama"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	podinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/pod"
+	secretinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/secret"
+	"knative.dev/pkg/configmap"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/resolver"
+
+	kafkainformer "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	"knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+)
+
+func NewController(ctx context.Context, _ configmap.Watcher, configs *config.Env) *controller.Impl {
+
+	sources.RegisterAlternateKafkaConditionSet(base.EgressConditionSet)
+
+	r := &Reconciler{
+		Reconciler: &base.Reconciler{
+			KubeClient:                  kubeclient.Get(ctx),
+			PodLister:                   podinformer.Get(ctx).Lister(),
+			SecretLister:                secretinformer.Get(ctx).Lister(),
+			DataPlaneConfigMapNamespace: configs.DataPlaneConfigMapNamespace,
+			DataPlaneConfigMapName:      configs.DataPlaneConfigMapName,
+			DataPlaneConfigFormat:       configs.DataPlaneConfigFormat,
+			SystemNamespace:             configs.SystemNamespace,
+			DispatcherLabel:             base.BrokerDispatcherLabel,
+			ReceiverLabel:               base.BrokerReceiverLabel,
+		},
+		Env:          configs,
+		ClusterAdmin: sarama.NewClusterAdmin,
+	}
+
+	impl := kafkasource.NewImpl(ctx, r)
+
+	r.Resolver = resolver.NewURIResolverFromTracker(ctx, impl.Tracker)
+
+	kafkaInformer := kafkainformer.Get(ctx)
+
+	kafkaInformer.Informer().AddEventHandler(controller.HandleAll(impl.Enqueue))
+
+	return impl
+}
diff --git a/control-plane/pkg/reconciler/source/source.go b/control-plane/pkg/reconciler/source/source.go
new file mode 100644
index 000000000..49d15af90
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/source.go
@@ -0,0 +1,193 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	"knative.dev/pkg/controller"
+	"knative.dev/pkg/reconciler"
+	"knative.dev/pkg/resolver"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
+	coreconfig "knative.dev/eventing-kafka-broker/control-plane/pkg/core/config"
+	kafkalogging "knative.dev/eventing-kafka-broker/control-plane/pkg/logging"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/base"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/reconciler/kafka"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/security"
+)
+
+type Reconciler struct {
+	*base.Reconciler
+	*config.Env
+
+	Resolver *resolver.URIResolver
+
+	// ClusterAdmin creates new sarama ClusterAdmin. It's convenient to add this as Reconciler field so that we can
+	// mock the function used during the reconciliation loop.
+	ClusterAdmin kafka.NewClusterAdminFunc
+}
+
+func (r *Reconciler) ReconcileKind(ctx context.Context, ks *sources.KafkaSource) reconciler.Event {
+
+	logger := kafkalogging.CreateReconcileMethodLogger(ctx, ks)
+
+	statusConditionManager := base.StatusConditionManager{
+		Object:   ks,
+		Configs:  r.Env,
+		Recorder: controller.GetEventRecorder(ctx),
+	}
+
+	secret, err := security.Secret(ctx, &SecretLocator{KafkaSource: ks}, r.SecretProviderFunc())
+	if err != nil {
+		return fmt.Errorf("failed to get secret: %w", err)
+	}
+	if secret != nil {
+		logger.Debug("Secret reference",
+			zap.String("apiVersion", secret.APIVersion),
+			zap.String("name", secret.Name),
+			zap.String("namespace", secret.Namespace),
+			zap.String("kind", secret.Kind),
+		)
+	}
+
+	// get security option for Sarama with secret info in it
+	securityOption := security.NewSaramaSecurityOptionFromSecret(secret)
+
+	if err := r.TrackSecret(secret, ks); err != nil {
+		return fmt.Errorf("failed to track secret: %w", err)
+	}
+
+	for _, t := range ks.Spec.Topics {
+		_, err := r.ClusterAdmin.IsTopicPresentAndValid(t, ks.Spec.BootstrapServers, securityOption)
+		if err != nil {
+			return fmt.Errorf("failed to verify topic validity: %w", err)
+		}
+	}
+
+	// Get contract config map.
+	contractConfigMap, err := r.GetOrCreateDataPlaneConfigMap(ctx)
+	if err != nil {
+		return statusConditionManager.FailedToGetConfigMap(err)
+	}
+
+	logger.Debug("Got contract config map")
+
+	// Get contract data.
+	ct, err := r.GetDataPlaneConfigMapData(logger, contractConfigMap)
+	if err != nil && ct == nil {
+		return statusConditionManager.FailedToGetDataFromConfigMap(err)
+	}
+
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
+
+	// Get resource configuration.
+	resource, err := r.getResource(ctx, ks, secret)
+	if err != nil {
+		return statusConditionManager.FailedToGetConfig(err)
+	}
+
+	brokerIndex := coreconfig.FindResource(ct, ks.GetUID())
+	// Update contract data with the new contract configuration
+	changed := coreconfig.AddOrUpdateResourceConfig(ct, resource, brokerIndex, logger)
+
+	logger.Debug("Change detector", zap.Int("changed", changed))
+
+	if changed == coreconfig.ResourceChanged {
+		// Resource changed, increment contract generation.
+		coreconfig.IncrementContractGeneration(ct)
+
+		// Update the configuration map with the new contract data.
+		if err := r.UpdateDataPlaneConfigMap(ctx, ct, contractConfigMap); err != nil {
+			logger.Error("failed to update data plane config map", zap.Error(
+				statusConditionManager.FailedToUpdateConfigMap(err),
+			))
+			return err
+		}
+		logger.Debug("Contract config map updated")
+	}
+	statusConditionManager.ConfigMapUpdated()
+
+	// We update dispatcher pods annotation regardless of our contract changed or not due to the fact
+	// that in a previous reconciliation we might have failed to update one of our data plane pod annotation, so we want
+	// to anyway update remaining annotations with the contract generation that was saved in the CM.
+
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		// Failing to update dispatcher pods annotation leads to config map refresh delayed by several seconds.
+		// Since the dispatcher side is the consumer side, we don't lose availability, and we can consider the Broker
+		// ready. So, log out the error and move on to the next step.
+		logger.Warn(
+			"Failed to update dispatcher pod annotation to trigger an immediate config map refresh",
+			zap.Error(err),
+		)
+
+		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
+	}
+
+	return statusConditionManager.Reconciled()
+}
+
+func (r *Reconciler) getResource(ctx context.Context, ks *sources.KafkaSource, secret *corev1.Secret) (*contract.Resource, error) {
+	destination, err := r.Resolver.URIFromDestinationV1(ctx, ks.Spec.Sink, ks)
+	if err != nil {
+		return nil, fmt.Errorf("failed to resolve destination: %w", err)
+	}
+
+	egress := &contract.Egress{
+		ConsumerGroup: ks.Spec.ConsumerGroup,
+		Destination:   destination.String(),
+		Uid:           string(ks.GetUID()),
+		EgressConfig: &contract.EgressConfig{
+			// TODO hardcoded retry num
+			Retry:         10,
+			BackoffPolicy: contract.BackoffPolicy_Exponential,
+			// TODO hardcoded backoff delay
+			BackoffDelay: 10000, // 10 seconds
+			// TODO set timeout
+			Timeout: 0,
+		},
+		// TODO delivery ordering
+		DeliveryOrder: contract.DeliveryOrder_ORDERED,
+	}
+	resource := &contract.Resource{
+		Uid:              string(ks.GetUID()),
+		Topics:           ks.Spec.Topics,
+		BootstrapServers: strings.Join(ks.Spec.BootstrapServers, ","),
+		Egresses:         []*contract.Egress{egress},
+		Auth:             &contract.Resource_AbsentAuth{},
+	}
+	if secret != nil {
+		resource.Auth = &contract.Resource_AuthSecret{
+			AuthSecret: &contract.Reference{
+				Uuid:      string(secret.GetUID()),
+				Namespace: secret.GetNamespace(),
+				Name:      secret.GetName(),
+				Version:   secret.GetResourceVersion(),
+			},
+		}
+	}
+	return resource, nil
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger.go b/control-plane/pkg/reconciler/trigger/trigger.go
index cdb526178..4915a2bfb 100644
--- a/control-plane/pkg/reconciler/trigger/trigger.go
+++ b/control-plane/pkg/reconciler/trigger/trigger.go
@@ -92,7 +92,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, trigger *eventing.Trigge
 	}
 
 	// Ignore Triggers that are associated with a Broker we don't own.
-	if isOur, brokerClass := isOurBroker(broker); !isOur {
+	if isKnativeKafkaBroker, brokerClass := isKnativeKafkaBroker(broker); !isKnativeKafkaBroker {
 		logger.Debug("Ignoring Trigger", zap.String(eventing.BrokerClassAnnotationKey, brokerClass))
 		return nil
 	}
@@ -317,7 +317,7 @@ func deleteTrigger(egresses []*contract.Egress, index int) []*contract.Egress {
 	return egresses[:len(egresses)-1]
 }
 
-func isOurBroker(broker *eventing.Broker) (bool, string) {
+func isKnativeKafkaBroker(broker *eventing.Broker) (bool, string) {
 	brokerClass := broker.GetAnnotations()[eventing.BrokerClassAnnotationKey]
 	return brokerClass == kafka.BrokerClass, brokerClass
 }
diff --git a/data-plane/.mvn/wrapper/maven-wrapper.properties b/data-plane/.mvn/wrapper/maven-wrapper.properties
index 642d572ce..a9f1ef87b 100644
--- a/data-plane/.mvn/wrapper/maven-wrapper.properties
+++ b/data-plane/.mvn/wrapper/maven-wrapper.properties
@@ -1,2 +1,2 @@
-distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.6.3/apache-maven-3.6.3-bin.zip
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.3/apache-maven-3.8.3-bin.zip
 wrapperUrl=https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar
diff --git a/data-plane/THIRD-PARTY.txt b/data-plane/THIRD-PARTY.txt
index df8f7275b..d8d6d7204 100644
--- a/data-plane/THIRD-PARTY.txt
+++ b/data-plane/THIRD-PARTY.txt
@@ -1,5 +1,5 @@
 
-Lists of 155 third-party dependencies.
+Lists of 151 third-party dependencies.
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Classic Module (ch.qos.logback:logback-classic:1.2.6 - http://logback.qos.ch/logback-classic)
      (Eclipse Public License - v 1.0) (GNU Lesser General Public License) Logback Core Module (ch.qos.logback:logback-core:1.2.6 - http://logback.qos.ch/logback-core)
      (The Apache Software License, Version 2.0) Jackson-annotations (com.fasterxml.jackson.core:jackson-annotations:2.11.4 - http://github.com/FasterXML/jackson)
@@ -96,10 +96,6 @@ Lists of 155 third-party dependencies.
      (The Apache License, Version 2.0) OpenTelemetry Java (io.opentelemetry:opentelemetry-semconv:1.6.0-alpha - https://github.com/open-telemetry/opentelemetry-java)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient (io.prometheus:simpleclient:0.10.0 - http://github.com/prometheus/client_java/simpleclient)
      (The Apache Software License, Version 2.0) Prometheus Java Simpleclient Common (io.prometheus:simpleclient_common:0.10.0 - http://github.com/prometheus/client_java/simpleclient_common)
-     (The Apache Software License, Version 2.0) Sundrio :: Annotations :: Builder (io.sundr:builder-annotations:0.13.1 - https://github.com/sundrio/sundrio/annotations/builder-annotations)
-     (The Apache Software License, Version 2.0) Sundrio :: Annotations :: Resourcecify (io.sundr:resourcecify-annotations:0.13.1 - https://github.com/sundrio/sundrio/annotations/resourcecify-annotations)
-     (The Apache Software License, Version 2.0) Sundrio :: Code generation (io.sundr:sundr-codegen:0.13.1 - https://github.com/sundrio/sundrio/sundr-codegen)
-     (The Apache Software License, Version 2.0) Sundrio :: Core (io.sundr:sundr-core:0.13.1 - https://github.com/sundrio/sundrio/sundr-core)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-auth-common (io.vertx:vertx-auth-common:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-auth/vertx-auth-common)
      (Eclipse Public License - v 1.0) (The Apache Software License, Version 2.0) vertx-circuit-breaker (io.vertx:vertx-circuit-breaker:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-ext/vertx-ext-parent/vertx-circuit-breaker)
      (Eclipse Public License - v 2.0) (The Apache Software License, Version 2.0) Vert.x Core (io.vertx:vertx-core:4.1.4 - http://nexus.sonatype.org/oss-repository-hosting.html/vertx-parent/vertx-core)
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 4670d1400..7a915d525 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
@@ -2795,6 +2795,33 @@ public final class DataPlaneContract {
      */
     dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder getReplyToOriginalTopicOrBuilder();
 
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     * @return Whether the discardReply field is set.
+     */
+    boolean hasDiscardReply();
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     * @return The discardReply.
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty getDiscardReply();
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     */
+    dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder getDiscardReplyOrBuilder();
+
     /**
      * <code>.Filter filter = 5;</code>
      * @return Whether the filter field is set.
@@ -3004,6 +3031,20 @@ public final class DataPlaneContract {
               deliveryOrder_ = rawValue;
               break;
             }
+            case 74: {
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder subBuilder = null;
+              if (replyStrategyCase_ == 9) {
+                subBuilder = ((dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_).toBuilder();
+              }
+              replyStrategy_ =
+                  input.readMessage(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.parser(), extensionRegistry);
+              if (subBuilder != null) {
+                subBuilder.mergeFrom((dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_);
+                replyStrategy_ = subBuilder.buildPartial();
+              }
+              replyStrategyCase_ = 9;
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -3043,6 +3084,7 @@ public final class DataPlaneContract {
             com.google.protobuf.AbstractMessage.InternalOneOfEnum {
       REPLYURL(3),
       REPLYTOORIGINALTOPIC(4),
+      DISCARDREPLY(9),
       REPLYSTRATEGY_NOT_SET(0);
       private final int value;
       private ReplyStrategyCase(int value) {
@@ -3062,6 +3104,7 @@ public final class DataPlaneContract {
         switch (value) {
           case 3: return REPLYURL;
           case 4: return REPLYTOORIGINALTOPIC;
+          case 9: return DISCARDREPLY;
           case 0: return REPLYSTRATEGY_NOT_SET;
           default: return null;
         }
@@ -3276,6 +3319,49 @@ public final class DataPlaneContract {
       return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
     }
 
+    public static final int DISCARDREPLY_FIELD_NUMBER = 9;
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     * @return Whether the discardReply field is set.
+     */
+    @java.lang.Override
+    public boolean hasDiscardReply() {
+      return replyStrategyCase_ == 9;
+    }
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     * @return The discardReply.
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty getDiscardReply() {
+      if (replyStrategyCase_ == 9) {
+         return (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_;
+      }
+      return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+    }
+    /**
+     * <pre>
+     * Discard response.
+     * </pre>
+     *
+     * <code>.Empty discardReply = 9;</code>
+     */
+    @java.lang.Override
+    public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder getDiscardReplyOrBuilder() {
+      if (replyStrategyCase_ == 9) {
+         return (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_;
+      }
+      return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+    }
+
     public static final int FILTER_FIELD_NUMBER = 5;
     private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Filter filter_;
     /**
@@ -3458,6 +3544,9 @@ public final class DataPlaneContract {
       if (deliveryOrder_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.DeliveryOrder.UNORDERED.getNumber()) {
         output.writeEnum(8, deliveryOrder_);
       }
+      if (replyStrategyCase_ == 9) {
+        output.writeMessage(9, (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -3495,6 +3584,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(8, deliveryOrder_);
       }
+      if (replyStrategyCase_ == 9) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(9, (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -3537,6 +3630,10 @@ public final class DataPlaneContract {
           if (!getReplyToOriginalTopic()
               .equals(other.getReplyToOriginalTopic())) return false;
           break;
+        case 9:
+          if (!getDiscardReply()
+              .equals(other.getDiscardReply())) return false;
+          break;
         case 0:
         default:
       }
@@ -3576,6 +3673,10 @@ public final class DataPlaneContract {
           hash = (37 * hash) + REPLYTOORIGINALTOPIC_FIELD_NUMBER;
           hash = (53 * hash) + getReplyToOriginalTopic().hashCode();
           break;
+        case 9:
+          hash = (37 * hash) + DISCARDREPLY_FIELD_NUMBER;
+          hash = (53 * hash) + getDiscardReply().hashCode();
+          break;
         case 0:
         default:
       }
@@ -3772,6 +3873,13 @@ public final class DataPlaneContract {
             result.replyStrategy_ = replyToOriginalTopicBuilder_.build();
           }
         }
+        if (replyStrategyCase_ == 9) {
+          if (discardReplyBuilder_ == null) {
+            result.replyStrategy_ = replyStrategy_;
+          } else {
+            result.replyStrategy_ = discardReplyBuilder_.build();
+          }
+        }
         if (filterBuilder_ == null) {
           result.filter_ = filter_;
         } else {
@@ -3865,6 +3973,10 @@ public final class DataPlaneContract {
             mergeReplyToOriginalTopic(other.getReplyToOriginalTopic());
             break;
           }
+          case DISCARDREPLY: {
+            mergeDiscardReply(other.getDiscardReply());
+            break;
+          }
           case REPLYSTRATEGY_NOT_SET: {
             break;
           }
@@ -4403,6 +4515,183 @@ public final class DataPlaneContract {
         return replyToOriginalTopicBuilder_;
       }
 
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder> discardReplyBuilder_;
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       * @return Whether the discardReply field is set.
+       */
+      @java.lang.Override
+      public boolean hasDiscardReply() {
+        return replyStrategyCase_ == 9;
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       * @return The discardReply.
+       */
+      @java.lang.Override
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty getDiscardReply() {
+        if (discardReplyBuilder_ == null) {
+          if (replyStrategyCase_ == 9) {
+            return (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_;
+          }
+          return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+        } else {
+          if (replyStrategyCase_ == 9) {
+            return discardReplyBuilder_.getMessage();
+          }
+          return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+        }
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      public Builder setDiscardReply(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty value) {
+        if (discardReplyBuilder_ == null) {
+          if (value == null) {
+            throw new NullPointerException();
+          }
+          replyStrategy_ = value;
+          onChanged();
+        } else {
+          discardReplyBuilder_.setMessage(value);
+        }
+        replyStrategyCase_ = 9;
+        return this;
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      public Builder setDiscardReply(
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder builderForValue) {
+        if (discardReplyBuilder_ == null) {
+          replyStrategy_ = builderForValue.build();
+          onChanged();
+        } else {
+          discardReplyBuilder_.setMessage(builderForValue.build());
+        }
+        replyStrategyCase_ = 9;
+        return this;
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      public Builder mergeDiscardReply(dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty value) {
+        if (discardReplyBuilder_ == null) {
+          if (replyStrategyCase_ == 9 &&
+              replyStrategy_ != dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance()) {
+            replyStrategy_ = dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.newBuilder((dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_)
+                .mergeFrom(value).buildPartial();
+          } else {
+            replyStrategy_ = value;
+          }
+          onChanged();
+        } else {
+          if (replyStrategyCase_ == 9) {
+            discardReplyBuilder_.mergeFrom(value);
+          }
+          discardReplyBuilder_.setMessage(value);
+        }
+        replyStrategyCase_ = 9;
+        return this;
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      public Builder clearDiscardReply() {
+        if (discardReplyBuilder_ == null) {
+          if (replyStrategyCase_ == 9) {
+            replyStrategyCase_ = 0;
+            replyStrategy_ = null;
+            onChanged();
+          }
+        } else {
+          if (replyStrategyCase_ == 9) {
+            replyStrategyCase_ = 0;
+            replyStrategy_ = null;
+          }
+          discardReplyBuilder_.clear();
+        }
+        return this;
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder getDiscardReplyBuilder() {
+        return getDiscardReplyFieldBuilder().getBuilder();
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      @java.lang.Override
+      public dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder getDiscardReplyOrBuilder() {
+        if ((replyStrategyCase_ == 9) && (discardReplyBuilder_ != null)) {
+          return discardReplyBuilder_.getMessageOrBuilder();
+        } else {
+          if (replyStrategyCase_ == 9) {
+            return (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_;
+          }
+          return dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+        }
+      }
+      /**
+       * <pre>
+       * Discard response.
+       * </pre>
+       *
+       * <code>.Empty discardReply = 9;</code>
+       */
+      private com.google.protobuf.SingleFieldBuilderV3<
+          dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder> 
+          getDiscardReplyFieldBuilder() {
+        if (discardReplyBuilder_ == null) {
+          if (!(replyStrategyCase_ == 9)) {
+            replyStrategy_ = dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.getDefaultInstance();
+          }
+          discardReplyBuilder_ = new com.google.protobuf.SingleFieldBuilderV3<
+              dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.EmptyOrBuilder>(
+                  (dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Empty) replyStrategy_,
+                  getParentForChildren(),
+                  isClean());
+          replyStrategy_ = null;
+        }
+        replyStrategyCase_ = 9;
+        onChanged();;
+        return discardReplyBuilder_;
+      }
+
       private dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Filter filter_;
       private com.google.protobuf.SingleFieldBuilderV3<
           dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Filter, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.Filter.Builder, dev.knative.eventing.kafka.broker.contract.DataPlaneContract.FilterOrBuilder> filterBuilder_;
@@ -11368,31 +11657,31 @@ public final class DataPlaneContract {
       "e\030\002 \001(\t:\0028\001\"\177\n\014EgressConfig\022\022\n\ndeadLette" +
       "r\030\001 \001(\t\022\r\n\005retry\030\002 \001(\r\022%\n\rbackoffPolicy\030" +
       "\003 \001(\0162\016.BackoffPolicy\022\024\n\014backoffDelay\030\004 " +
-      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\363\001\n\006Egress\022\025\n\rconsu" +
+      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\223\002\n\006Egress\022\025\n\rconsu" +
       "merGroup\030\001 \001(\t\022\023\n\013destination\030\002 \001(\t\022\022\n\010r" +
       "eplyUrl\030\003 \001(\tH\000\022&\n\024replyToOriginalTopic\030" +
-      "\004 \001(\0132\006.EmptyH\000\022\027\n\006filter\030\005 \001(\0132\007.Filter" +
-      "\022\013\n\003uid\030\006 \001(\t\022#\n\014egressConfig\030\007 \001(\0132\r.Eg" +
-      "ressConfig\022%\n\rdeliveryOrder\030\010 \001(\0162\016.Deli" +
-      "veryOrderB\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\"\344\001\n" +
-      "\010Resource\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022\030" +
-      "\n\020bootstrapServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001(" +
-      "\0132\010.Ingress\022#\n\014egressConfig\030\005 \001(\0132\r.Egre" +
-      "ssConfig\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\na" +
-      "bsentAuth\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret\030" +
-      "\010 \001(\0132\n.ReferenceH\000B\006\n\004Auth\"<\n\010Contract\022" +
-      "\022\n\ngeneration\030\001 \001(\004\022\034\n\tresources\030\002 \003(\0132\t" +
-      ".Resource*,\n\rBackoffPolicy\022\017\n\013Exponentia" +
-      "l\020\000\022\n\n\006Linear\020\001*+\n\rDeliveryOrder\022\r\n\tUNOR" +
-      "DERED\020\000\022\013\n\007ORDERED\020\001*)\n\013ContentMode\022\n\n\006B" +
-      "INARY\020\000\022\016\n\nSTRUCTURED\020\001B[\n*dev.knative.e" +
-      "venting.kafka.broker.contractB\021DataPlane" +
-      "ContractZ\032control-plane/pkg/contractb\006pr" +
-      "oto3"
+      "\004 \001(\0132\006.EmptyH\000\022\036\n\014discardReply\030\t \001(\0132\006." +
+      "EmptyH\000\022\027\n\006filter\030\005 \001(\0132\007.Filter\022\013\n\003uid\030" +
+      "\006 \001(\t\022#\n\014egressConfig\030\007 \001(\0132\r.EgressConf" +
+      "ig\022%\n\rdeliveryOrder\030\010 \001(\0162\016.DeliveryOrde" +
+      "rB\017\n\rreplyStrategy\"[\n\007Ingress\022!\n\013content" +
+      "Mode\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\tRefer" +
+      "ence\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\"\344\001\n\010Resourc" +
+      "e\022\013\n\003uid\030\001 \001(\t\022\016\n\006topics\030\002 \003(\t\022\030\n\020bootst" +
+      "rapServers\030\003 \001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingr" +
+      "ess\022#\n\014egressConfig\030\005 \001(\0132\r.EgressConfig" +
+      "\022\031\n\010egresses\030\006 \003(\0132\007.Egress\022\034\n\nabsentAut" +
+      "h\030\007 \001(\0132\006.EmptyH\000\022 \n\nauthSecret\030\010 \001(\0132\n." +
+      "ReferenceH\000B\006\n\004Auth\"<\n\010Contract\022\022\n\ngener" +
+      "ation\030\001 \001(\004\022\034\n\tresources\030\002 \003(\0132\t.Resourc" +
+      "e*,\n\rBackoffPolicy\022\017\n\013Exponential\020\000\022\n\n\006L" +
+      "inear\020\001*+\n\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022" +
+      "\013\n\007ORDERED\020\001*)\n\013ContentMode\022\n\n\006BINARY\020\000\022" +
+      "\016\n\nSTRUCTURED\020\001B[\n*dev.knative.eventing." +
+      "kafka.broker.contractB\021DataPlaneContract" +
+      "Z\032control-plane/pkg/contractb\006proto3"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -11427,7 +11716,7 @@ public final class DataPlaneContract {
     internal_static_Egress_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_Egress_descriptor,
-        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "ReplyStrategy", });
+        new java.lang.String[] { "ConsumerGroup", "Destination", "ReplyUrl", "ReplyToOriginalTopic", "DiscardReply", "Filter", "Uid", "EgressConfig", "DeliveryOrder", "ReplyStrategy", });
     internal_static_Ingress_descriptor =
       getDescriptor().getMessageTypes().get(4);
     internal_static_Ingress_fieldAccessorTable = new
diff --git a/data-plane/core/pom.xml b/data-plane/core/pom.xml
index 030e359f9..55c87a149 100644
--- a/data-plane/core/pom.xml
+++ b/data-plane/core/pom.xml
@@ -92,6 +92,10 @@
           <groupId>com.fasterxml.jackson.core</groupId>
           <artifactId>jackson-core</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>org.osgi</groupId>
+          <artifactId>org.osgi.core</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/NoopResponseHandler.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/NoopResponseHandler.java
new file mode 100644
index 000000000..9cf6df754
--- /dev/null
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/impl/NoopResponseHandler.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package dev.knative.eventing.kafka.broker.dispatcher.impl;
+
+import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
+import io.vertx.core.Future;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.ext.web.client.HttpResponse;
+
+/**
+ * {@link NoopResponseHandler} is a response handler that doesn't do anything with a given response.
+ * <p>
+ * KafkaSource should not be allowed to read and requeue responses.
+ */
+public class NoopResponseHandler implements ResponseHandler {
+
+  @Override
+  public Future<Void> handle(final HttpResponse<Buffer> response) {
+    return Future.succeededFuture();
+  }
+
+  @Override
+  public Future<Void> close() {
+    return Future.succeededFuture();
+  }
+}
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
index 6ecc3687b..5aa50b023 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImpl.java
@@ -27,7 +27,9 @@ import dev.knative.eventing.kafka.broker.dispatcher.ConsumerVerticleFactory;
 import dev.knative.eventing.kafka.broker.dispatcher.DeliveryOrder;
 import dev.knative.eventing.kafka.broker.dispatcher.Filter;
 import dev.knative.eventing.kafka.broker.dispatcher.RecordDispatcherListener;
+import dev.knative.eventing.kafka.broker.dispatcher.ResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.KafkaResponseHandler;
+import dev.knative.eventing.kafka.broker.dispatcher.impl.NoopResponseHandler;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.RecordDispatcherImpl;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.WebClientCloudEventSender;
 import dev.knative.eventing.kafka.broker.dispatcher.impl.consumer.BaseConsumerVerticle;
@@ -62,6 +64,7 @@ import java.util.Properties;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -145,8 +148,6 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
         KafkaConsumer<String, CloudEvent> consumer = createConsumer(vertx, consumerConfigs);
         AutoCloseable metricsCloser = Metrics.register(consumer.unwrap());
 
-        KafkaProducer<String, CloudEvent> producer = createProducer(vertx, producerConfigs);
-
         final var egressConfig =
           egress.hasEgressConfig() ?
             egress.getEgressConfig() :
@@ -166,11 +167,13 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
           new AttributesFilter(egress.getFilter().getAttributesMap()) :
           Filter.noop();
 
+        final var responseHandler = getNoopResponseHandlerOrDefault(egress, () -> getKafkaResponseHandler(vertx, producerConfigs, resource));
+
         final RecordDispatcherImpl recordDispatcher = new RecordDispatcherImpl(
           filter,
           egressSubscriberSender,
           egressDeadLetterSender,
-          new KafkaResponseHandler(producer, resource.getTopics(0)),
+          responseHandler,
           getOffsetManager(deliveryOrder, consumer, eventsSentCounter::increment),
           ConsumerTracer.create(
             ((VertxInternal) vertx).tracer(),
@@ -191,6 +194,21 @@ public class ConsumerVerticleFactoryImpl implements ConsumerVerticleFactory {
     return getConsumerVerticle(deliveryOrder, initializer, new HashSet<>(resource.getTopicsList()));
   }
 
+  static ResponseHandler getNoopResponseHandlerOrDefault(final DataPlaneContract.Egress egress,
+                                                         final Supplier<ResponseHandler> defaultSupplier) {
+    if (egress.hasDiscardReply()) {
+      return new NoopResponseHandler();
+    }
+    return defaultSupplier.get();
+  }
+
+  private KafkaResponseHandler getKafkaResponseHandler(final Vertx vertx,
+                                                       final Map<String, Object> producerConfigs,
+                                                       final DataPlaneContract.Resource resource) {
+    final KafkaProducer<String, CloudEvent> producer = createProducer(vertx, producerConfigs);
+    return new KafkaResponseHandler(producer, resource.getTopics(0));
+  }
+
   protected KafkaProducer<String, CloudEvent> createProducer(final Vertx vertx,
                                                              final Map<String, Object> producerConfigs) {
     Properties producerProperties = new Properties();
diff --git a/data-plane/docker/Dockerfile b/data-plane/docker/Dockerfile
index efd1e3326..171817737 100644
--- a/data-plane/docker/Dockerfile
+++ b/data-plane/docker/Dockerfile
@@ -24,9 +24,6 @@ ARG APP_JAR
 
 WORKDIR /build
 
-# https://github.com/AdoptOpenJDK/openjdk-docker/issues/260
-RUN DEBIAN_FRONTEND=noninteractive apt-get update && apt-get install -y binutils
-
 COPY pom.xml .
 COPY .editorconfig .
 COPY core/pom.xml core/pom.xml
diff --git a/data-plane/pom.xml b/data-plane/pom.xml
index 98441ac3c..f39d4cba8 100644
--- a/data-plane/pom.xml
+++ b/data-plane/pom.xml
@@ -36,7 +36,7 @@
     <maven.shade.plugin.version>3.2.4</maven.shade.plugin.version>
     <maven.editorconfig.plugin.version>0.1.1</maven.editorconfig.plugin.version>
     <maven.jacoco.plugin.version>0.8.7</maven.jacoco.plugin.version>
-    <maven.enforcer.plugin.version>3.0.0-M3</maven.enforcer.plugin.version>
+    <maven.enforcer.plugin.version>3.0.0</maven.enforcer.plugin.version>
     <maven.license.plugin.version>2.0.0</maven.license.plugin.version>
     <mycila.license.plugin.version>4.1</mycila.license.plugin.version>
 
@@ -166,6 +166,10 @@
             <artifactId>okhttp</artifactId>
             <groupId>com.squareup.okhttp3</groupId>
           </exclusion>
+          <exclusion>
+            <artifactId>builder-annotations</artifactId>
+            <groupId>io.sundr</groupId>
+          </exclusion>
         </exclusions>
       </dependency>
 
diff --git a/go.mod b/go.mod
index d310470ab..f031d8c19 100644
--- a/go.mod
+++ b/go.mod
@@ -21,8 +21,8 @@ require (
 	k8s.io/apiserver v0.21.4
 	k8s.io/client-go v0.21.4
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b
-	knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403
+	knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0
+	knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11
 	knative.dev/hack v0.0.0-20210806075220-815cd312d65c
 	knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97
 	knative.dev/reconciler-test v0.0.0-20210930064245-45904ca4383d
diff --git a/go.sum b/go.sum
index 6c85e98b6..49e5f5722 100644
--- a/go.sum
+++ b/go.sum
@@ -642,7 +642,6 @@ github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6So
 github.com/rogpeppe/fastuuid v1.2.0 h1:Ppwyp6VYCF1nvBTXL3trRso7mXMlRrw9ooo375wvi2s=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/rs/dnscache v0.0.0-20210201191234-295bba877686/go.mod h1:qe5TWALJ8/a1Lqznoc5BDHpYX/8HU60Hm2AwRmqzxqA=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
@@ -1302,17 +1301,15 @@ k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKU
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 knative.dev/control-protocol v0.0.0-20210929151023-4b73baddb22a/go.mod h1:RxOuZnxwP51dsuMxFVXFK+7109d3YUqCmAKFH9n+mG4=
 knative.dev/eventing v0.26.1-0.20210930102845-5b9ac948cefc/go.mod h1:HwxfiCneWF5L3AeDFuTWnEs/aFZqr/pdVWV9i5AH5vs=
-knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b h1:Mm+wWAHd+m8GG14SxEZvu96Ad9jDjmqE16DvM0u7Ink=
-knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b/go.mod h1:Bv9ByJlbDLOlpwWHISGMum9PnZ2yfQjcOTgcfckWF1M=
-knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403 h1:iqgzQxskz0a8kqhgjm7gpp8LW2av5yzTmMcs39Y9N+Y=
-knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403/go.mod h1:qK71ngnAY7DRJj7wb1pbW0cCKW5sbAn7fV14KaqP+OY=
+knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0 h1:HOct4MnEEnB8luibpEkGowzpywIRZl9R0EKOBXpI4X4=
+knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0/go.mod h1:YwdAAjLecPcaZkpyaNH3Ajn7kCzsfsvMGYS2hyC+R+I=
+knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11 h1:VCpDGwW6ychG+OcwJtZzh9CCZ5sa+T3M/7JjSBOpbAg=
+knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11/go.mod h1:+NeJwjvTos9goI5LshNCNJmSC9dNHCP8rsh2QroeR8Y=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c h1:nOXoDWAAItwr4o0dp3nHr6skgpVD4IvME/UX84YNl5k=
 knative.dev/hack v0.0.0-20210806075220-815cd312d65c/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210806075220-815cd312d65c/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
-knative.dev/networking v0.0.0-20210929162523-749575ef53f8/go.mod h1:IQMjYdQwtOsqJ+fQDs8MZGB/cPxXdMc4bpP7fdtXp2Q=
 knative.dev/pkg v0.0.0-20210914164111-4857ab6939e3/go.mod h1:jMSqkNMsrzuy+XR4Yr/BMy7SDVbUOl3KKB6+5MR+ZU8=
 knative.dev/pkg v0.0.0-20210929111822-2267a4cbebb8/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
-knative.dev/pkg v0.0.0-20211005203644-a00ba487121e/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
 knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97 h1:art4USpwHkiBKwu1C2X2KVHpDcH9L3mhGlfrHnQowS4=
 knative.dev/pkg v0.0.0-20211006213726-9179f78dcf97/go.mod h1:r27D20afKNeK+9aNOg+0qMv8JgQcyeP+CAYQIR1jEQY=
 knative.dev/reconciler-test v0.0.0-20210915181908-49fac7555086/go.mod h1:6yDmb26SINSmgw6wVy9qQwgRMewiW8ddkkwGLR0ZvOY=
diff --git a/hack/data-plane.sh b/hack/data-plane.sh
index 28f7bc8b8..c692afa19 100755
--- a/hack/data-plane.sh
+++ b/hack/data-plane.sh
@@ -43,8 +43,8 @@ readonly dispatcher="${KNATIVE_KAFKA_BROKER_DISPATCHER:-knative-kafka-broker-dis
 
 # The BASE_IMAGE must have system libraries (libc, zlib, etc) compatible with the JAVA_IMAGE because
 # Jlink generates a jdk linked to the same system libraries available on the base images.
-readonly BASE_IMAGE=gcr.io/distroless/java-debian10:base-nonroot # Based on debian:buster
-readonly JAVA_IMAGE=docker.io/adoptopenjdk/openjdk16:debian      # Based on debian:buster
+readonly BASE_IMAGE=gcr.io/distroless/java-debian11:base-nonroot  # Based on debian:buster
+readonly JAVA_IMAGE=docker.io/eclipse-temurin:17-jdk-centos7      # Based on centos7
 
 readonly RECEIVER_JAR="receiver-1.0-SNAPSHOT.jar"
 readonly RECEIVER_DIRECTORY=receiver
@@ -146,7 +146,7 @@ function k8s() {
 
 function data_plane_unit_tests() {
   pushd ${DATA_PLANE_DIR} || fail_test
-  ./mvnw clean verify -Dhttp.keepAlive=false -Dmaven.wagon.http.pool=false -Dmaven.wagon.http.retryHandler.count=6 --no-transfer-progress
+  ./mvnw clean verify -Dmaven.wagon.http.retryHandler.count=6 --no-transfer-progress
   mvn_output=$?
 
   echo "Copy test reports in ${ARTIFACTS}"
diff --git a/proto/contract.proto b/proto/contract.proto
index 411944fc5..c16118469 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -70,6 +70,9 @@ message Egress {
 
     // Send the response to a Kafka topic
     Empty replyToOriginalTopic = 4;
+
+    // Discard response.
+    Empty discardReply = 9;
   }
 
   Filter filter = 5;
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
new file mode 100644
index 000000000..d78c7a786
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/clientset_generated.go
@@ -0,0 +1,103 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/discovery"
+	fakediscovery "k8s.io/client-go/discovery/fake"
+	"k8s.io/client-go/testing"
+	clientset "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
+	fakebindingsv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+	fakekafkav1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+	fakemessagingv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1"
+	fakesourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake"
+)
+
+// NewSimpleClientset returns a clientset that will respond with the provided objects.
+// It's backed by a very simple object tracker that processes creates, updates and deletions as-is,
+// without applying any validations and/or defaults. It shouldn't be considered a replacement
+// for a real clientset and is mostly useful in simple unit tests.
+func NewSimpleClientset(objects ...runtime.Object) *Clientset {
+	o := testing.NewObjectTracker(scheme, codecs.UniversalDecoder())
+	for _, obj := range objects {
+		if err := o.Add(obj); err != nil {
+			panic(err)
+		}
+	}
+
+	cs := &Clientset{tracker: o}
+	cs.discovery = &fakediscovery.FakeDiscovery{Fake: &cs.Fake}
+	cs.AddReactor("*", "*", testing.ObjectReaction(o))
+	cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) {
+		gvr := action.GetResource()
+		ns := action.GetNamespace()
+		watch, err := o.Watch(gvr, ns)
+		if err != nil {
+			return false, nil, err
+		}
+		return true, watch, nil
+	})
+
+	return cs
+}
+
+// Clientset implements clientset.Interface. Meant to be embedded into a
+// struct to get a default implementation. This makes faking out just the method
+// you want to test easier.
+type Clientset struct {
+	testing.Fake
+	discovery *fakediscovery.FakeDiscovery
+	tracker   testing.ObjectTracker
+}
+
+func (c *Clientset) Discovery() discovery.DiscoveryInterface {
+	return c.discovery
+}
+
+func (c *Clientset) Tracker() testing.ObjectTracker {
+	return c.tracker
+}
+
+var _ clientset.Interface = &Clientset{}
+
+// BindingsV1beta1 retrieves the BindingsV1beta1Client
+func (c *Clientset) BindingsV1beta1() bindingsv1beta1.BindingsV1beta1Interface {
+	return &fakebindingsv1beta1.FakeBindingsV1beta1{Fake: &c.Fake}
+}
+
+// KafkaV1alpha1 retrieves the KafkaV1alpha1Client
+func (c *Clientset) KafkaV1alpha1() kafkav1alpha1.KafkaV1alpha1Interface {
+	return &fakekafkav1alpha1.FakeKafkaV1alpha1{Fake: &c.Fake}
+}
+
+// MessagingV1beta1 retrieves the MessagingV1beta1Client
+func (c *Clientset) MessagingV1beta1() messagingv1beta1.MessagingV1beta1Interface {
+	return &fakemessagingv1beta1.FakeMessagingV1beta1{Fake: &c.Fake}
+}
+
+// SourcesV1beta1 retrieves the SourcesV1beta1Client
+func (c *Clientset) SourcesV1beta1() sourcesv1beta1.SourcesV1beta1Interface {
+	return &fakesourcesv1beta1.FakeSourcesV1beta1{Fake: &c.Fake}
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go
new file mode 100644
index 000000000..3b00d536c
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// This package has the automatically generated fake clientset.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go
new file mode 100644
index 000000000..a396f69d6
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake/register.go
@@ -0,0 +1,62 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	serializer "k8s.io/apimachinery/pkg/runtime/serializer"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	bindingsv1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+	kafkav1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+var scheme = runtime.NewScheme()
+var codecs = serializer.NewCodecFactory(scheme)
+
+var localSchemeBuilder = runtime.SchemeBuilder{
+	bindingsv1beta1.AddToScheme,
+	kafkav1alpha1.AddToScheme,
+	messagingv1beta1.AddToScheme,
+	sourcesv1beta1.AddToScheme,
+}
+
+// AddToScheme adds all types of this clientset into the given scheme. This allows composition
+// of clientsets, like in:
+//
+//   import (
+//     "k8s.io/client-go/kubernetes"
+//     clientsetscheme "k8s.io/client-go/kubernetes/scheme"
+//     aggregatorclientsetscheme "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/scheme"
+//   )
+//
+//   kclientset, _ := kubernetes.NewForConfig(c)
+//   _ = aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
+//
+// After this, RawExtensions in Kubernetes types will serialize kube-aggregator types
+// correctly.
+var AddToScheme = localSchemeBuilder.AddToScheme
+
+func init() {
+	v1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"})
+	utilruntime.Must(AddToScheme(scheme))
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go
new file mode 100644
index 000000000..2237e7bfa
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_bindings_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1"
+)
+
+type FakeBindingsV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeBindingsV1beta1) KafkaBindings(namespace string) v1beta1.KafkaBindingInterface {
+	return &FakeKafkaBindings{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeBindingsV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go
new file mode 100644
index 000000000..609547ae2
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake/fake_kafkabinding.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1"
+)
+
+// FakeKafkaBindings implements KafkaBindingInterface
+type FakeKafkaBindings struct {
+	Fake *FakeBindingsV1beta1
+	ns   string
+}
+
+var kafkabindingsResource = schema.GroupVersionResource{Group: "bindings.knative.dev", Version: "v1beta1", Resource: "kafkabindings"}
+
+var kafkabindingsKind = schema.GroupVersionKind{Group: "bindings.knative.dev", Version: "v1beta1", Kind: "KafkaBinding"}
+
+// Get takes name of the kafkaBinding, and returns the corresponding kafkaBinding object, and an error if there is any.
+func (c *FakeKafkaBindings) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkabindingsResource, c.ns, name), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaBindings that match those selectors.
+func (c *FakeKafkaBindings) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaBindingList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkabindingsResource, kafkabindingsKind, c.ns, opts), &v1beta1.KafkaBindingList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaBindingList{ListMeta: obj.(*v1beta1.KafkaBindingList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaBindingList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaBindings.
+func (c *FakeKafkaBindings) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkabindingsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaBinding and creates it.  Returns the server's representation of the kafkaBinding, and an error, if there is any.
+func (c *FakeKafkaBindings) Create(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.CreateOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkabindingsResource, c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// Update takes the representation of a kafkaBinding and updates it. Returns the server's representation of the kafkaBinding, and an error, if there is any.
+func (c *FakeKafkaBindings) Update(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.UpdateOptions) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkabindingsResource, c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaBindings) UpdateStatus(ctx context.Context, kafkaBinding *v1beta1.KafkaBinding, opts v1.UpdateOptions) (*v1beta1.KafkaBinding, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkabindingsResource, "status", c.ns, kafkaBinding), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
+
+// Delete takes name of the kafkaBinding and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaBindings) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkabindingsResource, c.ns, name), &v1beta1.KafkaBinding{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaBindings) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkabindingsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaBindingList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaBinding.
+func (c *FakeKafkaBindings) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaBinding, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkabindingsResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaBinding{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaBinding), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go
new file mode 100644
index 000000000..53377b6b9
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_kafka_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1"
+)
+
+type FakeKafkaV1alpha1 struct {
+	*testing.Fake
+}
+
+func (c *FakeKafkaV1alpha1) ResetOffsets(namespace string) v1alpha1.ResetOffsetInterface {
+	return &FakeResetOffsets{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeKafkaV1alpha1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
new file mode 100644
index 000000000..af76f8c0b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake/fake_resetoffset.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1alpha1 "knative.dev/eventing-kafka/pkg/apis/kafka/v1alpha1"
+)
+
+// FakeResetOffsets implements ResetOffsetInterface
+type FakeResetOffsets struct {
+	Fake *FakeKafkaV1alpha1
+	ns   string
+}
+
+var resetoffsetsResource = schema.GroupVersionResource{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Resource: "resetoffsets"}
+
+var resetoffsetsKind = schema.GroupVersionKind{Group: "kafka.eventing.knative.dev", Version: "v1alpha1", Kind: "ResetOffset"}
+
+// Get takes name of the resetOffset, and returns the corresponding resetOffset object, and an error if there is any.
+func (c *FakeResetOffsets) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// List takes label and field selectors, and returns the list of ResetOffsets that match those selectors.
+func (c *FakeResetOffsets) List(ctx context.Context, opts v1.ListOptions) (result *v1alpha1.ResetOffsetList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(resetoffsetsResource, resetoffsetsKind, c.ns, opts), &v1alpha1.ResetOffsetList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1alpha1.ResetOffsetList{ListMeta: obj.(*v1alpha1.ResetOffsetList).ListMeta}
+	for _, item := range obj.(*v1alpha1.ResetOffsetList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested resetOffsets.
+func (c *FakeResetOffsets) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(resetoffsetsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a resetOffset and creates it.  Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Create(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.CreateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Update takes the representation of a resetOffset and updates it. Returns the server's representation of the resetOffset, and an error, if there is any.
+func (c *FakeResetOffsets) Update(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(resetoffsetsResource, c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeResetOffsets) UpdateStatus(ctx context.Context, resetOffset *v1alpha1.ResetOffset, opts v1.UpdateOptions) (*v1alpha1.ResetOffset, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(resetoffsetsResource, "status", c.ns, resetOffset), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
+
+// Delete takes name of the resetOffset and deletes it. Returns an error if one occurs.
+func (c *FakeResetOffsets) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(resetoffsetsResource, c.ns, name), &v1alpha1.ResetOffset{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeResetOffsets) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(resetoffsetsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1alpha1.ResetOffsetList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched resetOffset.
+func (c *FakeResetOffsets) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1alpha1.ResetOffset, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(resetoffsetsResource, c.ns, name, pt, data, subresources...), &v1alpha1.ResetOffset{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ResetOffset), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
new file mode 100644
index 000000000..5ab4e5a86
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_kafkachannel.go
@@ -0,0 +1,142 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
+)
+
+// FakeKafkaChannels implements KafkaChannelInterface
+type FakeKafkaChannels struct {
+	Fake *FakeMessagingV1beta1
+	ns   string
+}
+
+var kafkachannelsResource = schema.GroupVersionResource{Group: "messaging.knative.dev", Version: "v1beta1", Resource: "kafkachannels"}
+
+var kafkachannelsKind = schema.GroupVersionKind{Group: "messaging.knative.dev", Version: "v1beta1", Kind: "KafkaChannel"}
+
+// Get takes name of the kafkaChannel, and returns the corresponding kafkaChannel object, and an error if there is any.
+func (c *FakeKafkaChannels) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaChannels that match those selectors.
+func (c *FakeKafkaChannels) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaChannelList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkachannelsResource, kafkachannelsKind, c.ns, opts), &v1beta1.KafkaChannelList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaChannelList{ListMeta: obj.(*v1beta1.KafkaChannelList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaChannelList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaChannels.
+func (c *FakeKafkaChannels) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkachannelsResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaChannel and creates it.  Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Create(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.CreateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Update takes the representation of a kafkaChannel and updates it. Returns the server's representation of the kafkaChannel, and an error, if there is any.
+func (c *FakeKafkaChannels) Update(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkachannelsResource, c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaChannels) UpdateStatus(ctx context.Context, kafkaChannel *v1beta1.KafkaChannel, opts v1.UpdateOptions) (*v1beta1.KafkaChannel, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkachannelsResource, "status", c.ns, kafkaChannel), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
+
+// Delete takes name of the kafkaChannel and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaChannels) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkachannelsResource, c.ns, name), &v1beta1.KafkaChannel{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaChannels) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkachannelsResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaChannelList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaChannel.
+func (c *FakeKafkaChannels) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaChannel, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkachannelsResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaChannel{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaChannel), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
new file mode 100644
index 000000000..9befbac81
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake/fake_messaging_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1"
+)
+
+type FakeMessagingV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeMessagingV1beta1) KafkaChannels(namespace string) v1beta1.KafkaChannelInterface {
+	return &FakeKafkaChannels{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeMessagingV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go
new file mode 100644
index 000000000..d5a46ef55
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/doc.go
@@ -0,0 +1,20 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+// Package fake has the automatically generated clients.
+package fake
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
new file mode 100644
index 000000000..a35a59c17
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_kafkasource.go
@@ -0,0 +1,154 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	"context"
+
+	autoscalingv1 "k8s.io/api/autoscaling/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	schema "k8s.io/apimachinery/pkg/runtime/schema"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+)
+
+// FakeKafkaSources implements KafkaSourceInterface
+type FakeKafkaSources struct {
+	Fake *FakeSourcesV1beta1
+	ns   string
+}
+
+var kafkasourcesResource = schema.GroupVersionResource{Group: "sources.knative.dev", Version: "v1beta1", Resource: "kafkasources"}
+
+var kafkasourcesKind = schema.GroupVersionKind{Group: "sources.knative.dev", Version: "v1beta1", Kind: "KafkaSource"}
+
+// Get takes name of the kafkaSource, and returns the corresponding kafkaSource object, and an error if there is any.
+func (c *FakeKafkaSources) Get(ctx context.Context, name string, options v1.GetOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetAction(kafkasourcesResource, c.ns, name), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// List takes label and field selectors, and returns the list of KafkaSources that match those selectors.
+func (c *FakeKafkaSources) List(ctx context.Context, opts v1.ListOptions) (result *v1beta1.KafkaSourceList, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewListAction(kafkasourcesResource, kafkasourcesKind, c.ns, opts), &v1beta1.KafkaSourceList{})
+
+	if obj == nil {
+		return nil, err
+	}
+
+	label, _, _ := testing.ExtractFromListOptions(opts)
+	if label == nil {
+		label = labels.Everything()
+	}
+	list := &v1beta1.KafkaSourceList{ListMeta: obj.(*v1beta1.KafkaSourceList).ListMeta}
+	for _, item := range obj.(*v1beta1.KafkaSourceList).Items {
+		if label.Matches(labels.Set(item.Labels)) {
+			list.Items = append(list.Items, item)
+		}
+	}
+	return list, err
+}
+
+// Watch returns a watch.Interface that watches the requested kafkaSources.
+func (c *FakeKafkaSources) Watch(ctx context.Context, opts v1.ListOptions) (watch.Interface, error) {
+	return c.Fake.
+		InvokesWatch(testing.NewWatchAction(kafkasourcesResource, c.ns, opts))
+
+}
+
+// Create takes the representation of a kafkaSource and creates it.  Returns the server's representation of the kafkaSource, and an error, if there is any.
+func (c *FakeKafkaSources) Create(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.CreateOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewCreateAction(kafkasourcesResource, c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// Update takes the representation of a kafkaSource and updates it. Returns the server's representation of the kafkaSource, and an error, if there is any.
+func (c *FakeKafkaSources) Update(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.UpdateOptions) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateAction(kafkasourcesResource, c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeKafkaSources) UpdateStatus(ctx context.Context, kafkaSource *v1beta1.KafkaSource, opts v1.UpdateOptions) (*v1beta1.KafkaSource, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(kafkasourcesResource, "status", c.ns, kafkaSource), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// Delete takes name of the kafkaSource and deletes it. Returns an error if one occurs.
+func (c *FakeKafkaSources) Delete(ctx context.Context, name string, opts v1.DeleteOptions) error {
+	_, err := c.Fake.
+		Invokes(testing.NewDeleteAction(kafkasourcesResource, c.ns, name), &v1beta1.KafkaSource{})
+
+	return err
+}
+
+// DeleteCollection deletes a collection of objects.
+func (c *FakeKafkaSources) DeleteCollection(ctx context.Context, opts v1.DeleteOptions, listOpts v1.ListOptions) error {
+	action := testing.NewDeleteCollectionAction(kafkasourcesResource, c.ns, listOpts)
+
+	_, err := c.Fake.Invokes(action, &v1beta1.KafkaSourceList{})
+	return err
+}
+
+// Patch applies the patch and returns the patched kafkaSource.
+func (c *FakeKafkaSources) Patch(ctx context.Context, name string, pt types.PatchType, data []byte, opts v1.PatchOptions, subresources ...string) (result *v1beta1.KafkaSource, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewPatchSubresourceAction(kafkasourcesResource, c.ns, name, pt, data, subresources...), &v1beta1.KafkaSource{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1beta1.KafkaSource), err
+}
+
+// GetScale takes name of the kafkaSource, and returns the corresponding scale object, and an error if there is any.
+func (c *FakeKafkaSources) GetScale(ctx context.Context, kafkaSourceName string, options v1.GetOptions) (result *autoscalingv1.Scale, err error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewGetSubresourceAction(kafkasourcesResource, c.ns, "scale", kafkaSourceName), &autoscalingv1.Scale{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*autoscalingv1.Scale), err
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go
new file mode 100644
index 000000000..e3876454b
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake/fake_sources_client.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by client-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	rest "k8s.io/client-go/rest"
+	testing "k8s.io/client-go/testing"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1"
+)
+
+type FakeSourcesV1beta1 struct {
+	*testing.Fake
+}
+
+func (c *FakeSourcesV1beta1) KafkaSources(namespace string) v1beta1.KafkaSourceInterface {
+	return &FakeKafkaSources{c, namespace}
+}
+
+// RESTClient returns a RESTClient that is used to communicate
+// with API server by this client implementation.
+func (c *FakeSourcesV1beta1) RESTClient() rest.Interface {
+	var ret *rest.RESTClient
+	return ret
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go
new file mode 100644
index 000000000..96faa7fc2
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/client/fake/fake.go
@@ -0,0 +1,57 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	rest "k8s.io/client-go/rest"
+	fake "knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Fake.RegisterClient(withClient)
+	injection.Fake.RegisterClientFetcher(func(ctx context.Context) interface{} {
+		return Get(ctx)
+	})
+}
+
+func withClient(ctx context.Context, cfg *rest.Config) context.Context {
+	ctx, _ = With(ctx)
+	return ctx
+}
+
+func With(ctx context.Context, objects ...runtime.Object) (context.Context, *fake.Clientset) {
+	cs := fake.NewSimpleClientset(objects...)
+	return context.WithValue(ctx, client.Key{}, cs), cs
+}
+
+// Get extracts the Kubernetes client from the context.
+func Get(ctx context.Context) *fake.Clientset {
+	untyped := ctx.Value(client.Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake.Clientset from context.")
+	}
+	return untyped.(*fake.Clientset)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go
new file mode 100644
index 000000000..be34de9ef
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake/fake.go
@@ -0,0 +1,45 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	externalversions "knative.dev/eventing-kafka/pkg/client/informers/externalversions"
+	fake "knative.dev/eventing-kafka/pkg/client/injection/client/fake"
+	factory "knative.dev/eventing-kafka/pkg/client/injection/informers/factory"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = factory.Get
+
+func init() {
+	injection.Fake.RegisterInformerFactory(withInformerFactory)
+}
+
+func withInformerFactory(ctx context.Context) context.Context {
+	c := fake.Get(ctx)
+	opts := make([]externalversions.SharedInformerOption, 0, 1)
+	if injection.HasNamespaceScope(ctx) {
+		opts = append(opts, externalversions.WithNamespace(injection.GetNamespaceScope(ctx)))
+	}
+	return context.WithValue(ctx, factory.Key{},
+		externalversions.NewSharedInformerFactoryWithOptions(c, controller.GetResyncPeriod(ctx), opts...))
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go
new file mode 100644
index 000000000..80d1e5b6d
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake/fake.go
@@ -0,0 +1,40 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package fake
+
+import (
+	context "context"
+
+	fake "knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake"
+	kafkasource "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+)
+
+var Get = kafkasource.Get
+
+func init() {
+	injection.Fake.RegisterInformer(withInformer)
+}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := fake.Get(ctx)
+	inf := f.Sources().V1beta1().KafkaSources()
+	return context.WithValue(ctx, kafkasource.Key{}, inf), inf.Informer()
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go
new file mode 100644
index 000000000..5ee3d8632
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/kafkasource.go
@@ -0,0 +1,106 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	cache "k8s.io/client-go/tools/cache"
+	apissourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	v1beta1 "knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	factory "knative.dev/eventing-kafka/pkg/client/injection/informers/factory"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	controller "knative.dev/pkg/controller"
+	injection "knative.dev/pkg/injection"
+	logging "knative.dev/pkg/logging"
+)
+
+func init() {
+	injection.Default.RegisterInformer(withInformer)
+	injection.Dynamic.RegisterDynamicInformer(withDynamicInformer)
+}
+
+// Key is used for associating the Informer inside the context.Context.
+type Key struct{}
+
+func withInformer(ctx context.Context) (context.Context, controller.Informer) {
+	f := factory.Get(ctx)
+	inf := f.Sources().V1beta1().KafkaSources()
+	return context.WithValue(ctx, Key{}, inf), inf.Informer()
+}
+
+func withDynamicInformer(ctx context.Context) context.Context {
+	inf := &wrapper{client: client.Get(ctx)}
+	return context.WithValue(ctx, Key{}, inf)
+}
+
+// Get extracts the typed informer from the context.
+func Get(ctx context.Context) v1beta1.KafkaSourceInformer {
+	untyped := ctx.Value(Key{})
+	if untyped == nil {
+		logging.FromContext(ctx).Panic(
+			"Unable to fetch knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1.KafkaSourceInformer from context.")
+	}
+	return untyped.(v1beta1.KafkaSourceInformer)
+}
+
+type wrapper struct {
+	client versioned.Interface
+
+	namespace string
+}
+
+var _ v1beta1.KafkaSourceInformer = (*wrapper)(nil)
+var _ sourcesv1beta1.KafkaSourceLister = (*wrapper)(nil)
+
+func (w *wrapper) Informer() cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(nil, &apissourcesv1beta1.KafkaSource{}, 0, nil)
+}
+
+func (w *wrapper) Lister() sourcesv1beta1.KafkaSourceLister {
+	return w
+}
+
+func (w *wrapper) KafkaSources(namespace string) sourcesv1beta1.KafkaSourceNamespaceLister {
+	return &wrapper{client: w.client, namespace: namespace}
+}
+
+func (w *wrapper) List(selector labels.Selector) (ret []*apissourcesv1beta1.KafkaSource, err error) {
+	lo, err := w.client.SourcesV1beta1().KafkaSources(w.namespace).List(context.TODO(), v1.ListOptions{
+		LabelSelector: selector.String(),
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+	if err != nil {
+		return nil, err
+	}
+	for idx := range lo.Items {
+		ret = append(ret, &lo.Items[idx])
+	}
+	return ret, nil
+}
+
+func (w *wrapper) Get(name string) (*apissourcesv1beta1.KafkaSource, error) {
+	return w.client.SourcesV1beta1().KafkaSources(w.namespace).Get(context.TODO(), name, v1.GetOptions{
+		// TODO(mattmoor): Incorporate resourceVersion bounds based on staleness criteria.
+	})
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
new file mode 100644
index 000000000..93bcb825f
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/controller.go
@@ -0,0 +1,162 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+	fmt "fmt"
+	reflect "reflect"
+	strings "strings"
+
+	zap "go.uber.org/zap"
+	corev1 "k8s.io/api/core/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	watch "k8s.io/apimachinery/pkg/watch"
+	scheme "k8s.io/client-go/kubernetes/scheme"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	record "k8s.io/client-go/tools/record"
+	versionedscheme "knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme"
+	client "knative.dev/eventing-kafka/pkg/client/injection/client"
+	kafkasource "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource"
+	kubeclient "knative.dev/pkg/client/injection/kube/client"
+	controller "knative.dev/pkg/controller"
+	logging "knative.dev/pkg/logging"
+	logkey "knative.dev/pkg/logging/logkey"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+const (
+	defaultControllerAgentName = "kafkasource-controller"
+	defaultFinalizerName       = "kafkasources.sources.knative.dev"
+)
+
+// NewImpl returns a controller.Impl that handles queuing and feeding work from
+// the queue through an implementation of controller.Reconciler, delegating to
+// the provided Interface and optional Finalizer methods. OptionsFn is used to return
+// controller.ControllerOptions to be used by the internal reconciler.
+func NewImpl(ctx context.Context, r Interface, optionsFns ...controller.OptionsFn) *controller.Impl {
+	logger := logging.FromContext(ctx)
+
+	// Check the options function input. It should be 0 or 1.
+	if len(optionsFns) > 1 {
+		logger.Fatal("Up to one options function is supported, found: ", len(optionsFns))
+	}
+
+	kafkasourceInformer := kafkasource.Get(ctx)
+
+	lister := kafkasourceInformer.Lister()
+
+	var promoteFilterFunc func(obj interface{}) bool
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					if promoteFilterFunc != nil {
+						if ok := promoteFilterFunc(elt); !ok {
+							continue
+						}
+					}
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client.Get(ctx),
+		Lister:        lister,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	ctrType := reflect.TypeOf(r).Elem()
+	ctrTypeName := fmt.Sprintf("%s.%s", ctrType.PkgPath(), ctrType.Name())
+	ctrTypeName = strings.ReplaceAll(ctrTypeName, "/", ".")
+
+	logger = logger.With(
+		zap.String(logkey.ControllerType, ctrTypeName),
+		zap.String(logkey.Kind, "sources.knative.dev.KafkaSource"),
+	)
+
+	impl := controller.NewContext(ctx, rec, controller.ControllerOptions{WorkQueueName: ctrTypeName, Logger: logger})
+	agentName := defaultControllerAgentName
+
+	// Pass impl to the options. Save any optional results.
+	for _, fn := range optionsFns {
+		opts := fn(impl)
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.AgentName != "" {
+			agentName = opts.AgentName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+		if opts.PromoteFilterFunc != nil {
+			promoteFilterFunc = opts.PromoteFilterFunc
+		}
+	}
+
+	rec.Recorder = createRecorder(ctx, agentName)
+
+	return impl
+}
+
+func createRecorder(ctx context.Context, agentName string) record.EventRecorder {
+	logger := logging.FromContext(ctx)
+
+	recorder := controller.GetEventRecorder(ctx)
+	if recorder == nil {
+		// Create event broadcaster
+		logger.Debug("Creating event broadcaster")
+		eventBroadcaster := record.NewBroadcaster()
+		watches := []watch.Interface{
+			eventBroadcaster.StartLogging(logger.Named("event-broadcaster").Infof),
+			eventBroadcaster.StartRecordingToSink(
+				&v1.EventSinkImpl{Interface: kubeclient.Get(ctx).CoreV1().Events("")}),
+		}
+		recorder = eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: agentName})
+		go func() {
+			<-ctx.Done()
+			for _, w := range watches {
+				w.Stop()
+			}
+		}()
+	}
+
+	return recorder
+}
+
+func init() {
+	versionedscheme.AddToScheme(scheme.Scheme)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
new file mode 100644
index 000000000..268361e94
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/reconciler.go
@@ -0,0 +1,450 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	context "context"
+	json "encoding/json"
+	fmt "fmt"
+
+	zap "go.uber.org/zap"
+	v1 "k8s.io/api/core/v1"
+	equality "k8s.io/apimachinery/pkg/api/equality"
+	errors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	labels "k8s.io/apimachinery/pkg/labels"
+	types "k8s.io/apimachinery/pkg/types"
+	sets "k8s.io/apimachinery/pkg/util/sets"
+	record "k8s.io/client-go/tools/record"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	versioned "knative.dev/eventing-kafka/pkg/client/clientset/versioned"
+	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/client/listers/sources/v1beta1"
+	controller "knative.dev/pkg/controller"
+	kmp "knative.dev/pkg/kmp"
+	logging "knative.dev/pkg/logging"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// Interface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaSource.
+type Interface interface {
+	// ReconcileKind implements custom logic to reconcile v1beta1.KafkaSource. Any changes
+	// to the objects .Status or .Finalizers will be propagated to the stored
+	// object. It is recommended that implementors do not call any update calls
+	// for the Kind inside of ReconcileKind, it is the responsibility of the calling
+	// controller to propagate those properties. The resource passed to ReconcileKind
+	// will always have an empty deletion timestamp.
+	ReconcileKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+// Finalizer defines the strongly typed interfaces to be implemented by a
+// controller finalizing v1beta1.KafkaSource.
+type Finalizer interface {
+	// FinalizeKind implements custom logic to finalize v1beta1.KafkaSource. Any changes
+	// to the objects .Status or .Finalizers will be ignored. Returning a nil or
+	// Normal type reconciler.Event will allow the finalizer to be deleted on
+	// the resource. The resource passed to FinalizeKind will always have a set
+	// deletion timestamp.
+	FinalizeKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+// ReadOnlyInterface defines the strongly typed interfaces to be implemented by a
+// controller reconciling v1beta1.KafkaSource if they want to process resources for which
+// they are not the leader.
+type ReadOnlyInterface interface {
+	// ObserveKind implements logic to observe v1beta1.KafkaSource.
+	// This method should not write to the API.
+	ObserveKind(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+}
+
+type doReconcile func(ctx context.Context, o *v1beta1.KafkaSource) reconciler.Event
+
+// reconcilerImpl implements controller.Reconciler for v1beta1.KafkaSource resources.
+type reconcilerImpl struct {
+	// LeaderAwareFuncs is inlined to help us implement reconciler.LeaderAware.
+	reconciler.LeaderAwareFuncs
+
+	// Client is used to write back status updates.
+	Client versioned.Interface
+
+	// Listers index properties about resources.
+	Lister sourcesv1beta1.KafkaSourceLister
+
+	// Recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	Recorder record.EventRecorder
+
+	// configStore allows for decorating a context with config maps.
+	// +optional
+	configStore reconciler.ConfigStore
+
+	// reconciler is the implementation of the business logic of the resource.
+	reconciler Interface
+
+	// finalizerName is the name of the finalizer to reconcile.
+	finalizerName string
+
+	// skipStatusUpdates configures whether or not this reconciler automatically updates
+	// the status of the reconciled resource.
+	skipStatusUpdates bool
+}
+
+// Check that our Reconciler implements controller.Reconciler.
+var _ controller.Reconciler = (*reconcilerImpl)(nil)
+
+// Check that our generated Reconciler is always LeaderAware.
+var _ reconciler.LeaderAware = (*reconcilerImpl)(nil)
+
+func NewReconciler(ctx context.Context, logger *zap.SugaredLogger, client versioned.Interface, lister sourcesv1beta1.KafkaSourceLister, recorder record.EventRecorder, r Interface, options ...controller.Options) controller.Reconciler {
+	// Check the options function input. It should be 0 or 1.
+	if len(options) > 1 {
+		logger.Fatal("Up to one options struct is supported, found: ", len(options))
+	}
+
+	// Fail fast when users inadvertently implement the other LeaderAware interface.
+	// For the typed reconcilers, Promote shouldn't take any arguments.
+	if _, ok := r.(reconciler.LeaderAware); ok {
+		logger.Fatalf("%T implements the incorrect LeaderAware interface. Promote() should not take an argument as genreconciler handles the enqueuing automatically.", r)
+	}
+
+	rec := &reconcilerImpl{
+		LeaderAwareFuncs: reconciler.LeaderAwareFuncs{
+			PromoteFunc: func(bkt reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+				all, err := lister.List(labels.Everything())
+				if err != nil {
+					return err
+				}
+				for _, elt := range all {
+					// TODO: Consider letting users specify a filter in options.
+					enq(bkt, types.NamespacedName{
+						Namespace: elt.GetNamespace(),
+						Name:      elt.GetName(),
+					})
+				}
+				return nil
+			},
+		},
+		Client:        client,
+		Lister:        lister,
+		Recorder:      recorder,
+		reconciler:    r,
+		finalizerName: defaultFinalizerName,
+	}
+
+	for _, opts := range options {
+		if opts.ConfigStore != nil {
+			rec.configStore = opts.ConfigStore
+		}
+		if opts.FinalizerName != "" {
+			rec.finalizerName = opts.FinalizerName
+		}
+		if opts.SkipStatusUpdates {
+			rec.skipStatusUpdates = true
+		}
+		if opts.DemoteFunc != nil {
+			rec.DemoteFunc = opts.DemoteFunc
+		}
+	}
+
+	return rec
+}
+
+// Reconcile implements controller.Reconciler
+func (r *reconcilerImpl) Reconcile(ctx context.Context, key string) error {
+	logger := logging.FromContext(ctx)
+
+	// Initialize the reconciler state. This will convert the namespace/name
+	// string into a distinct namespace and name, determine if this instance of
+	// the reconciler is the leader, and any additional interfaces implemented
+	// by the reconciler. Returns an error is the resource key is invalid.
+	s, err := newState(key, r)
+	if err != nil {
+		logger.Error("Invalid resource key: ", key)
+		return nil
+	}
+
+	// If we are not the leader, and we don't implement either ReadOnly
+	// observer interfaces, then take a fast-path out.
+	if s.isNotLeaderNorObserver() {
+		return controller.NewSkipKey(key)
+	}
+
+	// If configStore is set, attach the frozen configuration to the context.
+	if r.configStore != nil {
+		ctx = r.configStore.ToContext(ctx)
+	}
+
+	// Add the recorder to context.
+	ctx = controller.WithEventRecorder(ctx, r.Recorder)
+
+	// Get the resource with this namespace/name.
+
+	getter := r.Lister.KafkaSources(s.namespace)
+
+	original, err := getter.Get(s.name)
+
+	if errors.IsNotFound(err) {
+		// The resource may no longer exist, in which case we stop processing and call
+		// the ObserveDeletion handler if appropriate.
+		logger.Debugf("Resource %q no longer exists", key)
+		if del, ok := r.reconciler.(reconciler.OnDeletionInterface); ok {
+			return del.ObserveDeletion(ctx, types.NamespacedName{
+				Namespace: s.namespace,
+				Name:      s.name,
+			})
+		}
+		return nil
+	} else if err != nil {
+		return err
+	}
+
+	// Don't modify the informers copy.
+	resource := original.DeepCopy()
+
+	var reconcileEvent reconciler.Event
+
+	name, do := s.reconcileMethodFor(resource)
+	// Append the target method to the logger.
+	logger = logger.With(zap.String("targetMethod", name))
+	switch name {
+	case reconciler.DoReconcileKind:
+		// Set and update the finalizer on resource if r.reconciler
+		// implements Finalizer.
+		if resource, err = r.setFinalizerIfFinalizer(ctx, resource); err != nil {
+			return fmt.Errorf("failed to set finalizers: %w", err)
+		}
+
+		if !r.skipStatusUpdates {
+			reconciler.PreProcessReconcile(ctx, resource)
+		}
+
+		// Reconcile this copy of the resource and then write back any status
+		// updates regardless of whether the reconciliation errored out.
+		reconcileEvent = do(ctx, resource)
+
+		if !r.skipStatusUpdates {
+			reconciler.PostProcessReconcile(ctx, resource, original)
+		}
+
+	case reconciler.DoFinalizeKind:
+		// For finalizing reconcilers, if this resource being marked for deletion
+		// and reconciled cleanly (nil or normal event), remove the finalizer.
+		reconcileEvent = do(ctx, resource)
+
+		if resource, err = r.clearFinalizer(ctx, resource, reconcileEvent); err != nil {
+			return fmt.Errorf("failed to clear finalizers: %w", err)
+		}
+
+	case reconciler.DoObserveKind:
+		// Observe any changes to this resource, since we are not the leader.
+		reconcileEvent = do(ctx, resource)
+
+	}
+
+	// Synchronize the status.
+	switch {
+	case r.skipStatusUpdates:
+		// This reconciler implementation is configured to skip resource updates.
+		// This may mean this reconciler does not observe spec, but reconciles external changes.
+	case equality.Semantic.DeepEqual(original.Status, resource.Status):
+		// If we didn't change anything then don't call updateStatus.
+		// This is important because the copy we loaded from the injectionInformer's
+		// cache may be stale and we don't want to overwrite a prior update
+		// to status with this stale state.
+	case !s.isLeader:
+		// High-availability reconcilers may have many replicas watching the resource, but only
+		// the elected leader is expected to write modifications.
+		logger.Warn("Saw status changes when we aren't the leader!")
+	default:
+		if err = r.updateStatus(ctx, original, resource); err != nil {
+			logger.Warnw("Failed to update resource status", zap.Error(err))
+			r.Recorder.Eventf(resource, v1.EventTypeWarning, "UpdateFailed",
+				"Failed to update status for %q: %v", resource.Name, err)
+			return err
+		}
+	}
+
+	// Report the reconciler event, if any.
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			logger.Infow("Returned an event", zap.Any("event", reconcileEvent))
+			r.Recorder.Event(resource, event.EventType, event.Reason, event.Error())
+
+			// the event was wrapped inside an error, consider the reconciliation as failed
+			if _, isEvent := reconcileEvent.(*reconciler.ReconcilerEvent); !isEvent {
+				return reconcileEvent
+			}
+			return nil
+		}
+
+		if controller.IsSkipKey(reconcileEvent) {
+			// This is a wrapped error, don't emit an event.
+		} else if ok, _ := controller.IsRequeueKey(reconcileEvent); ok {
+			// This is a wrapped error, don't emit an event.
+		} else {
+			logger.Errorw("Returned an error", zap.Error(reconcileEvent))
+			r.Recorder.Event(resource, v1.EventTypeWarning, "InternalError", reconcileEvent.Error())
+		}
+		return reconcileEvent
+	}
+
+	return nil
+}
+
+func (r *reconcilerImpl) updateStatus(ctx context.Context, existing *v1beta1.KafkaSource, desired *v1beta1.KafkaSource) error {
+	existing = existing.DeepCopy()
+	return reconciler.RetryUpdateConflicts(func(attempts int) (err error) {
+		// The first iteration tries to use the injectionInformer's state, subsequent attempts fetch the latest state via API.
+		if attempts > 0 {
+
+			getter := r.Client.SourcesV1beta1().KafkaSources(desired.Namespace)
+
+			existing, err = getter.Get(ctx, desired.Name, metav1.GetOptions{})
+			if err != nil {
+				return err
+			}
+		}
+
+		// If there's nothing to update, just return.
+		if equality.Semantic.DeepEqual(existing.Status, desired.Status) {
+			return nil
+		}
+
+		if diff, err := kmp.SafeDiff(existing.Status, desired.Status); err == nil && diff != "" {
+			logging.FromContext(ctx).Debug("Updating status with: ", diff)
+		}
+
+		existing.Status = desired.Status
+
+		updater := r.Client.SourcesV1beta1().KafkaSources(existing.Namespace)
+
+		_, err = updater.UpdateStatus(ctx, existing, metav1.UpdateOptions{})
+		return err
+	})
+}
+
+// updateFinalizersFiltered will update the Finalizers of the resource.
+// TODO: this method could be generic and sync all finalizers. For now it only
+// updates defaultFinalizerName or its override.
+func (r *reconcilerImpl) updateFinalizersFiltered(ctx context.Context, resource *v1beta1.KafkaSource) (*v1beta1.KafkaSource, error) {
+
+	getter := r.Lister.KafkaSources(resource.Namespace)
+
+	actual, err := getter.Get(resource.Name)
+	if err != nil {
+		return resource, err
+	}
+
+	// Don't modify the informers copy.
+	existing := actual.DeepCopy()
+
+	var finalizers []string
+
+	// If there's nothing to update, just return.
+	existingFinalizers := sets.NewString(existing.Finalizers...)
+	desiredFinalizers := sets.NewString(resource.Finalizers...)
+
+	if desiredFinalizers.Has(r.finalizerName) {
+		if existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Add the finalizer.
+		finalizers = append(existing.Finalizers, r.finalizerName)
+	} else {
+		if !existingFinalizers.Has(r.finalizerName) {
+			// Nothing to do.
+			return resource, nil
+		}
+		// Remove the finalizer.
+		existingFinalizers.Delete(r.finalizerName)
+		finalizers = existingFinalizers.List()
+	}
+
+	mergePatch := map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"finalizers":      finalizers,
+			"resourceVersion": existing.ResourceVersion,
+		},
+	}
+
+	patch, err := json.Marshal(mergePatch)
+	if err != nil {
+		return resource, err
+	}
+
+	patcher := r.Client.SourcesV1beta1().KafkaSources(resource.Namespace)
+
+	resourceName := resource.Name
+	updated, err := patcher.Patch(ctx, resourceName, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		r.Recorder.Eventf(existing, v1.EventTypeWarning, "FinalizerUpdateFailed",
+			"Failed to update finalizers for %q: %v", resourceName, err)
+	} else {
+		r.Recorder.Eventf(updated, v1.EventTypeNormal, "FinalizerUpdate",
+			"Updated %q finalizers", resource.GetName())
+	}
+	return updated, err
+}
+
+func (r *reconcilerImpl) setFinalizerIfFinalizer(ctx context.Context, resource *v1beta1.KafkaSource) (*v1beta1.KafkaSource, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	// If this resource is not being deleted, mark the finalizer.
+	if resource.GetDeletionTimestamp().IsZero() {
+		finalizers.Insert(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
+
+func (r *reconcilerImpl) clearFinalizer(ctx context.Context, resource *v1beta1.KafkaSource, reconcileEvent reconciler.Event) (*v1beta1.KafkaSource, error) {
+	if _, ok := r.reconciler.(Finalizer); !ok {
+		return resource, nil
+	}
+	if resource.GetDeletionTimestamp().IsZero() {
+		return resource, nil
+	}
+
+	finalizers := sets.NewString(resource.Finalizers...)
+
+	if reconcileEvent != nil {
+		var event *reconciler.ReconcilerEvent
+		if reconciler.EventAs(reconcileEvent, &event) {
+			if event.EventType == v1.EventTypeNormal {
+				finalizers.Delete(r.finalizerName)
+			}
+		}
+	} else {
+		finalizers.Delete(r.finalizerName)
+	}
+
+	resource.Finalizers = finalizers.List()
+
+	// Synchronize the finalizers filtered by r.finalizerName.
+	return r.updateFinalizersFiltered(ctx, resource)
+}
diff --git a/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go
new file mode 100644
index 000000000..3f5c5c64d
--- /dev/null
+++ b/vendor/knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource/state.go
@@ -0,0 +1,97 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Code generated by injection-gen. DO NOT EDIT.
+
+package kafkasource
+
+import (
+	fmt "fmt"
+
+	types "k8s.io/apimachinery/pkg/types"
+	cache "k8s.io/client-go/tools/cache"
+	v1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+	reconciler "knative.dev/pkg/reconciler"
+)
+
+// state is used to track the state of a reconciler in a single run.
+type state struct {
+	// key is the original reconciliation key from the queue.
+	key string
+	// namespace is the namespace split from the reconciliation key.
+	namespace string
+	// name is the name split from the reconciliation key.
+	name string
+	// reconciler is the reconciler.
+	reconciler Interface
+	// roi is the read only interface cast of the reconciler.
+	roi ReadOnlyInterface
+	// isROI (Read Only Interface) the reconciler only observes reconciliation.
+	isROI bool
+	// isLeader the instance of the reconciler is the elected leader.
+	isLeader bool
+}
+
+func newState(key string, r *reconcilerImpl) (*state, error) {
+	// Convert the namespace/name string into a distinct namespace and name.
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		return nil, fmt.Errorf("invalid resource key: %s", key)
+	}
+
+	roi, isROI := r.reconciler.(ReadOnlyInterface)
+
+	isLeader := r.IsLeaderFor(types.NamespacedName{
+		Namespace: namespace,
+		Name:      name,
+	})
+
+	return &state{
+		key:        key,
+		namespace:  namespace,
+		name:       name,
+		reconciler: r.reconciler,
+		roi:        roi,
+		isROI:      isROI,
+		isLeader:   isLeader,
+	}, nil
+}
+
+// isNotLeaderNorObserver checks to see if this reconciler with the current
+// state is enabled to do any work or not.
+// isNotLeaderNorObserver returns true when there is no work possible for the
+// reconciler.
+func (s *state) isNotLeaderNorObserver() bool {
+	if !s.isLeader && !s.isROI {
+		// If we are not the leader, and we don't implement the ReadOnly
+		// interface, then take a fast-path out.
+		return true
+	}
+	return false
+}
+
+func (s *state) reconcileMethodFor(o *v1beta1.KafkaSource) (string, doReconcile) {
+	if o.GetDeletionTimestamp().IsZero() {
+		if s.isLeader {
+			return reconciler.DoReconcileKind, s.reconciler.ReconcileKind
+		} else if s.isROI {
+			return reconciler.DoObserveKind, s.roi.ObserveKind
+		}
+	} else if fin, ok := s.reconciler.(Finalizer); s.isLeader && ok {
+		return reconciler.DoFinalizeKind, fin.FinalizeKind
+	}
+	return "unknown", nil
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index e483a421b..bec02ec65 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1028,7 +1028,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.26.1-0.20211006164845-481fd3e3331b
+# knative.dev/eventing v0.26.1-0.20211011151816-4bf0a232f3e0
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck
@@ -1136,7 +1136,7 @@ knative.dev/eventing/test/test_images/performance
 knative.dev/eventing/test/test_images/print
 knative.dev/eventing/test/test_images/recordevents
 knative.dev/eventing/test/test_images/request-sender
-# knative.dev/eventing-kafka v0.26.1-0.20211005035853-9fe562d91403
+# knative.dev/eventing-kafka v0.26.1-0.20211011084516-a8f8dec33c11
 ## explicit
 knative.dev/eventing-kafka/pkg/apis/bindings
 knative.dev/eventing-kafka/pkg/apis/bindings/v1beta1
@@ -1149,11 +1149,16 @@ knative.dev/eventing-kafka/pkg/apis/sources/config
 knative.dev/eventing-kafka/pkg/apis/sources/v1beta1
 knative.dev/eventing-kafka/pkg/channel/distributed/common/kafka/constants
 knative.dev/eventing-kafka/pkg/client/clientset/versioned
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/scheme
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/bindings/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/kafka/v1alpha1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/messaging/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1
+knative.dev/eventing-kafka/pkg/client/clientset/versioned/typed/sources/v1beta1/fake
 knative.dev/eventing-kafka/pkg/client/informers/externalversions
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/bindings/v1beta1
@@ -1165,9 +1170,14 @@ knative.dev/eventing-kafka/pkg/client/informers/externalversions/messaging/v1bet
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources
 knative.dev/eventing-kafka/pkg/client/informers/externalversions/sources/v1beta1
 knative.dev/eventing-kafka/pkg/client/injection/client
+knative.dev/eventing-kafka/pkg/client/injection/client/fake
 knative.dev/eventing-kafka/pkg/client/injection/informers/factory
+knative.dev/eventing-kafka/pkg/client/injection/informers/factory/fake
 knative.dev/eventing-kafka/pkg/client/injection/informers/messaging/v1beta1/kafkachannel
+knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource
+knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake
 knative.dev/eventing-kafka/pkg/client/injection/reconciler/messaging/v1beta1/kafkachannel
+knative.dev/eventing-kafka/pkg/client/injection/reconciler/sources/v1beta1/kafkasource
 knative.dev/eventing-kafka/pkg/client/listers/bindings/v1beta1
 knative.dev/eventing-kafka/pkg/client/listers/kafka/v1alpha1
 knative.dev/eventing-kafka/pkg/client/listers/messaging/v1beta1


Test patch:
diff --git a/.github/workflows/knative-go-test.yaml b/.github/workflows/knative-go-test.yaml
index 2c892f18d..14f2941d1 100644
--- a/.github/workflows/knative-go-test.yaml
+++ b/.github/workflows/knative-go-test.yaml
@@ -52,6 +52,12 @@ jobs:
       - name: Merge upstream
         if: github.event_name == 'pull_request'
         run: |
+          if ! git config user.name > /dev/null; then
+            git config user.name "John Doe"
+          fi
+          if ! git config user.email > /dev/null; then
+            git config user.email "johndoe@localhost"
+          fi
           git remote add upstream https://github.com/${{ github.repository }}.git
           git fetch upstream ${{ github.base_ref }}
           git pull --no-rebase --no-commit upstream ${{ github.base_ref }}
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 5de321403..55a8f5481 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -85,7 +85,7 @@ var (
 )
 
 func TestBrokerReconciler(t *testing.T) {
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index a89531352..8e5365dbc 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -36,7 +36,7 @@ import (
 	. "knative.dev/pkg/reconciler/testing"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing"
-	v1alpha1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/apis/eventing/v1alpha1"
 	fakeeventingkafkaclient "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/client/fake"
 	sinkreconciler "knative.dev/eventing-kafka-broker/control-plane/pkg/client/injection/reconciler/eventing/v1alpha1/kafkasink"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/receiver"
@@ -78,7 +78,7 @@ var (
 
 func TestSinkReconciler(t *testing.T) {
 
-	v1alpha1.RegisterConditionSet(base.ConditionSet)
+	v1alpha1.RegisterConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
@@ -821,7 +821,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 
 func TestSinkFinalizer(t *testing.T) {
 
-	v1alpha1.RegisterConditionSet(base.ConditionSet)
+	v1alpha1.RegisterConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/control-plane/pkg/reconciler/source/controller_test.go b/control-plane/pkg/reconciler/source/controller_test.go
new file mode 100644
index 000000000..20dfc6c67
--- /dev/null
+++ b/control-plane/pkg/reconciler/source/controller_test.go
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2021 The Knative Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package source
+
+import (
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakekubeclientset "k8s.io/client-go/kubernetes/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker/fake"
+	_ "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/trigger/fake"
+	_ "knative.dev/pkg/client/injection/ducks/duck/v1/addressable/fake"
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/pod/fake"
+	_ "knative.dev/pkg/client/injection/kube/informers/core/v1/secret/fake"
+	"knative.dev/pkg/configmap"
+	dynamicclient "knative.dev/pkg/injection/clients/dynamicclient/fake"
+	reconcilertesting "knative.dev/pkg/reconciler/testing"
+
+	_ "knative.dev/eventing-kafka/pkg/client/injection/informers/sources/v1beta1/kafkasource/fake"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/config"
+)
+
+func TestNewController(t *testing.T) {
+	ctx, _ := reconcilertesting.SetupFakeContext(t)
+
+	configs := &config.Env{
+		SystemNamespace:      "cm",
+		GeneralConfigMapName: "cm",
+	}
+
+	ctx, _ = fakekubeclient.With(
+		ctx,
+		&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configs.GeneralConfigMapName,
+				Namespace: configs.SystemNamespace,
+			},
+		},
+	)
+	dynamicScheme := runtime.NewScheme()
+	_ = fakekubeclientset.AddToScheme(dynamicScheme)
+
+	dynamicclient.With(ctx, dynamicScheme)
+
+	controller := NewController(
+		ctx,
+		configmap.NewStaticWatcher(&corev1.ConfigMap{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "cm",
+			},
+		}),
+		configs,
+	)
+	if controller == nil {
+		t.Error("failed to create controller: <nil>")
+	}
+}
diff --git a/control-plane/pkg/reconciler/trigger/trigger_test.go b/control-plane/pkg/reconciler/trigger/trigger_test.go
index 6733433ea..d30fc0d35 100644
--- a/control-plane/pkg/reconciler/trigger/trigger_test.go
+++ b/control-plane/pkg/reconciler/trigger/trigger_test.go
@@ -77,7 +77,7 @@ var (
 )
 
 func TestTriggerReconciler(t *testing.T) {
-	eventing.RegisterAlternateBrokerConditionSet(base.ConditionSet)
+	eventing.RegisterAlternateBrokerConditionSet(base.IngressConditionSet)
 
 	t.Parallel()
 
diff --git a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
index 969a184b8..a1f8317e4 100644
--- a/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
+++ b/data-plane/dispatcher/src/test/java/dev/knative/eventing/kafka/broker/dispatcher/main/ConsumerVerticleFactoryImplTest.java
@@ -28,6 +28,8 @@ import io.vertx.ext.web.client.WebClientOptions;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.backends.BackendRegistries;
 import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
 import org.junit.jupiter.api.Test;
@@ -181,4 +183,37 @@ public class ConsumerVerticleFactoryImplTest {
 
     assertThat(delay).isEqualTo(0);
   }
+
+  @Test
+  public void getNoopResponseHandler() {
+    final var counter = new AtomicInteger(0);
+    final var egress = DataPlaneContract.Egress.newBuilder()
+      .setDiscardReply(DataPlaneContract.Empty.newBuilder().build())
+      .build();
+
+    final var r = ConsumerVerticleFactoryImpl.getNoopResponseHandlerOrDefault(egress, () -> {
+      counter.incrementAndGet();
+      return null;
+    });
+
+    assertThat(r).isNotNull();
+    assertThat(counter.get()).isEqualTo(0);
+  }
+
+  @Test
+  public void getKafkaResponseHandler() {
+    final var counter = new AtomicInteger(0);
+
+    final var egress = DataPlaneContract.Egress.newBuilder()
+      .setReplyToOriginalTopic(DataPlaneContract.Empty.newBuilder().build())
+      .build();
+
+    final var r = ConsumerVerticleFactoryImpl.getNoopResponseHandlerOrDefault(egress, () -> {
+      counter.incrementAndGet();
+      return null;
+    });
+
+    assertThat(r).isNull();
+    assertThat(counter.get()).isEqualTo(1);
+  }
 }
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index d21451ae1..cdb86e904 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -26,10 +26,9 @@ fi
 
 if ! ${SKIP_INITIALIZE}; then
   initialize $@ --skip-istio-addon
+  save_release_artifacts || fail_test "Failed to save release artifacts"
 fi
 
-save_release_artifacts || fail_test "Failed to save release artifacts"
-
 if ! ${LOCAL_DEVELOPMENT}; then
   scale_controlplane kafka-controller kafka-webhook-eventing eventing-webhook eventing-controller
   apply_sacura || fail_test "Failed to apply Sacura"
diff --git a/third_party/eventing-kafka-latest/channel-crds.yaml b/third_party/eventing-kafka-latest/channel-crds.yaml
index f80de6871..9dc621df0 100644
--- a/third_party/eventing-kafka-latest/channel-crds.yaml
+++ b/third_party/eventing-kafka-latest/channel-crds.yaml
@@ -17,7 +17,7 @@ kind: CustomResourceDefinition
 metadata:
   name: kafkachannels.messaging.knative.dev
   labels:
-    kafka.eventing.knative.dev/release: "v20211005-9fe562d91"
+    kafka.eventing.knative.dev/release: "v20211012-a8f8dec33"
     knative.dev/crd-install: "true"
     messaging.knative.dev/subscribable: "true"
     duck.knative.dev/addressable: "true"
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
index 99ad8d18b..057429301 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/receiver/receiver.go
@@ -123,8 +123,13 @@ func (o *Receiver) Start(ctx context.Context, handlerFuncs ...func(handler http.
 	for _, dec := range handlerFuncs {
 		handler = dec(handler)
 	}
+	mux := http.NewServeMux()
+	mux.Handle("/", handler)
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+	})
 
-	server := &http.Server{Addr: ":8080", Handler: handler}
+	server := &http.Server{Addr: fmt.Sprintf(":%d", recordevents.EventRecordReceivePort), Handler: mux}
 
 	var err error
 	go func() {
diff --git a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
index 62e0a4a89..a1bc4b0ee 100644
--- a/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
+++ b/vendor/knative.dev/eventing/test/lib/recordevents/resources.go
@@ -23,12 +23,15 @@ import (
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	pkgtest "knative.dev/pkg/test"
 
 	testlib "knative.dev/eventing/test/lib"
 )
 
+const EventRecordReceivePort = 8080
+
 func noop(pod *corev1.Pod, client *testlib.Client) error {
 	return nil
 }
@@ -49,7 +52,8 @@ func DeployEventRecordOrFail(ctx context.Context, client *testlib.Client, name s
 		envOption("EVENT_GENERATORS", "receiver"),
 	)
 
-	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace)
+	readinessProbe := true
+	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace, readinessProbe)
 	client.CreatePodOrFail(eventRecordPod, options...)
 	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
@@ -67,7 +71,8 @@ func DeployEventSenderOrFail(ctx context.Context, client *testlib.Client, name s
 		envOption("SINK", sink),
 	)
 
-	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace)
+	readinessProbe := false // this is not needed for sender pod
+	eventRecordPod := recordEventsPod("recordevents", name, client.Namespace, readinessProbe)
 	client.CreatePodOrFail(eventRecordPod, options...)
 	err := pkgtest.WaitForPodRunning(ctx, client.Kube, name, client.Namespace)
 	if err != nil {
@@ -76,8 +81,8 @@ func DeployEventSenderOrFail(ctx context.Context, client *testlib.Client, name s
 	return eventRecordPod
 }
 
-func recordEventsPod(imageName string, name string, serviceAccountName string) *corev1.Pod {
-	return &corev1.Pod{
+func recordEventsPod(imageName string, name string, serviceAccountName string, readinessProbe bool) *corev1.Pod {
+	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:   name,
 			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
@@ -101,9 +106,26 @@ func recordEventsPod(imageName string, name string, serviceAccountName string) *
 					Name:  "EVENT_LOGS",
 					Value: "recorder,logger",
 				}},
+				Ports: []corev1.ContainerPort{
+					{
+						Name:          "receive",
+						ContainerPort: EventRecordReceivePort,
+					},
+				},
 			}},
 			ServiceAccountName: serviceAccountName,
 			RestartPolicy:      corev1.RestartPolicyAlways,
 		},
 	}
+	if readinessProbe {
+		pod.Spec.Containers[0].ReadinessProbe = &corev1.Probe{
+			Handler: corev1.Handler{
+				HTTPGet: &corev1.HTTPGetAction{
+					Port: intstr.FromString("receive"),
+					Path: "/healthz",
+				},
+			},
+		}
+	}
+	return pod
 }