Repository: knative-sandbox/eventing-kafka-broker
Pull request #1021 (2021-06-21 07:07:16)

Add single request timeout to the contract
Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

This adds the support for the experimental feature `delivery-timeout`: https://github.com/knative/eventing/issues/5148

Also part of what was discussed in #757

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- :gift: Add `EgressConfig.timeout` to configure the single request timeout

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

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

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

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
matzew: Process is:
1. edit the `proto/contract.proto`
2. call the `generate-proto.sh` script (which updates the two generated files)
3. manual update the factory here, to make use of the timeout

 Right?
slinkydeveloper: That's exactly what I did for this PR

Patch:
diff --git a/control-plane/pkg/contract/contract.pb.go b/control-plane/pkg/contract/contract.pb.go
index 255782c9f..807da7381 100644
--- a/control-plane/pkg/contract/contract.pb.go
+++ b/control-plane/pkg/contract/contract.pb.go
@@ -275,6 +275,8 @@ type EgressConfig struct {
 	BackoffPolicy BackoffPolicy `protobuf:"varint,3,opt,name=backoffPolicy,proto3,enum=BackoffPolicy" json:"backoffPolicy,omitempty"`
 	// backoffDelay is the delay before retrying in milliseconds.
 	BackoffDelay uint64 `protobuf:"varint,4,opt,name=backoffDelay,proto3" json:"backoffDelay,omitempty"`
+	// timeout is the single request timeout (not the overall retry timeout)
+	Timeout uint64 `protobuf:"varint,5,opt,name=timeout,proto3" json:"timeout,omitempty"`
 }
 
 func (x *EgressConfig) Reset() {
@@ -337,6 +339,13 @@ func (x *EgressConfig) GetBackoffDelay() uint64 {
 	return 0
 }
 
+func (x *EgressConfig) GetTimeout() uint64 {
+	if x != nil {
+		return x.Timeout
+	}
+	return 0
+}
+
 type Egress struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -885,7 +894,7 @@ var file_contract_proto_rawDesc = []byte{
 	0x0f, 0x41, 0x74, 0x74, 0x72, 0x69, 0x62, 0x75, 0x74, 0x65, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79,
 	0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b,
 	0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28,
-	0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0x9e, 0x01, 0x0a,
+	0x09, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xb8, 0x01, 0x0a,
 	0x0c, 0x45, 0x67, 0x72, 0x65, 0x73, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1e, 0x0a,
 	0x0a, 0x64, 0x65, 0x61, 0x64, 0x4c, 0x65, 0x74, 0x74, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28,
 	0x09, 0x52, 0x0a, 0x64, 0x65, 0x61, 0x64, 0x4c, 0x65, 0x74, 0x74, 0x65, 0x72, 0x12, 0x14, 0x0a,
@@ -895,83 +904,85 @@ var file_contract_proto_rawDesc = []byte{
 	0x6b, 0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x52, 0x0d, 0x62, 0x61, 0x63, 0x6b,
 	0x6f, 0x66, 0x66, 0x50, 0x6f, 0x6c, 0x69, 0x63, 0x79, 0x12, 0x22, 0x0a, 0x0c, 0x62, 0x61, 0x63,
 	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, 0x22, 0xd9, 0x02,
-	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, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20,
-	0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x74, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e,
-	0x12, 0x1c, 0x0a, 0x08, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x55, 0x72, 0x6c, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x09, 0x48, 0x00, 0x52, 0x08, 0x72, 0x65, 0x70, 0x6c, 0x79, 0x55, 0x72, 0x6c, 0x12, 0x3c,
-	0x0a, 0x14, 0x72, 0x65, 0x70, 0x6c, 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,
+	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,
+	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,
+	0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64,
+	0x65, 0x73, 0x74, 0x69, 0x6e, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1c, 0x0a, 0x08, 0x72, 0x65,
+	0x70, 0x6c, 0x79, 0x55, 0x72, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x08,
+	0x72, 0x65, 0x70, 0x6c, 0x79, 0x55, 0x72, 0x6c, 0x12, 0x3c, 0x0a, 0x14, 0x72, 0x65, 0x70, 0x6c,
+	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,
 }
 
 var (
diff --git a/control-plane/pkg/reconciler/base/reconciler.go b/control-plane/pkg/reconciler/base/reconciler.go
index f8a0bc2b9..d60b53057 100644
--- a/control-plane/pkg/reconciler/base/reconciler.go
+++ b/control-plane/pkg/reconciler/base/reconciler.go
@@ -19,6 +19,7 @@ package base
 import (
 	"context"
 	"fmt"
+	"strconv"
 
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/encoding/protojson"
@@ -29,7 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	corelisters "k8s.io/client-go/listers/core/v1"
-	"k8s.io/client-go/util/retry"
 	"knative.dev/pkg/tracker"
 
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/contract"
@@ -196,31 +196,19 @@ func (r *Reconciler) UpdateDataPlaneConfigMap(ctx context.Context, contract *con
 }
 
 func (r *Reconciler) UpdateDispatcherPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-
-	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-
-		labelSelector := r.dispatcherSelector()
-		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
-		if errors != nil {
-			return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
-		}
-
-		return r.updatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
-	})
+	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.dispatcherSelector())
+	if errors != nil {
+		return fmt.Errorf("failed to list dispatcher pods in namespace %s: %w", r.SystemNamespace, errors)
+	}
+	return r.updatePodsAnnotation(ctx, logger, "dispatcher", volumeGeneration, pods)
 }
 
 func (r *Reconciler) UpdateReceiverPodsAnnotation(ctx context.Context, logger *zap.Logger, volumeGeneration uint64) error {
-
-	return retry.RetryOnConflict(retry.DefaultBackoff, func() error {
-
-		labelSelector := r.receiverSelector()
-		pods, errors := r.PodLister.Pods(r.SystemNamespace).List(labelSelector)
-		if errors != nil {
-			return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
-		}
-
-		return r.updatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
-	})
+	pods, errors := r.PodLister.Pods(r.SystemNamespace).List(r.receiverSelector())
+	if errors != nil {
+		return fmt.Errorf("failed to list receiver pods in namespace %s: %w", r.SystemNamespace, errors)
+	}
+	return r.updatePodsAnnotation(ctx, logger, "receiver", volumeGeneration, pods)
 }
 
 func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logger, component string, volumeGeneration uint64, pods []*corev1.Pod) error {
@@ -243,6 +231,15 @@ func (r *Reconciler) updatePodsAnnotation(ctx context.Context, logger *zap.Logge
 			annotations = make(map[string]string, 1)
 		}
 
+		// Check whether pod's annotation is the expected one.
+		if v, ok := annotations[VolumeGenerationAnnotationKey]; ok {
+			v, err := strconv.ParseUint(v /* base */, 10 /* bitSize */, 64)
+			if err == nil && v == volumeGeneration {
+				// Volume generation already matches the expected volume generation number.
+				continue
+			}
+		}
+
 		annotations[VolumeGenerationAnnotationKey] = fmt.Sprint(volumeGeneration)
 		pod.SetAnnotations(annotations)
 
diff --git a/control-plane/pkg/reconciler/broker/broker.go b/control-plane/pkg/reconciler/broker/broker.go
index 989019c0c..cc4c87cb5 100644
--- a/control-plane/pkg/reconciler/broker/broker.go
+++ b/control-plane/pkg/reconciler/broker/broker.go
@@ -145,10 +145,7 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 		return statusConditionManager.FailedToGetDataFromConfigMap(err)
 	}
 
