Repository: knative-sandbox/eventing-kafka-broker
Pull request #2626 (2022-09-08 21:10:02)

Manual backport of proper webhook implementation
Fixes #

<!-- Please include the 'why' behind your changes if no issue exists -->

## Proposed Changes

- Manual backport of proper webhook implementation from this PR: #2562
- As #2562 was done when namespaced broker is around, I had to comment out some parts.
- Created https://github.com/knative-sandbox/eventing-kafka-broker/issues/2630 to track that

<!--
If this change has user-visible impact, follow the instructions below.
Examples include:

- :gift: Add new feature
- :bug: Fix bug
- :broom: Update or clean up current behavior
- :wastebasket: Remove feature or internal logic

Otherwise delete the rest of this template.
-->

**Release Note**

<!--
:page_facing_up: If this change has user-visible impact, write a release note in the block
below. Include the string "action required" if additional action is required of
users switching to the new release, for example in case of a breaking change.

Write as if you are speaking to users, not other Knative contributors. If this
change has no user-visible impact, no release-note is needed.
-->

```release-note

```

**Docs**

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->


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

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

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

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

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

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.6/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: once the present PR merges, I will cherry-pick it on top of release-1.5 in a new PR and assign it to you.

<details>

In response to [this](https://github.com/knative-sandbox/eventing-kafka-broker/pull/2626#issuecomment-1241236804):

>/cherry-pick release-1.5


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
aliok: /cherry-pick release-1.5
knative-prow[bot]: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.6 | 0391571587602bec72c1c96f6eeed80dec5b5acf | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2626/unit-tests_eventing-kafka-broker_release-1.6/1567983630079234048) | true | `/test unit-tests_eventing-kafka-broker_release-1.6`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2626](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (0391571) into [release-1.6](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/ba8b04000765dd45bdec7c47f1b83cea4950b926?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (ba8b040) will **increase** coverage by `21.95%`.
> The diff coverage is `n/a`.

```diff
@@                Coverage Diff                 @@
##             release-1.6    #2626       +/-   ##
==================================================
+ Coverage          59.85%   81.80%   +21.95%     
- Complexity             0      723      +723     
==================================================
  Files                 69       72        +3     
  Lines               6997     2501     -4496     
  Branches               0      228      +228     
==================================================
- Hits                4188     2046     -2142     
+ Misses              2512      329     -2183     
+ Partials             297      126      -171     
```

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

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [...ol-plane/pkg/apis/eventing/v1/broker\_validation.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MS9icm9rZXJfdmFsaWRhdGlvbi5nbw==) | | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [control-plane/pkg/contract/serde.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3Qvc2VyZGUuZ28=) | | |
| [...ol-plane/pkg/reconciler/trigger/v2/controllerv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3YyL2NvbnRyb2xsZXJ2Mi5nbw==) | | |
| [...-plane/pkg/reconciler/trigger/trigger\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL3RyaWdnZXJfbGlmZWN5Y2xlLmdv) | | |
| [...l-plane/pkg/reconciler/consumergroup/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2NvbnRyb2xsZXIuZ28=) | | |
| [...ntrol-plane/pkg/apis/eventing/v1alpha1/register.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9yZWdpc3Rlci5nbw==) | | |
| [...ntrol-plane/pkg/reconciler/consumergroup/filter.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jb25zdW1lcmdyb3VwL2ZpbHRlci5nbw==) | | |
| [...ls/kafka/eventing/v1alpha1/consumer\_group\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfZ3JvdXBfdHlwZXMuZ28=) | | |
| [control-plane/pkg/reconciler/sink/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9zaW5rL2NvbnRyb2xsZXIuZ28=) | | |
| ... and [131 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2626/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
unit-tests_eventing-kafka-broker_release-1.6 | 3b08a6aecb124a985061ccc087ffd2572e2803c6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2626/unit-tests_eventing-kafka-broker_release-1.6/1567987349592215552) | true | `/test unit-tests_eventing-kafka-broker_release-1.6`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
aliok: /test unit-tests_eventing-kafka-broker_release-1.6
knative-prow[bot]: @aliok: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
upgrade-tests_eventing-kafka-broker_release-1.6 | 3b08a6aecb124a985061ccc087ffd2572e2803c6 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2626/upgrade-tests_eventing-kafka-broker_release-1.6/1567987350221361152) | true | `/test upgrade-tests_eventing-kafka-broker_release-1.6`

[Your PR dashboard](https://prow.knative.dev/pr).

<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 -->
aliok: /test upgrade-tests_eventing-kafka-broker_release-1.6
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

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

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

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

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

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

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

Patch:
diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go
index 817f85797..fe73316c1 100644
--- a/control-plane/cmd/webhook-kafka/main.go
+++ b/control-plane/cmd/webhook-kafka/main.go
@@ -32,7 +32,7 @@ import (
 	"knative.dev/pkg/webhook/resourcesemantics/validation"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
-	eventingcorev1beta1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1"
 
 	messagingv1beta1 "knative.dev/eventing-kafka/pkg/apis/messaging/v1beta1"
 
@@ -49,10 +49,7 @@ var types = map[schema.GroupVersionKind]resourcesemantics.GenericCRD{
 	eventingv1alpha1.SchemeGroupVersion.WithKind("KafkaSink"):    &eventingv1alpha1.KafkaSink{},
 	sourcesv1beta1.SchemeGroupVersion.WithKind("KafkaSource"):    &sourcesv1beta1.KafkaSource{},
 	messagingv1beta1.SchemeGroupVersion.WithKind("KafkaChannel"): &messagingv1beta1.KafkaChannel{},
-}
-
-var validationCallbacks = map[schema.GroupVersionKind]validation.Callback{
-	eventingcorev1beta1.SchemeGroupVersion.WithKind("Broker"): eventingv1.BrokerValidationCallback(),
+	eventingcorev1.SchemeGroupVersion.WithKind("Broker"):         &eventingv1.BrokerStub{},
 }
 
 var defaultingCallbacks = map[schema.GroupVersionKind]defaulting.Callback{
@@ -131,9 +128,6 @@ func NewValidationAdmissionController(ctx context.Context, _ configmap.Watcher)
 
 		// Whether to disallow unknown fields.
 		true,
-
-		// Extra validating callbacks to be applied to resources.
-		validationCallbacks,
 	)
 }
 
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation.go b/control-plane/pkg/apis/eventing/v1/broker_validation.go
index fcfd551b8..88465108a 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation.go
@@ -21,25 +21,39 @@ import (
 	"strings"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+
+	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
+
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
-	"knative.dev/pkg/webhook"
-	"knative.dev/pkg/webhook/resourcesemantics/validation"
-
-	"knative.dev/eventing-kafka-broker/control-plane/pkg/kafka"
+	"knative.dev/pkg/webhook/resourcesemantics"
 )
 
-type BrokerPartial struct {
+type BrokerStub struct {
+	metav1.TypeMeta `json:",inline"`
+
 	// +optional
 	metav1.ObjectMeta `json:"metadata,omitempty"`
-	// Spec defines the desired state of the Broker.
-	Spec BrokerPartialSpec `json:"spec,omitempty"`
+
+	Spec eventing.BrokerSpec `json:"spec,omitempty"`
+
+	// +optional
+	Status eventing.BrokerStatus `json:"status,omitempty"`
 }
 
-type BrokerPartialSpec struct {
+var (
+	// Check that Broker is resourcesemantics.GenericCRD.
+	// We do this to make sure the validating/defaulting webhooks can operate on our stub.
+	// If we don't do this stubbing, we will end up doing the defaulting/validating using the upstream Broker type.
+	// However, we want to let the upstream webhook to do its job and not us. We only want to default/validate
+	// the parts relevant to us.
+	_ resourcesemantics.GenericCRD = (*BrokerStub)(nil)
+)
+
+type BrokerStubSpec struct {
 	// Config is a KReference to the configuration that specifies
 	// configuration options for this Broker. For example, this could be
 	// a pointer to a ConfigMap.
@@ -47,27 +61,55 @@ type BrokerPartialSpec struct {
 	Config *duckv1.KReference `json:"config,omitempty"`
 }
 
-func (b BrokerPartialSpec) Validate(context.Context) error {
-	if b.Config == nil {
-		return apis.ErrMissingField("config").ViaField("spec")
+func (b *BrokerStub) Validate(context.Context) *apis.FieldError {
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass && b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.NamespacedBrokerClass {
+	if b.Annotations[eventing.BrokerClassAnnotationKey] != kafka.BrokerClass {
+		// validation for the broker of other classes is done by the other webhooks
+		return nil
+	}
+
+	if b.Spec.Config == nil {
+		// validation of the Config requiredness is done by the upstream webhook and we want to let it do its job.
+		// otherwise API server will say that our Kafka admission controller refused it, but it is a generic validation.
+		return nil
 	}
-	if strings.ToLower(b.Config.Kind) != "configmap" {
-		return apis.ErrInvalidValue(b.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
+
+	// we specifically expect our broker class to use ConfigMap as the config type
+	if strings.ToLower(b.Spec.Config.Kind) != "configmap" {
+		return apis.ErrInvalidValue(b.Spec.Config.Kind, "kind", "Expected ConfigMap").ViaField("config").ViaField("spec")
 	}
+
+	// we specifically expect for our broker classes, that there's a namespace in the config
+	if b.Spec.Config.Namespace == "" {
+		return apis.ErrMissingField(b.Spec.Config.Namespace, "namespace").ViaField("config").ViaField("spec")
+	}
+
+	// TODO: uncomment the following check when we backport namespaced broker into 1.6
+	// for the namespaced broker, we expect the config to be in the same namespace as the broker
+	//if b.Annotations[eventing.BrokerClassAnnotationKey] == kafka.NamespacedBrokerClass && b.Spec.Config.Namespace != b.Namespace {
+	//	return apis.ErrInvalidValue(b.Spec.Config.Namespace, "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec")
+	//}
+
 	return nil
 }
 
-func validateBrokerFromUnstructured(ctx context.Context, unstructured *unstructured.Unstructured) error {
-	broker := BrokerPartial{}
-	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstructured.UnstructuredContent(), &broker); err != nil {
-		return err
-	}
-	if class, ok := broker.Annotations[eventing.BrokerClassAnnotationKey]; !ok || class != kafka.BrokerClass {
+func (b *BrokerStub) SetDefaults(ctx context.Context) {
+	// do nothing, we let the upstream webhook do its job
+}
+
+func (b *BrokerStub) DeepCopyObject() runtime.Object {
+	if b == nil {
 		return nil
 	}
-	return broker.Spec.Validate(ctx)
-}
 
-func BrokerValidationCallback() validation.Callback {
-	return validation.NewCallback(validateBrokerFromUnstructured, webhook.Create, webhook.Update)
+	out := &BrokerStub{
+		TypeMeta: b.TypeMeta, //simple struct
+	}
+
+	b.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
+	b.Spec.DeepCopyInto(&out.Spec)
+	b.Status.DeepCopyInto(&out.Status)
+
+	return out
 }


Test patch:
diff --git a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
index 0c08ba763..207dabf26 100644
--- a/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
+++ b/control-plane/pkg/apis/eventing/v1/broker_validation_test.go
@@ -20,88 +20,124 @@ import (
 	"context"
 	"testing"
 
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"github.com/google/go-cmp/cmp"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
-func TestValidateBrokerFromUnstructured(t *testing.T) {
+func TestValidate(t *testing.T) {
 	tests := []struct {
-		name         string
-		ctx          context.Context
-		unstructured *unstructured.Unstructured
-		wantErr      bool
-	}{
-		{
-			name: "no kafka broker",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{"spec": map[string]interface{}{}},
+		name string
+		b    BrokerStub
+		want *apis.FieldError
+	}{{
+		name: "missing annotation",
+		b:    BrokerStub{},
+	}, {
+		name: "empty annotation",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": ""},
 			},
-			wantErr: false,
 		},
-		{
-			name: "missing config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{},
+	}, {
+		name: "other broker class",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Foo"},
+			},
+		},
+	}, {
+		name: "no spec.config",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+		},
+	}, {
+		name: "no spec.config.namespace",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind: "ConfigMap",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "unknown kind",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "eventing.knative.dev/v1",
-							"kind":       "Broker",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrMissingField("", "namespace").ViaField("config").ViaField("spec"),
+	}, {
+		name: "spec.config is not configmap",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Kind:      "Service",
+					Namespace: "foo",
 				},
 			},
-			wantErr: true,
 		},
-		{
-			name: "ConfigMap in config",
-			ctx:  context.Background(),
-			unstructured: &unstructured.Unstructured{
-				Object: map[string]interface{}{
-					"metadata": map[string]interface{}{
-						"annotations": map[string]string{
-							"eventing.knative.dev/broker.class": "Kafka",
-						},
-					},
-					"spec": map[string]interface{}{
-						"config": map[string]string{
-							"apiVersion": "v1",
-							"kind":       "ConfigMap",
-							"namespace":  "ns",
-							"name":       "name",
-						},
-					},
+		want: apis.ErrInvalidValue("Service", "kind", "Expected ConfigMap").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "spec.config.namespace is different",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Kind:      "ConfigMap",
+		//				Namespace: "foo",
+		//			},
+		//		},
+		//	},
+		//	want: apis.ErrInvalidValue("foo", "namespace", "Expected ConfigMap in same namespace with broker resource").ViaField("config").ViaField("spec"),
+		// TODO: uncomment the following check when we backport namespaced broker into 1.6
+		//}, {
+		//	name: "valid config - namespaced broker",
+		//	b: BrokerStub{
+		//		ObjectMeta: metav1.ObjectMeta{
+		//			Namespace:   "my-namespace",
+		//			Annotations: map[string]string{"eventing.knative.dev/broker.class": "KafkaNamespaced"},
+		//		},
+		//		Spec: eventingv1.BrokerSpec{
+		//			Config: &duckv1.KReference{
+		//				Namespace:  "my-namespace",
+		//				Name:       "name",
+		//				Kind:       "ConfigMap",
+		//				APIVersion: "v1",
+		//			},
+		//		},
+		//	},
+	}, {
+		name: "valid config - regular broker",
+		b: BrokerStub{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace:   "my-namespace",
+				Annotations: map[string]string{"eventing.knative.dev/broker.class": "Kafka"},
+			},
+			Spec: eventingv1.BrokerSpec{
+				Config: &duckv1.KReference{
+					Name:       "name",
+					Namespace:  "my-namespace",
+					Kind:       "ConfigMap",
+					APIVersion: "v1",
 				},
 			},
-			wantErr: false,
 		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if err := validateBrokerFromUnstructured(tt.ctx, tt.unstructured); (err != nil) != tt.wantErr {
-				t.Errorf("validateBrokerFromUnstructured() error = %v, wantErr %v", err, tt.wantErr)
+	}}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := test.b.Validate(context.Background())
+			if diff := cmp.Diff(test.want.Error(), got.Error()); diff != "" {
+				t.Error("Broker.Validate (-want, +got) =", diff)
 			}
 		})
 	}