Repository: knative-sandbox/eventing-kafka
Pull request #729 (2021-06-22 13:24:18)

Fix no kafka channel address issue
Fixes #

```
=== CONT  TestSingleStructuredEventForChannel/KafkaChannel-messaging.knative.dev/v1beta1
    operation.go:95: Failed to get all KResources ready failed waiting for {TypeMeta:{Kind:Subscription APIVersion:messaging.knative.dev/v1} ObjectMeta:{Name:e2e-singleevent-subscription-structured GenerateName: Namespace:eventing-e2e24 SelfLink: UID: ResourceVersion: Generation:0 CreationTimestamp:0001-01-01 00:00:00 +0000 UTC DeletionTimestamp:<nil> DeletionGracePeriodSeconds:<nil> Labels:map[] Annotations:map[] OwnerReferences:[] Finalizers:[] ClusterName: ManagedFields:[]}} to become ready: timed out waiting for the condition
    operation.go:116: Failed to get all test resources ready: failed waiting for {TypeMeta:{Kind:Subscription APIVersion:messaging.knative.dev/v1} ObjectMeta:{Name:e2e-singleevent-subscription-structured GenerateName: Namespace:eventing-e2e24 SelfLink: UID: ResourceVersion: Generation:0 CreationTimestamp:0001-01-01 00:00:00 +0000 UTC DeletionTimestamp:<nil> DeletionGracePeriodSeconds:<nil> Labels:map[] Annotations:map[] OwnerReferences:[] Finalizers:[] ClusterName: ManagedFields:[]}} to become ready: timed out waiting for the condition
```

Consolidated KafkaChannel was not getting its address in status.
The root cause is that Sarama config is not read and because of that the controller can't even come to the stage that it initializes the status nor scales the dispatcher.

## Proposed Changes

-
-
-

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- 🎁 Add new feature
- 🐛 Fix bug
- 🧽 Update or clean up current behavior
- 🗑️ Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
🗒️ 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**

<!--
📖 If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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/blob/main/OWNERS)~~ [aliok]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow-robot: @aliok: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-knative-sandbox-eventing-kafka-unit-tests | 35cf0b50524423efe04ae4d50ad2fd70a29fb8fd | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/729/pull-knative-sandbox-eventing-kafka-unit-tests/1407328602223022080) | `/test pull-knative-sandbox-eventing-kafka-unit-tests`

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

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
knative-metrics-robot: The following is the coverage report on the affected files.
Say `/test pull-knative-sandbox-eventing-kafka-go-coverage` to re-run this coverage report