-	logger.Debug(
-		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, ct),
-	)
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	// Get resource configuration.
 	brokerResource, err := r.getBrokerResource(ctx, topic, broker, secret, topicConfig)
@@ -162,9 +159,10 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 
 	logger.Debug("Change detector", zap.Int("changed", changed))
 
-	coreconfig.IncrementContractGeneration(ct)
-
 	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(
@@ -176,38 +174,39 @@ func (r *Reconciler) reconcileKind(ctx context.Context, broker *eventing.Broker)
 	}
 	statusConditionManager.ConfigMapUpdated()
 
-	if changed == coreconfig.ResourceChanged {
+	// We update receiver and 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.
+
+	// We reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
+	// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
+	// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
+	// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
+	// the update even if here eventually means seconds or minutes after the actual update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		logger.Error("Failed to update receiver pod annotation", zap.Error(
+			statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
+		))
+		return err
+	}
 
-		// After #37 we reject events to a non-existing Broker, which means that we cannot consider a Broker Ready if all
-		// receivers haven't got the Broker, so update failures to receiver pods is a hard failure.
-		// On the other side, dispatcher pods care about Triggers, and the Broker object is used as a configuration
-		// prototype for all associated Triggers, so we consider that it's fine on the dispatcher side to receive eventually
-		// the update even if here eventually means seconds or minutes after the actual update.
+	logger.Debug("Updated receiver pod annotation")
 
