diff --git a/control-plane/cmd/webhook-kafka/main.go b/control-plane/cmd/webhook-kafka/main.go index 662d93339d..020e4f6cae 100644 --- a/control-plane/cmd/webhook-kafka/main.go +++ b/control-plane/cmd/webhook-kafka/main.go @@ -21,6 +21,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "knative.dev/pkg/apis" "knative.dev/pkg/configmap" "knative.dev/pkg/controller" "knative.dev/pkg/injection/sharedmain" @@ -32,10 +33,11 @@ import ( "knative.dev/pkg/webhook/resourcesemantics/defaulting" "knative.dev/pkg/webhook/resourcesemantics/validation" - sourcesv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1" eventingcorev1 "knative.dev/eventing/pkg/apis/eventing/v1" "knative.dev/eventing/pkg/apis/feature" + sourcesv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1" + messagingv1beta1 "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/messaging/v1beta1" "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/core" @@ -65,7 +67,7 @@ func NewDefaultingAdmissionController(ctx context.Context, _ configmap.Watcher) // A function that infuses the context passed to Validate/SetDefaults with custom metadata. ctxFunc := func(ctx context.Context) context.Context { - return ctx + return apis.AllowDifferentNamespace(ctx) } return defaulting.NewAdmissionController(ctx, @@ -90,7 +92,7 @@ func NewPodDefaultingAdmissionController(ctx context.Context, _ configmap.Watche // A function that infuses the context passed to Validate/SetDefaults with custom metadata. ctxFunc := func(ctx context.Context) context.Context { - return ctx + return apis.AllowDifferentNamespace(ctx) } return defaulting.NewAdmissionController(ctx, @@ -121,7 +123,7 @@ func NewValidationAdmissionController(ctx context.Context, cmw configmap.Watcher // Decorate contexts with the current state of the config. ctxFunc := func(ctx context.Context) context.Context { - return featureStore.ToContext(ctx) + return apis.AllowDifferentNamespace(featureStore.ToContext(ctx)) } return validation.NewAdmissionController(ctx, diff --git a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go index dd7d043c92..c8e9ff6a7b 100644 --- a/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go +++ b/control-plane/pkg/apis/internals/kafka/eventing/v1alpha1/consumer_group_validation_test.go @@ -22,11 +22,12 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/utils/pointer" - kafkasource "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1" eventingduck "knative.dev/eventing/pkg/apis/duck/v1" "knative.dev/eventing/pkg/apis/feature" "knative.dev/pkg/apis" duckv1 "knative.dev/pkg/apis/duck/v1" + + kafkasource "knative.dev/eventing-kafka-broker/control-plane/pkg/apis/sources/v1beta1" ) const ( @@ -114,6 +115,33 @@ func TestConsumerGroup_Validate(t *testing.T) { }, wantErr: false, }, + { + name: "subscriber different namespace", + ctx: apis.AllowDifferentNamespace(context.Background()), + given: &ConsumerGroup{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-cg", + Namespace: "my-ns", + }, + Spec: ConsumerGroupSpec{ + Replicas: pointer.Int32(1), + Selector: map[string]string{"app": "app"}, + Template: ConsumerTemplateSpec{ + Spec: ConsumerSpec{ + Subscriber: duckv1.Destination{ + Ref: &duckv1.KReference{ + Kind: "Sequence", + Namespace: "ma-ns-2", + Name: "my-seq", + APIVersion: "flows.knative.dev/v1", + }, + }, + }, + }, + }, + }, + wantErr: false, + }, { name: "invalid delivery - timeout feature disabled", ctx: context.Background(),