File | Old Coverage | New Coverage | Delta
---- |:------------:|:------------:|:-----:
[pkg/channel/consolidated/utils/util.go](https://storage.cloud.google.com/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka/729/pull-knative-sandbox-eventing-kafka-go-coverage/1407328602252382208/artifacts/line-cov.html#file0) | 100.0% | 96.4% | -3.6


Patch:
diff --git a/go.mod b/go.mod
index 428b56cb3..e778c50ba 100644
--- a/go.mod
+++ b/go.mod
@@ -34,10 +34,10 @@ require (
 	k8s.io/apimachinery v0.20.7
 	k8s.io/client-go v0.20.7
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
-	knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79
-	knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5
-	knative.dev/hack v0.0.0-20210614141220-66ab1a098940
-	knative.dev/networking v0.0.0-20210618141751-9b6da1991d38
-	knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
-	knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
+	knative.dev/control-protocol v0.0.0-20210622195244-f6f46782b93d
+	knative.dev/eventing v0.23.1-0.20210622190628-74d711b6a367
+	knative.dev/hack v0.0.0-20210622141627-e28525d8d260
+	knative.dev/networking v0.0.0-20210622182128-53f45d6d2cfa
+	knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80
+	knative.dev/reconciler-test v0.0.0-20210622190528-b155197924aa
 )
diff --git a/go.sum b/go.sum
index 803ddbfd3..debe5c9f7 100644
--- a/go.sum
+++ b/go.sum
@@ -1346,23 +1346,22 @@ k8s.io/kube-openapi v0.0.0-20201113171705-d219536bb9fd/go.mod h1:WOJ3KddDSol4tAG
 k8s.io/utils v0.0.0-20200729134348-d5654de09c73/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920 h1:CbnUZsM497iRC5QMVkHwyl8s2tB3g7yaSHkYPkpgelw=
 k8s.io/utils v0.0.0-20201110183641-67b214c5f920/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
-knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79 h1:CepZkHLRpqnQtdbYtXa7R/V5LbjxzkKIQdQxNOFyrhY=
-knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79/go.mod h1:iRE6O4mY88gH6xWl6ZFhG6sJobzrZjFrz5xhX4h93Ns=
-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/control-protocol v0.0.0-20210622195244-f6f46782b93d h1:vRyaa/YIKhpbgYmNArTkyDp4OkJqmQP0EQCZZVE1pbo=
+knative.dev/control-protocol v0.0.0-20210622195244-f6f46782b93d/go.mod h1:7KaKumGBRwejhysypmTvkIdFbs0RQ5ZgiziTGehvZTg=
+knative.dev/eventing v0.23.1-0.20210622190628-74d711b6a367 h1:bvTh6kPpmAnCDeAjKLPNCyS8ThV4d/M7Y4ov44s6MgM=
+knative.dev/eventing v0.23.1-0.20210622190628-74d711b6a367/go.mod h1:btTVjBp88nAKBMtAQaLvo4A75MF+h2TYY8hdnCrHKz4=
 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/networking v0.0.0-20210618141751-9b6da1991d38 h1:LDo9zIQL3EQQ2kEhqkVZR1dvAbofEc0sdmD7dLPSZ9M=
-knative.dev/networking v0.0.0-20210618141751-9b6da1991d38/go.mod h1:A6gUyRVaXmDebDRVsVdcOz65S8JCnEHKprNvAeaj2jE=
+knative.dev/hack v0.0.0-20210622141627-e28525d8d260 h1:f2eMtOubAOc/Q7JlvFPDKXiPlJVK+VpX2Cot8hRzCgQ=
+knative.dev/hack v0.0.0-20210622141627-e28525d8d260/go.mod h1:PHt8x8yX5Z9pPquBEfIj0X66f8iWkWfR0S/sarACJrI=
+knative.dev/hack/schema v0.0.0-20210622141627-e28525d8d260/go.mod h1:ffjwmdcrH5vN3mPhO8RrF2KfNnbHeCE2C60A+2cv3U0=
+knative.dev/networking v0.0.0-20210622182128-53f45d6d2cfa h1:MprAGBX3eRaBZFRXC3ZjsnhnjttfprRVXdxmTeEzC2o=
+knative.dev/networking v0.0.0-20210622182128-53f45d6d2cfa/go.mod h1:vwPACNE712tyoEG4fjyUIgfL4xkbXFugx8bxW+QrKn4=
 knative.dev/pkg v0.0.0-20210602095030-0e61d6763dd6/go.mod h1:/hT1IXHPs8ZwQ8yTJtfaYWpLmsLhC2s6/v/ydF7PoiE=
-knative.dev/pkg v0.0.0-20210616195222-841aa7369ca1/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-20210602211045-db062eb7ce42/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
-knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7 h1:BNP1vs/FNDfpqFb+FaDzwZ9qnr3X+ms5u3lZDZk2ivw=
-knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
+knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80 h1:GHJ3lglE0/YHfBMMJqluqUNLOmsNXh7s7DBnfrkpRMM=
+knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80/go.mod h1:kGegTnbZ+ljFjAE3E1+8wgaH2LMv8qYi+72o3F3cbdc=
+knative.dev/reconciler-test v0.0.0-20210622182628-46348d29ae6b/go.mod h1:Ggh7/FyaVr9R1kQiuB9KpNsvaKVoPlZ0w8OoxUGi9Ys=
+knative.dev/reconciler-test v0.0.0-20210622190528-b155197924aa h1:l1+le4HOK8+CJiy3ZJekgJfSVLw2VU9d5NvIyEH9dI0=
+knative.dev/reconciler-test v0.0.0-20210622190528-b155197924aa/go.mod h1:4wqv2WyWUC5yhTesRUVwgjv/fHTHny1RYBfdB6tVDok=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=
diff --git a/pkg/channel/consolidated/utils/util.go b/pkg/channel/consolidated/utils/util.go
index f8bc3024b..1d602485d 100644
--- a/pkg/channel/consolidated/utils/util.go
+++ b/pkg/channel/consolidated/utils/util.go
@@ -84,6 +84,11 @@ func GetKafkaConfig(ctx context.Context, clientId string, configMap map[string]s
 		)
 		// Since LoadSettings isn't going to be called in this situation, we need to call getAuth explicitly
 		eventingKafkaConfig.Auth = getAuth(ctx, eventingKafkaConfig.Kafka.AuthSecretName, eventingKafkaConfig.Kafka.AuthSecretNamespace)
+		ekConfigFromOldFormat, err := sarama.LoadSettings(ctx, clientId, configMap, getAuth)
+		if err != nil {
+			return nil, err
+		}
+		eventingKafkaConfig.Sarama = ekConfigFromOldFormat.Sarama
 	} else {
 		eventingKafkaConfig, err = sarama.LoadSettings(ctx, clientId, configMap, getAuth)
 	}
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
index 0577bd584..b82bbc4f9 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1/delivery_types.go
@@ -20,6 +20,7 @@ import (
 	"context"
 
 	"github.com/rickb777/date/period"
+	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
@@ -37,6 +38,15 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request. The value must be greater than 0.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	// Note: This API is EXPERIMENTAL and might break anytime. For more details: https://github.com/knative/eventing/issues/5148
+	// +optional
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -65,6 +75,17 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		if feature.FromContext(ctx).IsEnabled(feature.DeliveryTimeout) {
+			t, te := period.Parse(*ds.Timeout)
+			if te != nil || t.IsZero() {
+				errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+			}
+		} else {
+			errs = errs.Also(apis.ErrDisallowedFields("timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
index 902fb4b1f..0501d0e60 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1/zz_generated.deepcopy.go
@@ -146,6 +146,11 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(string)
+		**out = **in
+	}
 	if in.BackoffPolicy != nil {
 		in, out := &in.BackoffPolicy, &out.BackoffPolicy
 		*out = new(BackoffPolicyType)
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
index d8a3bc880..d48ccde0c 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_conversion.go
@@ -31,6 +31,7 @@ func (source *DeliverySpec) ConvertTo(ctx context.Context, to apis.Convertible)
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == BackoffPolicyLinear {
 				linear := eventingduckv1.BackoffPolicyLinear
@@ -55,6 +56,7 @@ func (sink *DeliverySpec) ConvertFrom(ctx context.Context, from apis.Convertible
 	case *eventingduckv1.DeliverySpec:
 		sink.Retry = source.Retry
 		sink.BackoffDelay = source.BackoffDelay
+		sink.Timeout = source.Timeout
 		if source.BackoffPolicy != nil {
 			if *source.BackoffPolicy == eventingduckv1.BackoffPolicyLinear {
 				linear := BackoffPolicyLinear
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
index ac2001704..33ebd3ae2 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/delivery_types.go
@@ -38,6 +38,13 @@ type DeliverySpec struct {
 	// +optional
 	Retry *int32 `json:"retry,omitempty"`
 
+	// Timeout is the timeout of each single request.
+	// More information on Duration format:
+	//  - https://www.iso.org/iso-8601-date-and-time-format.html
+	//  - https://en.wikipedia.org/wiki/ISO_8601
+	//
+	Timeout *string `json:"timeout,omitempty"`
+
 	// BackoffPolicy is the retry backoff policy (linear, exponential).
 	// +optional
 	BackoffPolicy *BackoffPolicyType `json:"backoffPolicy,omitempty"`
@@ -66,6 +73,13 @@ func (ds *DeliverySpec) Validate(ctx context.Context) *apis.FieldError {
 		errs = errs.Also(apis.ErrInvalidValue(*ds.Retry, "retry"))
 	}
 
+	if ds.Timeout != nil {
+		_, te := period.Parse(*ds.Timeout)
+		if te != nil {
+			errs = errs.Also(apis.ErrInvalidValue(*ds.Timeout, "timeout"))
+		}
+	}
+
 	if ds.BackoffPolicy != nil {
 		switch *ds.BackoffPolicy {
 		case BackoffPolicyExponential, BackoffPolicyLinear:
diff --git a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
index c68ae7471..175fbad98 100644
--- a/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
+++ b/vendor/knative.dev/eventing/pkg/apis/duck/v1beta1/zz_generated.deepcopy.go
@@ -146,6 +146,11 @@ func (in *DeliverySpec) DeepCopyInto(out *DeliverySpec) {
 		*out = new(int32)
 		**out = **in
 	}
+	if in.Timeout != nil {
+		in, out := &in.Timeout, &out.Timeout
+		*out = new(string)
+		**out = **in
+	}
 	if in.BackoffPolicy != nil {
 		in, out := &in.BackoffPolicy, &out.BackoffPolicy
 		*out = new(BackoffPolicyType)
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go b/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go
new file mode 100644
index 000000000..2a5f9226a
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/api_validation.go
@@ -0,0 +1,85 @@
+/*
+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 feature
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"knative.dev/pkg/apis"
+)
+
+// ValidateAPIFields checks that the experimental features fields are disabled if the experimental flag is disabled.
+// experimentalFields can contain a string with dots, to identify sub-structs, like "Destination.Ref.APIVersion"
+func ValidateAPIFields(ctx context.Context, featureName string, object interface{}, experimentalFields ...string) (errs *apis.FieldError) {
+	obj := reflect.ValueOf(object)
+	obj = reflect.Indirect(obj)
+	if obj.Kind() != reflect.Struct {
+		return nil
+	}
+
+	// If feature not enabled, let's check the field is not used
+	if !FromContext(ctx).IsEnabled(featureName) {
+		for _, fieldName := range experimentalFields {
+			fieldVal := walk(obj, strings.Split(fieldName, ".")...)
+
+			if !fieldVal.IsZero() {
+				errs = errs.Also(&apis.FieldError{
+					Message: fmt.Sprintf("Disallowed field because the experimental feature '%s' is disabled", featureName),
+					Paths:   []string{fmt.Sprintf("%s.%s", obj.Type().Name(), fieldName)},
+				})
+			}
+		}
+	}
+
+	return errs
+}
+
+// ValidateAnnotations checks that the experimental features annotations are disabled if the experimental flag is disabled
+func ValidateAnnotations(ctx context.Context, featureName string, object metav1.Object, experimentalAnnotations ...string) (errs *apis.FieldError) {
+	// If feature not enabled, let's check the annotation is not used
+	if !FromContext(ctx).IsEnabled(featureName) {
+		for _, annotation := range experimentalAnnotations {
+			if _, ok := object.GetAnnotations()[annotation]; ok {
+				errs = errs.Also(&apis.FieldError{
+					Message: fmt.Sprintf("Disallowed annotation because the experimental feature '%s' is disabled", featureName),
+					Paths:   []string{annotation},
+				})
+			}
+		}
+	}
+
+	return errs
+}
+
+func walk(value reflect.Value, paths ...string) reflect.Value {
+	switch value.Kind() {
+	case reflect.Struct:
+		newVal := value.FieldByName(paths[0])
+		if len(paths) == 1 {
+			return newVal
+		}
+		return walk(value.FieldByName(paths[0]), paths[1:]...)
+	case reflect.Ptr:
+		return walk(reflect.Indirect(value), paths...)
+	default:
+		return reflect.Zero(value.Type())
+	}
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/features.go b/vendor/knative.dev/eventing/pkg/apis/feature/features.go
new file mode 100644
index 000000000..b12355886
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/features.go
@@ -0,0 +1,80 @@
+/*
+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 feature
+
+import (
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+// Flag is a string value which can be either Enabled, Disabled, or Allowed.
+type Flag string
+
+const (
+	// Enabled turns on an optional behavior.
+	Enabled Flag = "Enabled"
+	// Disabled turns off an optional behavior.
+	Disabled Flag = "Disabled"
+	// Allowed neither explicitly disables or enables a behavior.
+	// eg. allow a client to control behavior with an annotation or allow a new value through validation.
+	Allowed Flag = "Allowed"
+)
+
+// Flags is a map containing all the enabled/disabled flags for the experimental features.
+// Missing entry in the map means feature is equal to feature not enabled.
+type Flags map[string]Flag
+
+// IsEnabled returns true if the feature is enabled
+func (e Flags) IsEnabled(featureName string) bool {
+	return e != nil && e[featureName] == Enabled
+}
+
+// IsAllowed returns true if the feature is enabled or allowed
+func (e Flags) IsAllowed(featureName string) bool {
+	return e.IsEnabled(featureName) || (e != nil && e[featureName] == Allowed)
+}
+
+// NewFlagsConfigFromMap creates a Flags from the supplied Map
+func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
+	flags := Flags{}
+
+	for k, v := range data {
+		if strings.HasPrefix(k, "_") {
+			// Ignore all the keys starting with _
+			continue
+		}
+		sanitizedKey := strings.TrimSpace(k)
+		if strings.EqualFold(v, string(Allowed)) {
+			flags[sanitizedKey] = Allowed
+		} else if strings.EqualFold(v, string(Disabled)) {
+			flags[sanitizedKey] = Disabled
+		} else if strings.EqualFold(v, string(Enabled)) {
+			flags[sanitizedKey] = Enabled
+		} else {
+			return Flags{}, fmt.Errorf("cannot parse the boolean flag '%s' = '%s'. Allowed values: [true, false]", k, v)
+		}
+	}
+
+	return flags, nil
+}
+
+// NewFlagsConfigFromConfigMap creates a Flags from the supplied configMap
+func NewFlagsConfigFromConfigMap(config *corev1.ConfigMap) (Flags, error) {
+	return NewFlagsConfigFromMap(config.Data)
+}
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
new file mode 100644
index 000000000..050316858
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go
@@ -0,0 +1,22 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package feature
+
+const (
+	KReferenceGroup = "kreference-group"
+	DeliveryTimeout = "delivery-timeout"
+)
diff --git a/vendor/knative.dev/eventing/pkg/apis/feature/store.go b/vendor/knative.dev/eventing/pkg/apis/feature/store.go
new file mode 100644
index 000000000..a22f313c2
--- /dev/null
+++ b/vendor/knative.dev/eventing/pkg/apis/feature/store.go
@@ -0,0 +1,108 @@
+/*
+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 feature
+
+import (
+	"context"
+
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	"knative.dev/pkg/configmap"
+)
+
+const (
+	// FlagsConfigName is the name of config map containing the experimental features flags
+	FlagsConfigName = "config-features"
+)
+
+type cfgKey struct{}
+
+// FromContext extracts a Config from the provided context.
+func FromContext(ctx context.Context) Flags {
+	x, ok := ctx.Value(cfgKey{}).(Flags)
+	if ok {
+		return x
+	}
+	return nil
+}
+
+// FromContextOrDefaults is like FromContext, but when no Flags is attached it
+// returns an empty Flags.
+func FromContextOrDefaults(ctx context.Context) Flags {
+	if cfg := FromContext(ctx); cfg != nil {
+		return cfg
+	}
+	return Flags{}
+}
+
+// ToContext attaches the provided Flags to the provided context, returning the
+// new context with the Flags attached.
+func ToContext(ctx context.Context, c Flags) context.Context {
+	return fillContextWithFeatureSpecificFlags(context.WithValue(ctx, cfgKey{}, c), c)
+}
+
+// Store is a typed wrapper around configmap.Untyped store to handle our configmaps.
+// +k8s:deepcopy-gen=false
+type Store struct {
+	*configmap.UntypedStore
+}
+
+// NewStore creates a new store of Configs and optionally calls functions when ConfigMaps are updated.
+func NewStore(logger configmap.Logger, onAfterStore ...func(name string, value interface{})) *Store {
+	store := &Store{
+		UntypedStore: configmap.NewUntypedStore(
+			"feature-flags",
+			logger,
+			configmap.Constructors{
+				FlagsConfigName: NewFlagsConfigFromConfigMap,
+			},
+			onAfterStore...,
+		),
+	}
+
+	return store
+}
+
+// ToContext attaches the current Config state to the provided context.
+func (s *Store) ToContext(ctx context.Context) context.Context {
+	return ToContext(ctx, s.Load())
+}
+
+// IsEnabled is a shortcut for Load().IsEnabled(featureName)
+func (s *Store) IsEnabled(featureName string) bool {
+	return s.Load().IsEnabled(featureName)
+}
+
+// IsAllowed is a shortcut for Load().IsAllowed(featureName)
+func (s *Store) IsAllowed(featureName string) bool {
+	return s.Load().IsAllowed(featureName)
+}
+
+// Load creates a Config from the current config state of the Store.
+func (s *Store) Load() Flags {
+	loaded := s.UntypedLoad(FlagsConfigName)
+	if loaded == nil {
+		return Flags(nil)
+	}
+	return loaded.(Flags)
+}
+
+func fillContextWithFeatureSpecificFlags(ctx context.Context, flags Flags) context.Context {
+	if flags.IsEnabled(KReferenceGroup) {
+		ctx = duckv1.KReferenceGroupAllowed(ctx)
+	}
+	return ctx
+}
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
index 8ff4c6c70..6d223f7da 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/message_sender.go
@@ -57,8 +57,18 @@ func (s *HTTPMessageSender) SendWithRetries(req *nethttp.Request, config *RetryC
 		return s.Send(req)
 	}
 
+	client := s.Client
+	if config.RequestTimeout != 0 {
+		client = &nethttp.Client{
+			Transport:     client.Transport,
+			CheckRedirect: client.CheckRedirect,
+			Jar:           client.Jar,
+			Timeout:       config.RequestTimeout,
+		}
+	}
+
 	retryableClient := retryablehttp.Client{
-		HTTPClient:   s.Client,
+		HTTPClient:   client,
 		RetryWaitMin: defaultRetryWaitMin,
 		RetryWaitMax: defaultRetryWaitMax,
 		RetryMax:     config.RetryMax,
diff --git a/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go b/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
index b43e93e08..416049b46 100644
--- a/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
+++ b/vendor/knative.dev/eventing/pkg/kncloudevents/retries.go
@@ -63,6 +63,9 @@ type RetryConfig struct {
 
 	CheckRetry CheckRetry
 	Backoff    Backoff
+
+	// RequestTimeout represents the timeout of the single request
+	RequestTimeout time.Duration
 }
 
 func NoRetries() RetryConfig {
diff --git a/vendor/knative.dev/pkg/leaderelection/config.go b/vendor/knative.dev/pkg/leaderelection/config.go
index 3d9df1dc7..8d785c522 100644
--- a/vendor/knative.dev/pkg/leaderelection/config.go
+++ b/vendor/knative.dev/pkg/leaderelection/config.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/kelseyhightower/envconfig"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 
@@ -79,11 +78,6 @@ type Config struct {
 	LeaseDuration time.Duration
 	RenewDeadline time.Duration
 	RetryPeriod   time.Duration
-
-	// This field is deprecated and will be removed once downstream
-	// repositories have removed their validation of it.
-	// TODO(https://github.com/knative/pkg/issues/1478): Remove this field.
-	EnabledComponents sets.String
 }
 
 func (c *Config) GetComponentConfig(name string) ComponentConfig {
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 94e78001f..c13c0a458 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -1110,7 +1110,7 @@ k8s.io/utils/buffer
 k8s.io/utils/integer
 k8s.io/utils/pointer
 k8s.io/utils/trace
-# knative.dev/control-protocol v0.0.0-20210616092621-1469a722be79
+# knative.dev/control-protocol v0.0.0-20210622195244-f6f46782b93d
 ## explicit
 knative.dev/control-protocol/pkg
 knative.dev/control-protocol/pkg/certificates
@@ -1118,7 +1118,7 @@ knative.dev/control-protocol/pkg/message
 knative.dev/control-protocol/pkg/network
 knative.dev/control-protocol/pkg/reconciler
 knative.dev/control-protocol/pkg/service
-# knative.dev/eventing v0.23.1-0.20210621164716-b312f0e1c8e5
+# knative.dev/eventing v0.23.1-0.20210622190628-74d711b6a367
 ## explicit
 knative.dev/eventing/pkg/adapter/v2
 knative.dev/eventing/pkg/adapter/v2/test
@@ -1130,6 +1130,7 @@ knative.dev/eventing/pkg/apis/duck/v1beta1
 knative.dev/eventing/pkg/apis/eventing
 knative.dev/eventing/pkg/apis/eventing/v1
 knative.dev/eventing/pkg/apis/eventing/v1beta1
+knative.dev/eventing/pkg/apis/feature
 knative.dev/eventing/pkg/apis/flows
 knative.dev/eventing/pkg/apis/flows/v1
 knative.dev/eventing/pkg/apis/messaging
@@ -1236,14 +1237,14 @@ knative.dev/eventing/test/upgrade/prober/wathola/fetcher
 knative.dev/eventing/test/upgrade/prober/wathola/forwarder
 knative.dev/eventing/test/upgrade/prober/wathola/receiver
 knative.dev/eventing/test/upgrade/prober/wathola/sender
-# knative.dev/hack v0.0.0-20210614141220-66ab1a098940
+# knative.dev/hack v0.0.0-20210622141627-e28525d8d260
 ## explicit
 knative.dev/hack
 knative.dev/hack/shell
-# knative.dev/networking v0.0.0-20210618141751-9b6da1991d38
+# knative.dev/networking v0.0.0-20210622182128-53f45d6d2cfa
 ## explicit
 knative.dev/networking/pkg/prober
-# knative.dev/pkg v0.0.0-20210618060751-f454995ff92b
+# knative.dev/pkg v0.0.0-20210622173328-dd0db4b05c80
 ## explicit
 knative.dev/pkg/apiextensions/storageversion
 knative.dev/pkg/apiextensions/storageversion/cmd/migrate
@@ -1349,7 +1350,7 @@ knative.dev/pkg/webhook/resourcesemantics
 knative.dev/pkg/webhook/resourcesemantics/conversion
 knative.dev/pkg/webhook/resourcesemantics/defaulting
 knative.dev/pkg/webhook/resourcesemantics/validation
-# knative.dev/reconciler-test v0.0.0-20210603210445-0071c48281c7
+# knative.dev/reconciler-test v0.0.0-20210622190528-b155197924aa
 ## explicit
 knative.dev/reconciler-test/cmd/eventshub
 knative.dev/reconciler-test/pkg/environment


Test patch:
diff --git a/pkg/channel/consolidated/utils/util_test.go b/pkg/channel/consolidated/utils/util_test.go
index 765cb54fa..a8965b47d 100644
--- a/pkg/channel/consolidated/utils/util_test.go
+++ b/pkg/channel/consolidated/utils/util_test.go
@@ -282,6 +282,9 @@ func TestGetKafkaConfig(t *testing.T) {
 				EventingKafka: &config.EventingKafkaConfig{
 					CloudEvents: defaultCloudEvents,
 					Kafka:       kafkaConfig("kafkabroker.kafka:9092", kafkasarama.DefaultAuthSecretName, system.Namespace()),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
@@ -293,6 +296,9 @@ func TestGetKafkaConfig(t *testing.T) {
 				EventingKafka: &config.EventingKafkaConfig{
 					CloudEvents: defaultCloudEvents,
 					Kafka:       kafkaConfig("kafkabroker.kafka:9092", "kafka-auth-secret", "default"),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
@@ -304,6 +310,9 @@ func TestGetKafkaConfig(t *testing.T) {
 				EventingKafka: &config.EventingKafkaConfig{
 					CloudEvents: defaultCloudEvents,
 					Kafka:       kafkaConfig("kafkabroker1.kafka:9092,kafkabroker2.kafka:9092", kafkasarama.DefaultAuthSecretName, system.Namespace()),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
@@ -318,6 +327,9 @@ func TestGetKafkaConfig(t *testing.T) {
 						MaxIdleConnsPerHost: 100,
 					},
 					Kafka: kafkaConfig("kafkabroker.kafka:9092", kafkasarama.DefaultAuthSecretName, system.Namespace()),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
@@ -332,6 +344,9 @@ func TestGetKafkaConfig(t *testing.T) {
 						MaxIdleConnsPerHost: 900,
 					},
 					Kafka: kafkaConfig("kafkabroker.kafka:9092", kafkasarama.DefaultAuthSecretName, system.Namespace()),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
@@ -346,6 +361,9 @@ func TestGetKafkaConfig(t *testing.T) {
 						MaxIdleConnsPerHost: 600,
 					},
 					Kafka: kafkaConfig("kafkabroker.kafka:9092", kafkasarama.DefaultAuthSecretName, system.Namespace()),
+					Sarama: config.EKSaramaConfig{
+						Config: saramaEmptyConfig,
+					},
 				},
 			},
 		},
diff --git a/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go b/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
index 64d85c97a..4d516260c 100644
--- a/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
+++ b/vendor/knative.dev/reconciler-test/pkg/eventshub/recorder_vent/recorder.go
@@ -42,6 +42,14 @@ type recorder struct {
 }
 
 func (r *recorder) Vent(observed eventshub.EventInfo) error {
+	if observed.Event != nil {
+		var event corev1.Event
+		err := json.Unmarshal(observed.Event.DataEncoded, &event)
+		if err == nil && event.Reason == eventshub.CloudEventObservedReason {
+			// prevent event loops from happening with APIServerSources
+			return nil
+		}
+	}
 	b, err := json.Marshal(observed)
 	if err != nil {
 		return err