-		// Update volume generation annotation of receiver pods
-		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			logger.Error("Failed to update receiver pod annotation", zap.Error(
-				statusConditionManager.FailedToUpdateReceiverPodsAnnotation(err),
-			))
-			return err
-		}
+	// 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),
+		)
 
-		logger.Debug("Updated receiver pod annotation")
-
-		// 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")
-		}
+		statusConditionManager.FailedToUpdateDispatcherPodsAnnotation(err)
+	} else {
+		logger.Debug("Updated dispatcher pod annotation")
 	}
 
 	return statusConditionManager.Reconciled()
@@ -236,10 +235,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 		return fmt.Errorf("failed to get contract: %w", err)
 	}
 
-	logger.Debug(
-		"Got contract data from config map",
-		zap.Any(base.ContractLogKey, ct),
-	)
+	logger.Debug("Got contract data from config map", zap.Any(base.ContractLogKey, ct))
 
 	brokerIndex := coreconfig.FindResource(ct, broker.UID)
 	if brokerIndex != coreconfig.NoResource {
@@ -247,19 +243,37 @@ func (r *Reconciler) finalizeKind(ctx context.Context, broker *eventing.Broker)
 
 		logger.Debug("Broker deleted", zap.Int("index", brokerIndex))
 
+		// 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 {
 			return err
 		}
-
 		logger.Debug("Contract config map updated")
+	}
+
+	// We update receiver and 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 update anyway remaining annotations with the contract generation that was saved in the CM.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
 
-		// There is no need to update volume generation and dispatcher pod annotation, updates to the config map will
-		// eventually be seen by the dispatcher pod and resources will be deleted accordingly.
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+	// Update volume generation annotation of dispatcher pods
+	if err := r.UpdateDispatcherPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
 	}
 
+	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
+	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
+	//  because topic metadata aren't available anymore.
+	// 	See (under discussions KIPs, unlikely to be accepted as they are):
+	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
+	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+
 	topicConfig, brokerConfig, err := r.topicConfig(logger, broker)
 	if err != nil {
 		return fmt.Errorf("failed to resolve broker config: %w", err)
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink.go b/control-plane/pkg/reconciler/sink/kafka_sink.go
index 0a4ebf6cd..0ff56cda0 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink.go
@@ -166,9 +166,9 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 	// Update contract data with the new sink configuration.
 	changed := coreconfig.AddOrUpdateResourceConfig(ct, sinkConfig, sinkIndex, logger)
 
-	coreconfig.IncrementContractGeneration(ct)
-
 	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(
@@ -181,18 +181,21 @@ func (r *Reconciler) reconcileKind(ctx context.Context, ks *eventing.KafkaSink)
 
 	logger.Debug("Config map updated")
 
-	if changed == coreconfig.ResourceChanged {
-		// After #37 we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
-		// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
+	// We update receiver 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.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
 
-		// Update volume generation annotation of receiver pods
-		if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
-			return err
-		}
+	// Since we reject events to a non-existing Sink, which means that we cannot consider a Sink Ready if all
+	// receivers haven't got the Sink, so update failures to receiver pods is a hard failure.
 
-		logger.Debug("Updated receiver pod annotation")
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
 	}
 
+	logger.Debug("Updated receiver pod annotation")
+
 	return statusConditionManager.Reconciled()
 }
 
@@ -230,6 +233,7 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 
 		logger.Debug("Sink deleted", zap.Int("index", sinkIndex))
 
+		// Resource changed, increment contract generation.
 		coreconfig.IncrementContractGeneration(ct)
 
 		// Update the configuration map with the new contract data.
@@ -240,6 +244,23 @@ func (r *Reconciler) finalizeKind(ctx context.Context, ks *eventing.KafkaSink) e
 		logger.Debug("Sinks config map updated")
 	}
 
+	// We update receiver 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.
+	// Note: if there aren't changes to be done at the pod annotation level, we just skip the update.
+
+	// Update volume generation annotation of receiver pods
+	if err := r.UpdateReceiverPodsAnnotation(ctx, logger, ct.Generation); err != nil {
+		return err
+	}
+
+	// TODO probe (as in #974) and check if status code is 404 otherwise requeue and return.
+	//  Rationale: after deleting a topic closing a producer ends up blocking and requesting metadata for max.block.ms
+	//  because topic metadata aren't available anymore.
+	// 	See (under discussions KIPs, unlikely to be accepted as they are):
+	// 	- https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181306446
+	// 	- https://cwiki.apache.org/confluence/display/KAFKA/KIP-286%3A+producer.send%28%29+should+not+block+on+metadata+update
+
 	if ks.GetStatus().Annotations[base.TopicOwnerAnnotation] == ControllerTopicOwner {
 		securityOption, _, err := security.NewOptionFromSecret(ctx, &SecretLocator{KafkaSink: ks}, r.SecretProviderFunc())
 		if err != nil {
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 a02060d33..4670d1400 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
@@ -1730,6 +1730,16 @@ public final class DataPlaneContract {
      * @return The backoffDelay.
      */
     long getBackoffDelay();
+
+    /**
+     * <pre>
+     * timeout is the single request timeout (not the overall retry timeout)
+     * </pre>
+     *
+     * <code>uint64 timeout = 5;</code>
+     * @return The timeout.
+     */
+    long getTimeout();
   }
   /**
    * Protobuf type {@code EgressConfig}
@@ -1800,6 +1810,11 @@ public final class DataPlaneContract {
               backoffDelay_ = input.readUInt64();
               break;
             }
+            case 40: {
+
+              timeout_ = input.readUInt64();
+              break;
+            }
             default: {
               if (!parseUnknownField(
                   input, unknownFields, extensionRegistry, tag)) {
@@ -1937,6 +1952,21 @@ public final class DataPlaneContract {
       return backoffDelay_;
     }
 
+    public static final int TIMEOUT_FIELD_NUMBER = 5;
+    private long timeout_;
+    /**
+     * <pre>
+     * timeout is the single request timeout (not the overall retry timeout)
+     * </pre>
+     *
+     * <code>uint64 timeout = 5;</code>
+     * @return The timeout.
+     */
+    @java.lang.Override
+    public long getTimeout() {
+      return timeout_;
+    }
+
     private byte memoizedIsInitialized = -1;
     @java.lang.Override
     public final boolean isInitialized() {
@@ -1963,6 +1993,9 @@ public final class DataPlaneContract {
       if (backoffDelay_ != 0L) {
         output.writeUInt64(4, backoffDelay_);
       }
+      if (timeout_ != 0L) {
+        output.writeUInt64(5, timeout_);
+      }
       unknownFields.writeTo(output);
     }
 
@@ -1987,6 +2020,10 @@ public final class DataPlaneContract {
         size += com.google.protobuf.CodedOutputStream
           .computeUInt64Size(4, backoffDelay_);
       }
+      if (timeout_ != 0L) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(5, timeout_);
+      }
       size += unknownFields.getSerializedSize();
       memoizedSize = size;
       return size;
@@ -2009,6 +2046,8 @@ public final class DataPlaneContract {
       if (backoffPolicy_ != other.backoffPolicy_) return false;
       if (getBackoffDelay()
           != other.getBackoffDelay()) return false;
+      if (getTimeout()
+          != other.getTimeout()) return false;
       if (!unknownFields.equals(other.unknownFields)) return false;
       return true;
     }
@@ -2029,6 +2068,9 @@ public final class DataPlaneContract {
       hash = (37 * hash) + BACKOFFDELAY_FIELD_NUMBER;
       hash = (53 * hash) + com.google.protobuf.Internal.hashLong(
           getBackoffDelay());
+      hash = (37 * hash) + TIMEOUT_FIELD_NUMBER;
+      hash = (53 * hash) + com.google.protobuf.Internal.hashLong(
+          getTimeout());
       hash = (29 * hash) + unknownFields.hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -2170,6 +2212,8 @@ public final class DataPlaneContract {
 
         backoffDelay_ = 0L;
 
+        timeout_ = 0L;
+
         return this;
       }
 
@@ -2200,6 +2244,7 @@ public final class DataPlaneContract {
         result.retry_ = retry_;
         result.backoffPolicy_ = backoffPolicy_;
         result.backoffDelay_ = backoffDelay_;
+        result.timeout_ = timeout_;
         onBuilt();
         return result;
       }
@@ -2261,6 +2306,9 @@ public final class DataPlaneContract {
         if (other.getBackoffDelay() != 0L) {
           setBackoffDelay(other.getBackoffDelay());
         }
+        if (other.getTimeout() != 0L) {
+          setTimeout(other.getTimeout());
+        }
         this.mergeUnknownFields(other.unknownFields);
         onChanged();
         return this;
@@ -2551,6 +2599,49 @@ public final class DataPlaneContract {
         onChanged();
         return this;
       }
+
+      private long timeout_ ;
+      /**
+       * <pre>
+       * timeout is the single request timeout (not the overall retry timeout)
+       * </pre>
+       *
+       * <code>uint64 timeout = 5;</code>
+       * @return The timeout.
+       */
+      @java.lang.Override
+      public long getTimeout() {
+        return timeout_;
+      }
+      /**
+       * <pre>
+       * timeout is the single request timeout (not the overall retry timeout)
+       * </pre>
+       *
+       * <code>uint64 timeout = 5;</code>
+       * @param value The timeout to set.
+       * @return This builder for chaining.
+       */
+      public Builder setTimeout(long value) {
+        
+        timeout_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <pre>
+       * timeout is the single request timeout (not the overall retry timeout)
+       * </pre>
+       *
+       * <code>uint64 timeout = 5;</code>
+       * @return This builder for chaining.
+       */
+      public Builder clearTimeout() {
+        
+        timeout_ = 0L;
+        onChanged();
+        return this;
+      }
       @java.lang.Override
       public final Builder setUnknownFields(
           final com.google.protobuf.UnknownFieldSet unknownFields) {
@@ -11274,33 +11365,34 @@ public final class DataPlaneContract {
       "\n\016contract.proto\"\007\n\005Empty\"h\n\006Filter\022+\n\na" +
       "ttributes\030\001 \003(\0132\027.Filter.AttributesEntry" +
       "\0321\n\017AttributesEntry\022\013\n\003key\030\001 \001(\t\022\r\n\005valu" +
-      "e\030\002 \001(\t:\0028\001\"n\n\014EgressConfig\022\022\n\ndeadLette" +
+      "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\"\363\001\n\006Egress\022\025\n\rconsumerGroup\030\001 \001(\t\022\023\n" +
-      "\013destination\030\002 \001(\t\022\022\n\010replyUrl\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\014" +
-      "egressConfig\030\007 \001(\0132\r.EgressConfig\022%\n\rdel" +
-      "iveryOrder\030\010 \001(\0162\016.DeliveryOrderB\017\n\rrepl" +
-      "yStrategy\"[\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\004u" +
-      "uid\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\020bootstrapServer" +
-      "s\030\003 \001(\t\022\031\n\007ingress\030\004 \001(\0132\010.Ingress\022#\n\014eg" +
-      "ressConfig\030\005 \001(\0132\r.EgressConfig\022\031\n\010egres" +
-      "ses\030\006 \003(\0132\007.Egress\022\034\n\nabsentAuth\030\007 \001(\0132\006" +
-      ".EmptyH\000\022 \n\nauthSecret\030\010 \001(\0132\n.Reference" +
-      "H\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\rBack" +
-      "offPolicy\022\017\n\013Exponential\020\000\022\n\n\006Linear\020\001*+" +
-      "\n\rDeliveryOrder\022\r\n\tUNORDERED\020\000\022\013\n\007ORDERE" +
-      "D\020\001*)\n\013ContentMode\022\n\n\006BINARY\020\000\022\016\n\nSTRUCT" +
-      "URED\020\001B[\n*dev.knative.eventing.kafka.bro" +
-      "ker.contractB\021DataPlaneContractZ\032control" +
-      "-plane/pkg/contractb\006proto3"
+      "\001(\004\022\017\n\007timeout\030\005 \001(\004\"\363\001\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"
     };
     descriptor = com.google.protobuf.Descriptors.FileDescriptor
       .internalBuildGeneratedFileFrom(descriptorData,
@@ -11329,7 +11421,7 @@ public final class DataPlaneContract {
     internal_static_EgressConfig_fieldAccessorTable = new
       com.google.protobuf.GeneratedMessageV3.FieldAccessorTable(
         internal_static_EgressConfig_descriptor,
-        new java.lang.String[] { "DeadLetter", "Retry", "BackoffPolicy", "BackoffDelay", });
+        new java.lang.String[] { "DeadLetter", "Retry", "BackoffPolicy", "BackoffDelay", "Timeout", });
     internal_static_Egress_descriptor =
       getDescriptor().getMessageTypes().get(3);
     internal_static_Egress_fieldAccessorTable = new
diff --git a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
index 532bf0f87..30f1b7ad9 100644
--- a/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
+++ b/data-plane/dispatcher/src/main/java/dev/knative/eventing/kafka/broker/dispatcher/http/HttpConsumerVerticleFactory.java
@@ -228,23 +228,34 @@ public class HttpConsumerVerticleFactory implements ConsumerVerticleFactory {
   }
 
   private static CircuitBreakerOptions createCircuitBreakerOptions(final DataPlaneContract.EgressConfig egressConfig) {
-    if (egressConfig != null && egressConfig.getRetry() > 0) {
-      return new CircuitBreakerOptions()
-        // TODO reset timeout should be configurable or, at least, set by the control plane
-        .setResetTimeout(
-          egressConfig.getBackoffDelay() > 0 ?
-            egressConfig.getBackoffDelay() :
-            CircuitBreakerOptions.DEFAULT_RESET_TIMEOUT
-        )
-        // TODO max failures should be configurable or, at least, set by the control plane
-        .setMaxFailures(egressConfig.getRetry() * 2)
-        .setMaxRetries(egressConfig.getRetry())
-        // This disables circuit breaker notifications on the event bus
-        .setNotificationAddress(null);
+    CircuitBreakerOptions options = new CircuitBreakerOptions();
+
+    // This disables circuit breaker notifications on the event bus
+    options.setNotificationAddress(null);
+
+    if (egressConfig != null) {
+      // Single request timeout
+      options.setTimeout(
+        egressConfig.getTimeout() > 0 ?
+          egressConfig.getTimeout() :
+          CircuitBreakerOptions.DEFAULT_TIMEOUT
+      );
+
+      // Retry options
+      if (egressConfig.getRetry() > 0) {
+        options
+          // TODO reset timeout should be configurable or, at least, set by the control plane
+          .setResetTimeout(
+            egressConfig.getBackoffDelay() > 0 ?
+              egressConfig.getBackoffDelay() :
+              CircuitBreakerOptions.DEFAULT_RESET_TIMEOUT
+          )
+          // TODO max failures should be configurable or, at least, set by the control plane
+          .setMaxFailures(egressConfig.getRetry() * 2)
+          .setMaxRetries(egressConfig.getRetry());
+      }
     }
-    return new CircuitBreakerOptions()
-      // This disables circuit breaker notifications on the event bus
-      .setNotificationAddress(null);
+    return options;
   }
 
   /* package visibility for test */
diff --git a/go.mod b/go.mod
index 933a53520..aec4d028d 100644
--- a/go.mod
+++ b/go.mod
@@ -21,7 +21,7 @@ require (
 	k8s.io/apiserver v0.20.7
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20210111153108-fddb29f9d009
-	knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2
+	knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5
 	knative.dev/hack v0.0.0-20210614141220-66ab1a098940
 	knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
 	knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
diff --git a/go.sum b/go.sum
index b3a3c7293..66a57441f 100644
--- a/go.sum
+++ b/go.sum
@@ -1312,14 +1312,13 @@ k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009 h1:0T5IaWHO3sJTEmCP6mUlBvMukxPKUQWqiI/YuiBNMiQ=
 k8s.io/utils v0.0.0-20210111153108-fddb29f9d009/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2 h1:Vz1hXjF7CQ6YMMQXi1XfkModc34WQQTc/hmkKk9+p9M=
-knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2/go.mod h1:8fnNPdQgBT7NT+yCFgE1kuvN0mMsfnCfPIwnlyPrxy0=
+knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5 h1:zNhrpEfWKq5SMyBetHQbQXw2rc8CAbvS9vFZpErrFJo=
+knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5/go.mod h1:oy/wTY1ulXVuZD1ROldNYq+xd29Pe2IYmD84JiCzy5o=
 knative.dev/hack v0.0.0-20210601210329-de04b70e00d0/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940 h1:Aao7SsI087Pa8x+LDLaBdj/ylduPh2bJ8HbB7UjXElE=
 knative.dev/hack v0.0.0-20210614141220-66ab1a098940/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
 knative.dev/hack/schema v0.0.0-20210614141220-66ab1a098940/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
 knative.dev/pkg v0.0.0-20210602095030-0e61d6763dd6/go.mod h1:/hT1IXHPs8ZwQ8yTJtfaYWpLmsLhC2s6/v/ydF7PoiE=
-knative.dev/pkg v0.0.0-20210615143321-77ff8d962c73/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
 knative.dev/pkg v0.0.0-20210618060751-f454995ff92b h1:oPoZUyOvI3fNrxDlpRdB3eW4lPiLIwbNBm+25nEnVco=
 knative.dev/pkg v0.0.0-20210618060751-f454995ff92b/go.mod h1:QVfbwA5ovshGnXDcx54l2QPYhGKppw/JgLYJWe7sA6A=
 knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7 h1:BNP1vs/FNDfpqFb+FaDzwZ9qnr3X+ms5u3lZDZk2ivw=
diff --git a/proto/contract.proto b/proto/contract.proto
index 2941b59e9..411944fc5 100644
--- a/proto/contract.proto
+++ b/proto/contract.proto
@@ -46,6 +46,9 @@ message EgressConfig {
 
   // backoffDelay is the delay before retrying in milliseconds.
   uint64 backoffDelay = 4;
+
+  // timeout is the single request timeout (not the overall retry timeout)
+  uint64 timeout = 5;
 }
 
 // Check dev.knative.eventing.kafka.broker.dispatcher.consumer.DeliveryOrder for more details
diff --git a/vendor/modules.txt b/vendor/modules.txt
index ae9ff4d4c..5ea65da69 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1026,7 +1026,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/eventing v0.23.1-0.20210618081650-b164d3c042d2
+# knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5
 ## explicit
 knative.dev/eventing/pkg/apis/config
 knative.dev/eventing/pkg/apis/duck


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/broker_test.go b/control-plane/pkg/reconciler/broker/broker_test.go
index 461fbf2ba..4526a8e22 100644
--- a/control-plane/pkg/reconciler/broker/broker_test.go
+++ b/control-plane/pkg/reconciler/broker/broker_test.go
@@ -107,11 +107,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -169,8 +169,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "3"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -817,11 +817,11 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "3",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -889,12 +889,10 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				NewConfigMap(&configs, nil),
 				NewService(),
 				BrokerReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
@@ -1201,8 +1199,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1265,8 +1263,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1329,8 +1327,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1390,8 +1388,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1462,8 +1460,8 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 					Generation: 1,
 				}, &configs),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
 			},
 			Key: testKey,
 			WantEvents: []string{
@@ -1490,6 +1488,56 @@ func brokerReconciliation(t *testing.T, format string, configs Configs) {
 				BootstrapServersConfigMapKey: bootstrapServers,
 			},
 		},
+		{
+			Name: "Reconciled normal - unchanged contract - changed data plane pods annotation",
+			Objects: []runtime.Object{
+				NewBroker(
+					WithDelivery(),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:              BrokerUUID,
+							Topics:           []string{BrokerTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_BINARY, IngressType: &contract.Ingress_Path{Path: receiver.Path(BrokerNamespace, BrokerName)}},
+							BootstrapServers: bootstrapServers,
+							EgressConfig:     &contract.EgressConfig{DeadLetter: "http://test-service.test-service-namespace.svc.cluster.local/"},
+						},
+					},
+					Generation: 1,
+				}, &configs),
+				NewService(),
+				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "0"}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				BrokerReceiverPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+				BrokerDispatcherPodUpdate(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "1"}),
+			},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewBroker(
+						WithDelivery(),
+						reconcilertesting.WithInitBrokerConditions,
+						BrokerConfigMapUpdatedReady(&configs),
+						BrokerDataPlaneAvailable,
+						BrokerTopicReady,
+						BrokerConfigParsed,
+						BrokerAddressable(&configs),
+					),
+				},
+			},
+			OtherTestData: map[string]interface{}{
+				BootstrapServersConfigMapKey: bootstrapServers,
+			},
+		},
 	}
 
 	for i := range table {
@@ -1610,8 +1658,6 @@ func brokerFinalization(t *testing.T, format string, configs Configs) {
 					WithDelivery(),
 				),
 				NewService(),
-				BrokerReceiverPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
-				BrokerDispatcherPod(configs.SystemNamespace, map[string]string{base.VolumeGenerationAnnotationKey: "2"}),
 			},
 			Key: testKey,
 			WantCreates: []runtime.Object{
diff --git a/control-plane/pkg/reconciler/sink/kafka_sink_test.go b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
index b39980929..a89531352 100644
--- a/control-plane/pkg/reconciler/sink/kafka_sink_test.go
+++ b/control-plane/pkg/reconciler/sink/kafka_sink_test.go
@@ -102,8 +102,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
-					"annotation_to_preserve":           "value_to_preserve",
+					"annotation_to_preserve": "value_to_preserve",
 				}),
 			},
 			Key: testKey,
@@ -156,7 +155,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				NewSSLSecret(SinkNamespace, "secret-1"),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -221,7 +220,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -334,7 +333,7 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 				),
 				NewConfigMap(&configs, nil),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "1",
+					base.VolumeGenerationAnnotationKey: "0",
 					"annotation_to_preserve":           "value_to_preserve",
 				}),
 			},
@@ -723,16 +722,75 @@ func sinkReconciliation(t *testing.T, format string, configs broker.Configs) {
 							BootstrapServers: bootstrapServers,
 						},
 					},
+					Generation: 1,
 				}, &configs),
 				SinkReceiverPod(configs.SystemNamespace, map[string]string{
-					base.VolumeGenerationAnnotationKey: "2",
+					base.VolumeGenerationAnnotationKey: "1",
 				}),
 			},
 			Key: testKey,
 			WantEvents: []string{
 				finalizerUpdatedEvent,
 			},
-			WantUpdates: []clientgotesting.UpdateActionImpl{},
+			WantPatches: []clientgotesting.PatchActionImpl{
+				patchFinalizers(),
+			},
+			WantStatusUpdates: []clientgotesting.UpdateActionImpl{
+				{
+					Object: NewSink(
+						SinkControllerOwnsTopic,
+						InitSinkConditions,
+						SinkDataPlaneAvailable,
+						SinkConfigParsed,
+						BootstrapServers(bootstrapServersArr),
+						SinkConfigMapUpdatedReady(&configs.Env),
+						SinkTopicReady,
+						SinkTopicReadyWithOwner(SinkTopic(), sink.ControllerTopicOwner),
+						SinkAddressable(&configs.Env),
+					),
+				},
+			},
+		},
+		{
+			Name: "Reconciled normal - unchanged contract - changed receiver pod annotation",
+			Objects: []runtime.Object{
+				NewSink(
+					SinkControllerOwnsTopic,
+					BootstrapServers(bootstrapServersArr),
+				),
+				NewConfigMapFromContract(&contract.Contract{
+					Resources: []*contract.Resource{
+						{
+							Uid:     "5384faa4-6bdf-428d-b6c2-d6f89ce1d44b",
+							Topics:  []string{"my-existing-topic-a"},
+							Ingress: &contract.Ingress{IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+						},
+						{
+							Uid:    "5384faa4-6bdf-428d-b6c2-d6f89ce1d44a",
+							Topics: []string{"my-existing-topic-b"},
+						},
+						{
+							Uid:              SinkUUID,
+							Topics:           []string{SinkTopic()},
+							Ingress:          &contract.Ingress{ContentMode: contract.ContentMode_STRUCTURED, IngressType: &contract.Ingress_Path{Path: receiver.Path(SinkNamespace, SinkName)}},
+							BootstrapServers: bootstrapServers,
+						},
+					},
+					Generation: 1,
+				}, &configs),
+				SinkReceiverPod(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "0",
+				}),
+			},
+			Key: testKey,
+			WantEvents: []string{
+				finalizerUpdatedEvent,
+			},
+			WantUpdates: []clientgotesting.UpdateActionImpl{
+				SinkReceiverPodUpdate(configs.SystemNamespace, map[string]string{
+					base.VolumeGenerationAnnotationKey: "1",
+				}),
+			},
 			WantPatches: []clientgotesting.PatchActionImpl{
 				patchFinalizers(),
 			},
diff --git a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
index 2d665b254..b42e75967 100644
--- a/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
+++ b/vendor/knative.dev/eventing/pkg/reconciler/testing/v1/subscription.go
@@ -125,6 +125,27 @@ func WithSubscriptionChannel(gvk metav1.GroupVersionKind, name string) Subscript
 	}
 }
 
+func WithSubscriptionChannelUsingGroup(gvk metav1.GroupVersionKind, name string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Spec.Channel = duckv1.KReference{
+			Group: gvk.Group,
+			Kind:  gvk.Kind,
+			Name:  name,
+		}
+	}
+}
+
+func WithSubscriptionChannelUsingApiVersionAndGroup(gvk metav1.GroupVersionKind, name string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Spec.Channel = duckv1.KReference{
+			APIVersion: apiVersion(gvk),
+			Group:      gvk.Group,
+			Kind:       gvk.Kind,
+			Name:       name,
+		}
+	}
+}
+
 func WithSubscriptionSubscriberRef(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
 	return func(s *v1.Subscription) {
 		s.Spec.Subscriber = &duckv1.Destination{
@@ -151,6 +172,20 @@ func WithSubscriptionSubscriberRefUsingGroup(gvk metav1.GroupVersionKind, name,
 	}
 }
 
+func WithSubscriptionSubscriberRefUsingApiVersionAndGroup(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
+	return func(s *v1.Subscription) {
+		s.Spec.Subscriber = &duckv1.Destination{
+			Ref: &duckv1.KReference{
+				APIVersion: apiVersion(gvk),
+				Group:      gvk.Group,
+				Kind:       gvk.Kind,
+				Name:       name,
+				Namespace:  namespace,
+			},
+		}
+	}
+}
+
 func WithSubscriptionDeliveryRef(gvk metav1.GroupVersionKind, name, namespace string) SubscriptionOption {
 	return func(s *v1.Subscription) {
 		s.Spec.Delivery = &eventingduckv1.DeliverySpec{