diff --git a/cmd/main.go b/cmd/main.go
index 95ab7d61..0eddd564 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -219,7 +219,7 @@ func main() { //nolint:funlen // main function needs to initialize many object
 	}
 
 	if errs := webhook.CleanupResources(ctx, k8sClient); len(errs) > 0 {
-		setupLog.Error(errors.Join(errs...), "unable to cleanup kubernetes webhook resources")
+		setupLog.Error(errors.Join(errs...), "failed to cleanup kubernetes webhook resources")
 		syncLogger(ctrLogger)
 		os.Exit(1)
 	}
diff --git a/internal/controller/eventing/subscription/eventmesh/reconciler.go b/internal/controller/eventing/subscription/eventmesh/reconciler.go
index 7552b74e..fcaecff1 100644
--- a/internal/controller/eventing/subscription/eventmesh/reconciler.go
+++ b/internal/controller/eventing/subscription/eventmesh/reconciler.go
@@ -28,11 +28,11 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	controllererrors "github.com/kyma-project/eventing-manager/internal/controller/errors"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/internal/controller/events"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	"github.com/kyma-project/eventing-manager/pkg/backend/eventmesh"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	"github.com/kyma-project/eventing-manager/pkg/constants"
 	"github.com/kyma-project/eventing-manager/pkg/ems/api/events/types"
@@ -55,7 +55,7 @@ type Reconciler struct {
 	oauth2credentials *eventmesh.OAuth2ClientCredentials
 	// nameMapper is used to map the Kyma subscription name to a subscription name on EventMesh.
 	nameMapper                     backendutils.NameMapper
-	sinkValidator                  sink.Validator
+	subscriptionValidator          validator.SubscriptionValidator
 	collector                      *metrics.Collector
 	syncConditionWebhookCallStatus syncConditionWebhookCallStatusFunc
 }
@@ -73,8 +73,8 @@ const (
 
 func NewReconciler(client client.Client, logger *logger.Logger, recorder record.EventRecorder,
 	cfg env.Config, cleaner cleaner.Cleaner, eventMeshBackend eventmesh.Backend,
-	credential *eventmesh.OAuth2ClientCredentials, mapper backendutils.NameMapper, validator sink.Validator,
-	collector *metrics.Collector, domain string,
+	credential *eventmesh.OAuth2ClientCredentials, mapper backendutils.NameMapper,
+	subscriptionValidator validator.SubscriptionValidator, collector *metrics.Collector, domain string,
 ) *Reconciler {
 	if err := eventMeshBackend.Initialize(cfg); err != nil {
 		logger.WithContext().Errorw("Failed to start reconciler", "name",
@@ -90,7 +90,7 @@ func NewReconciler(client client.Client, logger *logger.Logger, recorder record.
 		cleaner:                        cleaner,
 		oauth2credentials:              credential,
 		nameMapper:                     mapper,
-		sinkValidator:                  validator,
+		subscriptionValidator:          subscriptionValidator,
 		collector:                      collector,
 		syncConditionWebhookCallStatus: syncConditionWebhookCallStatus,
 	}
@@ -144,7 +144,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req kctrl.Request) (kctrl.Re
 	}
 
 	// Validate subscription.
-	if validationErr := r.handleSubscriptionValidation(ctx, sub); validationErr != nil {
+	if validationErr := r.validateSubscription(ctx, sub); validationErr != nil {
 		if updateErr := r.updateStatus(ctx, currentSubscription, sub, log); updateErr != nil {
 			return kctrl.Result{}, errors.Join(validationErr, updateErr)
 		}
@@ -184,25 +184,18 @@ func (r *Reconciler) Reconcile(ctx context.Context, req kctrl.Request) (kctrl.Re
 	return result, nil
 }
 
-func (r *Reconciler) handleSubscriptionValidation(ctx context.Context, desiredSubscription *eventingv1alpha2.Subscription) error {
+func (r *Reconciler) validateSubscription(ctx context.Context, subscription *eventingv1alpha2.Subscription) error {
 	var err error
-	if err = r.validateSubscriptionSpec(ctx, desiredSubscription); err != nil {
-		desiredSubscription.Status.SetNotReady()
-		desiredSubscription.Status.ClearTypes()
-		desiredSubscription.Status.ClearBackend()
-		desiredSubscription.Status.ClearConditions()
+	if err = r.subscriptionValidator.Validate(ctx, *subscription); err != nil {
+		subscription.Status.SetNotReady()
+		subscription.Status.ClearTypes()
+		subscription.Status.ClearBackend()
+		subscription.Status.ClearConditions()
 	}
-	desiredSubscription.Status.SetSubscriptionSpecValidCondition(err)
+	subscription.Status.SetSubscriptionSpecValidCondition(err)
 	return err
 }
 
-func (r *Reconciler) validateSubscriptionSpec(ctx context.Context, subscription *eventingv1alpha2.Subscription) error {
-	if errList := subscription.ValidateSpec(); len(errList) > 0 {
-		return errors.Join(errList.ToAggregate())
-	}
-	return r.sinkValidator.Validate(ctx, subscription)
-}
-
 // updateSubscription updates the subscription changes to k8s.
 func (r *Reconciler) updateSubscription(ctx context.Context, sub *eventingv1alpha2.Subscription, logger *zap.SugaredLogger) error {
 	namespacedName := &ktypes.NamespacedName{
diff --git a/internal/controller/eventing/subscription/eventmesh/reconciler_internal_integration_test.go b/internal/controller/eventing/subscription/eventmesh/reconciler_internal_integration_test.go
index 0171d0d6..7603a498 100644
--- a/internal/controller/eventing/subscription/eventmesh/reconciler_internal_integration_test.go
+++ b/internal/controller/eventing/subscription/eventmesh/reconciler_internal_integration_test.go
@@ -24,11 +24,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	"github.com/kyma-project/eventing-manager/pkg/backend/eventmesh"
 	"github.com/kyma-project/eventing-manager/pkg/backend/eventmesh/mocks"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	"github.com/kyma-project/eventing-manager/pkg/ems/api/events/types"
 	"github.com/kyma-project/eventing-manager/pkg/env"
@@ -79,8 +79,8 @@ func TestReconciler_Reconcile(t *testing.T) {
 	backendSyncErr := errors.New("backend sync error")
 	backendDeleteErr := errors.New("backend delete error")
 	validatorErr := errors.New("invalid sink")
-	happyValidator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return nil })
-	unhappyValidator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return validatorErr })
+	happyValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
+	unhappyValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return validatorErr })
 
 	testCases := []struct {
 		name                 string
@@ -262,7 +262,7 @@ func TestReconciler_APIRuleConfig(t *testing.T) {
 		eventingtesting.WithMaxInFlight(10),
 	)
 
-	validator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return nil })
+	subscriptionValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
 
 	col := metrics.NewCollector()
 
@@ -292,7 +292,7 @@ func TestReconciler_APIRuleConfig(t *testing.T) {
 						testenv.backend,
 						testenv.credentials,
 						testenv.mapper,
-						validator,
+						subscriptionValidator,
 						col,
 						utils.Domain),
 					testenv.fakeClient
@@ -322,7 +322,7 @@ func TestReconciler_APIRuleConfig(t *testing.T) {
 						testenv.backend,
 						testenv.credentials,
 						testenv.mapper,
-						validator,
+						subscriptionValidator,
 						col,
 						utils.Domain),
 					testenv.fakeClient
@@ -393,7 +393,7 @@ func TestReconciler_APIRuleConfig_Upgrade(t *testing.T) {
 		eventingtesting.WithMaxInFlight(10),
 	)
 
-	validator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return nil })
+	subscriptionValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
 	col := metrics.NewCollector()
 
 	testCases := []struct {
@@ -423,7 +423,7 @@ func TestReconciler_APIRuleConfig_Upgrade(t *testing.T) {
 						testenv.backend,
 						testenv.credentials,
 						testenv.mapper,
-						validator,
+						subscriptionValidator,
 						col,
 						utils.Domain),
 					testenv.fakeClient
@@ -459,7 +459,7 @@ func TestReconciler_APIRuleConfig_Upgrade(t *testing.T) {
 						testenv.backend,
 						testenv.credentials,
 						testenv.mapper,
-						validator,
+						subscriptionValidator,
 						col,
 						utils.Domain),
 					testenv.fakeClient
@@ -579,7 +579,7 @@ func TestReconciler_APIRuleConfig_Upgrade(t *testing.T) {
 func TestReconciler_PreserveBackendHashes(t *testing.T) {
 	ctx := context.Background()
 	collector := metrics.NewCollector()
-	validator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return nil })
+	subscriptionValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
 
 	const (
 		ev2hash            = int64(118518533334734)
@@ -620,7 +620,7 @@ func TestReconciler_PreserveBackendHashes(t *testing.T) {
 				te.backend.On("Initialize", mock.Anything).Return(nil)
 				te.backend.On("SyncSubscription", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
 				return NewReconciler(te.fakeClient, te.logger, te.recorder, te.cfg, te.cleaner,
-					te.backend, te.credentials, te.mapper, validator, collector, utils.Domain), te.fakeClient
+					te.backend, te.credentials, te.mapper, subscriptionValidator, collector, utils.Domain), te.fakeClient
 			},
 			wantEv2Hash:            ev2hash,
 			wantEventMeshHash:      eventMeshHash,
@@ -650,7 +650,7 @@ func TestReconciler_PreserveBackendHashes(t *testing.T) {
 				te.backend.On("Initialize", mock.Anything).Return(nil)
 				te.backend.On("SyncSubscription", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
 				return NewReconciler(te.fakeClient, te.logger, te.recorder, te.cfg, te.cleaner,
-					te.backend, te.credentials, te.mapper, validator, collector, utils.Domain), te.fakeClient
+					te.backend, te.credentials, te.mapper, subscriptionValidator, collector, utils.Domain), te.fakeClient
 			},
 			wantEv2Hash:            ev2hash,
 			wantEventMeshHash:      eventMeshHash,
diff --git a/internal/controller/eventing/subscription/eventmesh/test/reconciler_integration_test.go b/internal/controller/eventing/subscription/eventmesh/test/reconciler_integration_test.go
index 4f65d09d..25f4acbc 100644
--- a/internal/controller/eventing/subscription/eventmesh/test/reconciler_integration_test.go
+++ b/internal/controller/eventing/subscription/eventmesh/test/reconciler_integration_test.go
@@ -53,7 +53,7 @@ func Test_CreateSubscription(t *testing.T) {
 	testCases := []struct {
 		name                     string
 		givenSubscriptionFunc    func(namespace string) *eventingv1alpha2.Subscription
-		wantSubscriptionMatchers gomegatypes.GomegaMatcher
+		wantSubscriptionMatchers func(namespace string) gomegatypes.GomegaMatcher
 		wantEventMeshSubMatchers gomegatypes.GomegaMatcher
 		wantEventMeshSubCheck    bool
 		wantAPIRuleCheck         bool
@@ -74,13 +74,15 @@ func Test_CreateSubscription(t *testing.T) {
 					eventingtesting.WithMaxInFlight(10),
 				)
 			},
-			wantSubscriptionMatchers: gomega.And(
-				eventingtesting.HaveSubscriptionNotReady(),
-				eventingtesting.HaveCondition(eventingv1alpha2.MakeCondition(
-					eventingv1alpha2.ConditionSubscriptionSpecValid,
-					eventingv1alpha2.ConditionReasonSubscriptionSpecHasValidationErrors,
-					kcorev1.ConditionFalse, "Sink validation failed: Service \"invalid\" not found")),
-			),
+			wantSubscriptionMatchers: func(namespace string) gomegatypes.GomegaMatcher {
+				return gomega.And(
+					eventingtesting.HaveSubscriptionNotReady(),
+					eventingtesting.HaveCondition(eventingv1alpha2.MakeCondition(
+						eventingv1alpha2.ConditionSubscriptionSpecValid,
+						eventingv1alpha2.ConditionReasonSubscriptionSpecHasValidationErrors,
+						kcorev1.ConditionFalse, fmt.Sprintf("Subscription validation failed: Sink validation failed: service invalid.%s not found in the cluster", namespace))),
+				)
+			},
 		},
 		{
 			name: "should succeed to create subscription if types are non-empty",
@@ -96,19 +98,21 @@ func Test_CreateSubscription(t *testing.T) {
 					eventingtesting.WithMaxInFlight(10),
 				)
 			},
-			wantSubscriptionMatchers: gomega.And(
-				eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
-				eventingtesting.HaveSubscriptionActiveCondition(),
-				eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
-					{
-						OriginalType: fmt.Sprintf("%s0", eventingtesting.OrderCreatedV1EventNotClean),
-						CleanType:    fmt.Sprintf("%s0", eventingtesting.OrderCreatedV1Event),
-					}, {
-						OriginalType: fmt.Sprintf("%s1", eventingtesting.OrderCreatedV1EventNotClean),
-						CleanType:    fmt.Sprintf("%s1", eventingtesting.OrderCreatedV1Event),
-					},
-				}),
-			),
+			wantSubscriptionMatchers: func(namespace string) gomegatypes.GomegaMatcher {
+				return gomega.And(
+					eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
+					eventingtesting.HaveSubscriptionActiveCondition(),
+					eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
+						{
+							OriginalType: fmt.Sprintf("%s0", eventingtesting.OrderCreatedV1EventNotClean),
+							CleanType:    fmt.Sprintf("%s0", eventingtesting.OrderCreatedV1Event),
+						}, {
+							OriginalType: fmt.Sprintf("%s1", eventingtesting.OrderCreatedV1EventNotClean),
+							CleanType:    fmt.Sprintf("%s1", eventingtesting.OrderCreatedV1Event),
+						},
+					}),
+				)
+			},
 			wantEventMeshSubMatchers: gomega.And(
 				eventmeshsubmatchers.HaveEvents(emstypes.Events{
 					{
@@ -135,9 +139,11 @@ func Test_CreateSubscription(t *testing.T) {
 					eventingtesting.WithSinkURL(eventingtesting.ValidSinkURL(namespace, testName)),
 				)
 			},
-			wantSubscriptionMatchers: gomega.And(
-				eventingtesting.HaveSubscriptionActiveCondition(),
-			),
+			wantSubscriptionMatchers: func(namespace string) gomegatypes.GomegaMatcher {
+				return gomega.And(
+					eventingtesting.HaveSubscriptionActiveCondition(),
+				)
+			},
 			wantEventMeshSubMatchers: gomega.And(
 				// should have default values for protocol and webhook auth
 				eventmeshsubmatchers.HaveContentMode(emTestEnsemble.envConfig.ContentMode),
@@ -163,16 +169,18 @@ func Test_CreateSubscription(t *testing.T) {
 					eventingtesting.WithSinkURL(eventingtesting.ValidSinkURL(namespace, testName)),
 				)
 			},
-			wantSubscriptionMatchers: gomega.And(
-				eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
-				eventingtesting.HaveSubscriptionActiveCondition(),
-				eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
-					{
-						OriginalType: eventingtesting.EventMeshExactType,
-						CleanType:    eventingtesting.EventMeshExactType,
-					},
-				}),
-			),
+			wantSubscriptionMatchers: func(namespace string) gomegatypes.GomegaMatcher {
+				return gomega.And(
+					eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
+					eventingtesting.HaveSubscriptionActiveCondition(),
+					eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
+						{
+							OriginalType: eventingtesting.EventMeshExactType,
+							CleanType:    eventingtesting.EventMeshExactType,
+						},
+					}),
+				)
+			},
 			wantEventMeshSubMatchers: gomega.And(
 				eventmeshsubmatchers.HaveEvents(emstypes.Events{
 					{
@@ -197,16 +205,18 @@ func Test_CreateSubscription(t *testing.T) {
 					eventingtesting.WithMaxInFlight(10),
 				)
 			},
-			wantSubscriptionMatchers: gomega.And(
-				eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
-				eventingtesting.HaveSubscriptionActiveCondition(),
-				eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
-					{
-						OriginalType: eventingtesting.OrderCreatedV1EventNotClean,
-						CleanType:    eventingtesting.OrderCreatedV1Event,
-					},
-				}),
-			),
+			wantSubscriptionMatchers: func(namespace string) gomegatypes.GomegaMatcher {
+				return gomega.And(
+					eventingtesting.HaveSubscriptionFinalizer(eventingv1alpha2.Finalizer),
+					eventingtesting.HaveSubscriptionActiveCondition(),
+					eventingtesting.HaveCleanEventTypes([]eventingv1alpha2.EventType{
+						{
+							OriginalType: eventingtesting.OrderCreatedV1EventNotClean,
+							CleanType:    eventingtesting.OrderCreatedV1Event,
+						},
+					}),
+				)
+			},
 			wantEventMeshSubMatchers: gomega.And(
 				eventmeshsubmatchers.HaveEvents(emstypes.Events{
 					{
@@ -242,7 +252,7 @@ func Test_CreateSubscription(t *testing.T) {
 			ensureK8sResourceCreated(ctx, t, givenSubscription)
 
 			// check if the subscription is as required
-			getSubscriptionAssert(ctx, g, givenSubscription).Should(testcase.wantSubscriptionMatchers)
+			getSubscriptionAssert(ctx, g, givenSubscription).Should(testcase.wantSubscriptionMatchers(testNamespace))
 
 			if testcase.wantAPIRuleCheck {
 				// check if an APIRule was created for the subscription
@@ -545,14 +555,16 @@ func Test_FixingSinkAndApiRule(t *testing.T) {
 		)
 	}
 
-	wantSubscriptionWithoutSinkMatchers := gomega.And(
-		eventingtesting.HaveCondition(eventingv1alpha2.MakeCondition(
-			eventingv1alpha2.ConditionSubscriptionSpecValid,
-			eventingv1alpha2.ConditionReasonSubscriptionSpecHasValidationErrors,
-			kcorev1.ConditionFalse,
-			"Sink validation failed: Service \"invalid\" not found",
-		)),
-	)
+	wantSubscriptionWithoutSinkMatchers := func(namespace string) gomegatypes.GomegaMatcher {
+		return gomega.And(
+			eventingtesting.HaveCondition(eventingv1alpha2.MakeCondition(
+				eventingv1alpha2.ConditionSubscriptionSpecValid,
+				eventingv1alpha2.ConditionReasonSubscriptionSpecHasValidationErrors,
+				kcorev1.ConditionFalse,
+				fmt.Sprintf("Subscription validation failed: Sink validation failed: service invalid.%s not found in the cluster", namespace),
+			)),
+		)
+	}
 
 	givenUpdateSubscriptionWithSinkFunc := func(namespace, name, sinkFormat, path string) *eventingv1alpha2.Subscription {
 		return eventingtesting.NewSubscription(name, namespace,
@@ -611,7 +623,7 @@ func Test_FixingSinkAndApiRule(t *testing.T) {
 			ensureK8sResourceCreated(ctx, t, givenSubscription)
 			createdSubscription := givenSubscription.DeepCopy()
 			// check if the created subscription is correct
-			getSubscriptionAssert(ctx, g, createdSubscription).Should(wantSubscriptionWithoutSinkMatchers)
+			getSubscriptionAssert(ctx, g, createdSubscription).Should(wantSubscriptionWithoutSinkMatchers(testNamespace))
 
 			// update subscription with valid sink
 			givenUpdateSubscription.ResourceVersion = createdSubscription.ResourceVersion
diff --git a/internal/controller/eventing/subscription/eventmesh/test/utils.go b/internal/controller/eventing/subscription/eventmesh/test/utils.go
index 0b083732..d0fafa71 100644
--- a/internal/controller/eventing/subscription/eventmesh/test/utils.go
+++ b/internal/controller/eventing/subscription/eventmesh/test/utils.go
@@ -32,10 +32,10 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	subscriptioncontrollereventmesh "github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/eventmesh"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	backendeventmesh "github.com/kyma-project/eventing-manager/pkg/backend/eventmesh"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	"github.com/kyma-project/eventing-manager/pkg/constants"
 	emstypes "github.com/kyma-project/eventing-manager/pkg/ems/api/events/types"
@@ -122,11 +122,14 @@ func setupSuite() error {
 
 	// setup eventMesh reconciler
 	recorder := k8sManager.GetEventRecorderFor("eventing-controller")
-	sinkValidator := sink.NewValidator(k8sManager.GetClient(), recorder)
 	emTestEnsemble.envConfig = getEnvConfig()
 
 	eventMesh, credentials := setupEventMesh(defaultLogger)
 
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(k8sManager.GetClient())
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
+
 	col := metrics.NewCollector()
 	testReconciler := subscriptioncontrollereventmesh.NewReconciler(
 		k8sManager.GetClient(),
@@ -137,7 +140,7 @@ func setupSuite() error {
 		eventMesh,
 		credentials,
 		emTestEnsemble.nameMapper,
-		sinkValidator,
+		subscriptionValidator,
 		col,
 		testutils.Domain,
 	)
diff --git a/internal/controller/eventing/subscription/eventmesh/testwebhookauth/utils.go b/internal/controller/eventing/subscription/eventmesh/testwebhookauth/utils.go
index 30bb7adb..509f64b1 100644
--- a/internal/controller/eventing/subscription/eventmesh/testwebhookauth/utils.go
+++ b/internal/controller/eventing/subscription/eventmesh/testwebhookauth/utils.go
@@ -30,10 +30,10 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	subscriptioncontrollereventmesh "github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/eventmesh"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	backendeventmesh "github.com/kyma-project/eventing-manager/pkg/backend/eventmesh"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	emstypes "github.com/kyma-project/eventing-manager/pkg/ems/api/events/types"
 	"github.com/kyma-project/eventing-manager/pkg/env"
@@ -124,10 +124,14 @@ func setupSuite() (*eventMeshTestEnsemble, error) {
 
 	// setup eventMesh reconciler
 	recorder := k8sManager.GetEventRecorderFor("eventing-controller")
-	sinkValidator := sink.NewValidator(k8sManager.GetClient(), recorder)
 	emTestEnsemble.envConfig = getEnvConfig(emTestEnsemble)
 	eventMeshBackend = backendeventmesh.NewEventMesh(credentials, emTestEnsemble.nameMapper, defaultLogger)
 	col := metrics.NewCollector()
+
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(k8sManager.GetClient())
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
+
 	testReconciler = subscriptioncontrollereventmesh.NewReconciler(
 		k8sManager.GetClient(),
 		defaultLogger,
@@ -137,7 +141,7 @@ func setupSuite() (*eventMeshTestEnsemble, error) {
 		eventMeshBackend,
 		credentials,
 		emTestEnsemble.nameMapper,
-		sinkValidator,
+		subscriptionValidator,
 		col,
 		testutils.Domain,
 	)
diff --git a/internal/controller/eventing/subscription/jetstream/reconciler.go b/internal/controller/eventing/subscription/jetstream/reconciler.go
index e816d70d..199f2f02 100644
--- a/internal/controller/eventing/subscription/jetstream/reconciler.go
+++ b/internal/controller/eventing/subscription/jetstream/reconciler.go
@@ -20,11 +20,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/internal/controller/events"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	"github.com/kyma-project/eventing-manager/pkg/backend/jetstream"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	emerrors "github.com/kyma-project/eventing-manager/pkg/errors"
 	"github.com/kyma-project/eventing-manager/pkg/logger"
@@ -40,28 +40,28 @@ const (
 
 type Reconciler struct {
 	client.Client
-	Backend             jetstream.Backend
-	recorder            record.EventRecorder
-	logger              *logger.Logger
-	cleaner             cleaner.Cleaner
-	sinkValidator       sink.Validator
-	customEventsChannel chan event.GenericEvent
-	collector           *metrics.Collector
+	Backend               jetstream.Backend
+	recorder              record.EventRecorder
+	logger                *logger.Logger
+	cleaner               cleaner.Cleaner
+	subscriptionValidator validator.SubscriptionValidator
+	customEventsChannel   chan event.GenericEvent
+	collector             *metrics.Collector
 }
 
 func NewReconciler(client client.Client, jsBackend jetstream.Backend,
 	logger *logger.Logger, recorder record.EventRecorder, cleaner cleaner.Cleaner,
-	defaultSinkValidator sink.Validator, collector *metrics.Collector,
+	subscriptionValidator validator.SubscriptionValidator, collector *metrics.Collector,
 ) *Reconciler {
 	reconciler := &Reconciler{
-		Client:              client,
-		Backend:             jsBackend,
-		recorder:            recorder,
-		logger:              logger,
-		cleaner:             cleaner,
-		sinkValidator:       defaultSinkValidator,
-		customEventsChannel: make(chan event.GenericEvent),
-		collector:           collector,
+		Client:                client,
+		Backend:               jsBackend,
+		recorder:              recorder,
+		logger:                logger,
+		cleaner:               cleaner,
+		subscriptionValidator: subscriptionValidator,
+		customEventsChannel:   make(chan event.GenericEvent),
+		collector:             collector,
 	}
 	return reconciler
 }
@@ -131,8 +131,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req kctrl.Request) (kctrl.Re
 	}
 
 	// Validate subscription.
-	if validationErr := r.handleSubscriptionValidation(ctx, desiredSubscription); validationErr != nil {
-		if errors.Is(validationErr, sink.ErrSinkValidationFailed) {
+	if validationErr := r.validateSubscription(ctx, desiredSubscription); validationErr != nil {
+		if errors.Is(validationErr, validator.ErrSinkValidationFailed) {
 			if deleteErr := r.Backend.DeleteSubscriptionsOnly(desiredSubscription); deleteErr != nil {
 				log.Errorw("Failed to delete JetStream subscriptions", "error", deleteErr)
 				return kctrl.Result{}, deleteErr
@@ -171,25 +171,18 @@ func (r *Reconciler) Reconcile(ctx context.Context, req kctrl.Request) (kctrl.Re
 	return kctrl.Result{}, r.syncSubscriptionStatus(ctx, desiredSubscription, nil, log)
 }
 
-func (r *Reconciler) handleSubscriptionValidation(ctx context.Context, desiredSubscription *eventingv1alpha2.Subscription) error {
+func (r *Reconciler) validateSubscription(ctx context.Context, subscription *eventingv1alpha2.Subscription) error {
 	var err error
-	if err = r.validateSubscriptionSpec(ctx, desiredSubscription); err != nil {
-		desiredSubscription.Status.SetNotReady()
-		desiredSubscription.Status.ClearTypes()
-		desiredSubscription.Status.ClearBackend()
-		desiredSubscription.Status.ClearConditions()
+	if err = r.subscriptionValidator.Validate(ctx, *subscription); err != nil {
+		subscription.Status.SetNotReady()
+		subscription.Status.ClearTypes()
+		subscription.Status.ClearBackend()
+		subscription.Status.ClearConditions()
 	}
-	desiredSubscription.Status.SetSubscriptionSpecValidCondition(err)
+	subscription.Status.SetSubscriptionSpecValidCondition(err)
 	return err
 }
 
-func (r *Reconciler) validateSubscriptionSpec(ctx context.Context, subscription *eventingv1alpha2.Subscription) error {
-	if errList := subscription.ValidateSpec(); len(errList) > 0 {
-		return errors.Join(errList.ToAggregate())
-	}
-	return r.sinkValidator.Validate(ctx, subscription)
-}
-
 func (r *Reconciler) updateSubscriptionMetrics(current, desired *eventingv1alpha2.Subscription) {
 	for _, currentType := range current.Status.Backend.Types {
 		found := false
diff --git a/internal/controller/eventing/subscription/jetstream/reconciler_integration_test.go b/internal/controller/eventing/subscription/jetstream/reconciler_integration_test.go
index 1f117d8b..f03f1099 100644
--- a/internal/controller/eventing/subscription/jetstream/reconciler_integration_test.go
+++ b/internal/controller/eventing/subscription/jetstream/reconciler_integration_test.go
@@ -11,7 +11,6 @@ import (
 	gomegatypes "github.com/onsi/gomega/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	kcorev1 "k8s.io/api/core/v1"
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	eventingtesting "github.com/kyma-project/eventing-manager/testing"
@@ -187,12 +186,9 @@ func Test_CreateSubscription(t *testing.T) {
 				K8sSubscription: []gomegatypes.GomegaMatcher{
 					eventingtesting.HaveCondition(
 						ConditionInvalidSink(
-							"Sink validation failed: Service \"testapp\" not found",
+							"Subscription validation failed: Sink validation failed: service testapp.test not found in the cluster",
 						)),
 				},
-				K8sEvents: []kcorev1.Event{
-					EventInvalidSink("Sink does not correspond to a valid cluster local svc"),
-				},
 			},
 		},
 		{
diff --git a/internal/controller/eventing/subscription/jetstream/reconciler_internal_unit_test.go b/internal/controller/eventing/subscription/jetstream/reconciler_internal_unit_test.go
index b713db95..ab618553 100644
--- a/internal/controller/eventing/subscription/jetstream/reconciler_internal_unit_test.go
+++ b/internal/controller/eventing/subscription/jetstream/reconciler_internal_unit_test.go
@@ -20,11 +20,11 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	"github.com/kyma-project/eventing-manager/pkg/backend/jetstream"
 	backendjetstreammocks "github.com/kyma-project/eventing-manager/pkg/backend/jetstream/mocks"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	"github.com/kyma-project/eventing-manager/pkg/env"
 	"github.com/kyma-project/eventing-manager/pkg/logger"
 	eventingtesting "github.com/kyma-project/eventing-manager/testing"
@@ -64,8 +64,8 @@ func Test_Reconcile(t *testing.T) {
 	missingSubSyncErr := jetstream.ErrMissingSubscription
 	backendDeleteErr := errors.New("backend delete error")
 	validatorErr := errors.New("invalid sink")
-	happyValidator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return nil })
-	unhappyValidator := sink.ValidatorFunc(func(_ context.Context, s *eventingv1alpha2.Subscription) error { return validatorErr })
+	happyValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
+	unhappyValidator := validator.SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return validatorErr })
 	collector := metrics.NewCollector()
 
 	testCases := []struct {
@@ -746,16 +746,19 @@ func setupTestEnvironment(t *testing.T, objs ...client.Object) *TestEnvironment
 		t.Fatalf("initialize logger failed: %v", err)
 	}
 	jsCleaner := cleaner.NewJetStreamCleaner(defaultLogger)
-	defaultSinkValidator := sink.NewValidator(fakeClient, recorder)
+
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(fakeClient)
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
 
 	reconciler := Reconciler{
-		Backend:       mockedBackend,
-		Client:        fakeClient,
-		logger:        defaultLogger,
-		recorder:      recorder,
-		sinkValidator: defaultSinkValidator,
-		cleaner:       jsCleaner,
-		collector:     metrics.NewCollector(),
+		Backend:               mockedBackend,
+		Client:                fakeClient,
+		logger:                defaultLogger,
+		recorder:              recorder,
+		subscriptionValidator: subscriptionValidator,
+		cleaner:               jsCleaner,
+		collector:             metrics.NewCollector(),
 	}
 
 	return &TestEnvironment{
diff --git a/internal/controller/eventing/subscription/jetstream/test_utils_test.go b/internal/controller/eventing/subscription/jetstream/test_utils_test.go
index ffba32e1..9867a251 100644
--- a/internal/controller/eventing/subscription/jetstream/test_utils_test.go
+++ b/internal/controller/eventing/subscription/jetstream/test_utils_test.go
@@ -29,11 +29,11 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	subscriptioncontrollerjetstream "github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/jetstream"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/internal/controller/events"
 	backendcleaner "github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	"github.com/kyma-project/eventing-manager/pkg/backend/jetstream"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	"github.com/kyma-project/eventing-manager/pkg/env"
 	"github.com/kyma-project/eventing-manager/pkg/logger"
 	eventingtesting "github.com/kyma-project/eventing-manager/testing"
@@ -172,13 +172,17 @@ func startReconciler() error {
 	k8sClient := k8sManager.GetClient()
 	recorder := k8sManager.GetEventRecorderFor("eventing-controller-jetstream")
 
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(k8sClient)
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
+
 	jsTestEnsemble.Reconciler = subscriptioncontrollerjetstream.NewReconciler(
 		k8sClient,
 		jetStreamHandler,
 		defaultLogger,
 		recorder,
 		cleaner,
-		sink.NewValidator(k8sClient, recorder),
+		subscriptionValidator,
 		metricsCollector,
 	)
 
diff --git a/internal/controller/eventing/subscription/validator/sink.go b/internal/controller/eventing/subscription/validator/sink.go
new file mode 100644
index 00000000..26fe2629
--- /dev/null
+++ b/internal/controller/eventing/subscription/validator/sink.go
@@ -0,0 +1,62 @@
+package validator
+
+import (
+	"context"
+	"fmt"
+
+	pkgerrors "github.com/pkg/errors"
+	kcorev1 "k8s.io/api/core/v1"
+	ktypes "k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/kyma-project/eventing-manager/pkg/utils"
+)
+
+var ErrSinkValidationFailed = pkgerrors.New("Sink validation failed")
+
+type SinkValidator interface {
+	Validate(ctx context.Context, sink string) error
+}
+
+type sinkValidator struct {
+	client client.Client
+}
+
+// Perform a compile-time check.
+var _ SinkValidator = &sinkValidator{}
+
+func NewSinkValidator(client client.Client) SinkValidator {
+	return &sinkValidator{client: client}
+}
+
+func (sv sinkValidator) Validate(ctx context.Context, sink string) error {
+	_, subDomains, err := utils.GetSinkData(sink)
+	if err != nil {
+		return fmt.Errorf("%w: %w", ErrSinkValidationFailed, err)
+	}
+
+	if len(subDomains) < 2 {
+		return fmt.Errorf("%w: sink format should contain the service name.namespace", ErrSinkValidationFailed)
+	}
+
+	namespace, name := subDomains[1], subDomains[0]
+	if !sv.serviceExists(ctx, namespace, name) {
+		return fmt.Errorf("%w: service %s.%s not found in the cluster", ErrSinkValidationFailed, name, namespace)
+	}
+
+	return nil
+}
+
+func (sv sinkValidator) serviceExists(ctx context.Context, namespace, name string) bool {
+	return sv.client.Get(ctx, ktypes.NamespacedName{Namespace: namespace, Name: name}, &kcorev1.Service{}) == nil
+}
+
+// SinkValidatorFunc implements the SinkValidator interface.
+type SinkValidatorFunc func(ctx context.Context, sink string) error
+
+// Perform a compile-time check.
+var _ SinkValidator = SinkValidatorFunc(func(_ context.Context, _ string) error { return nil })
+
+func (svf SinkValidatorFunc) Validate(ctx context.Context, sink string) error {
+	return svf(ctx, sink)
+}
diff --git a/internal/controller/eventing/subscription/validator/sink_test.go b/internal/controller/eventing/subscription/validator/sink_test.go
new file mode 100644
index 00000000..301f2801
--- /dev/null
+++ b/internal/controller/eventing/subscription/validator/sink_test.go
@@ -0,0 +1,83 @@
+package validator
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	kcorev1 "k8s.io/api/core/v1"
+	kmetav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+)
+
+func TestSinkValidator(t *testing.T) {
+	// given
+	const (
+		namespaceName = "test-namespace"
+		serviceName   = "test-service"
+	)
+
+	ctx := context.Background()
+	fakeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()
+	validator := NewSinkValidator(fakeClient)
+
+	tests := []struct {
+		name         string
+		givenSink    string
+		givenService *kcorev1.Service
+		wantErr      error
+	}{
+		{
+			name:         "With empty sink",
+			givenSink:    "",
+			givenService: nil,
+			wantErr:      ErrSinkValidationFailed,
+		},
+		{
+			name:         "With invalid sink URL",
+			givenSink:    "[:invalid:url:]",
+			givenService: nil,
+			wantErr:      ErrSinkValidationFailed,
+		},
+		{
+			name:         "With invalid sink format",
+			givenSink:    "http://insuffecient",
+			givenService: nil,
+			wantErr:      ErrSinkValidationFailed,
+		},
+		{
+			name:         "With non-existing service",
+			givenSink:    fmt.Sprintf("https://%s.%s", serviceName, namespaceName),
+			givenService: nil,
+			wantErr:      ErrSinkValidationFailed,
+		},
+		{
+			name:      "With existing service",
+			givenSink: fmt.Sprintf("https://%s.%s", serviceName, namespaceName),
+			givenService: &kcorev1.Service{
+				ObjectMeta: kmetav1.ObjectMeta{Name: serviceName, Namespace: namespaceName},
+			},
+			wantErr: nil,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// create service if needed
+			if test.givenService != nil {
+				require.NoError(t, fakeClient.Create(ctx, test.givenService))
+			}
+
+			// when
+			gotErr := validator.Validate(ctx, test.givenSink)
+
+			// then
+			if test.wantErr == nil {
+				require.NoError(t, gotErr)
+			} else {
+				require.ErrorIs(t, gotErr, ErrSinkValidationFailed)
+			}
+		})
+	}
+}
diff --git a/internal/controller/eventing/subscription/validator/subscription.go b/internal/controller/eventing/subscription/validator/subscription.go
new file mode 100644
index 00000000..07080310
--- /dev/null
+++ b/internal/controller/eventing/subscription/validator/subscription.go
@@ -0,0 +1,47 @@
+package validator
+
+import (
+	"context"
+	"fmt"
+
+	pkgerrors "github.com/pkg/errors"
+
+	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
+)
+
+var ErrSubscriptionValidationFailed = pkgerrors.New("Subscription validation failed")
+
+type SubscriptionValidator interface {
+	Validate(ctx context.Context, subscription eventingv1alpha2.Subscription) error
+}
+
+type subscriptionValidator struct {
+	sinkValidator SinkValidator
+}
+
+// Perform a compile-time check.
+var _ SubscriptionValidator = &subscriptionValidator{}
+
+func NewSubscriptionValidator(sinkValidator SinkValidator) SubscriptionValidator {
+	return &subscriptionValidator{sinkValidator: sinkValidator}
+}
+
+func (sv *subscriptionValidator) Validate(ctx context.Context, subscription eventingv1alpha2.Subscription) error {
+	if errs := subscription.ValidateSpec(); len(errs) > 0 {
+		return fmt.Errorf("%w: %w", ErrSubscriptionValidationFailed, errs.ToAggregate())
+	}
+	if err := sv.sinkValidator.Validate(ctx, subscription.Spec.Sink); err != nil {
+		return fmt.Errorf("%w: %w", ErrSubscriptionValidationFailed, err)
+	}
+	return nil
+}
+
+// SubscriptionValidatorFunc implements the SinkValidator interface.
+type SubscriptionValidatorFunc func(ctx context.Context, subscription eventingv1alpha2.Subscription) error
+
+// Perform a compile-time check.
+var _ SubscriptionValidator = SubscriptionValidatorFunc(func(_ context.Context, _ eventingv1alpha2.Subscription) error { return nil })
+
+func (svf SubscriptionValidatorFunc) Validate(ctx context.Context, subscription eventingv1alpha2.Subscription) error {
+	return svf(ctx, subscription)
+}
diff --git a/internal/controller/eventing/subscription/validator/subscription_test.go b/internal/controller/eventing/subscription/validator/subscription_test.go
new file mode 100644
index 00000000..b53c73ea
--- /dev/null
+++ b/internal/controller/eventing/subscription/validator/subscription_test.go
@@ -0,0 +1,3 @@
+package validator
+
+// TODO(marcobebway): Add tests for the subscription validator.
diff --git a/internal/webhook/cleanup.go b/internal/webhook/cleanup.go
index 2d866577..a17a9b54 100644
--- a/internal/webhook/cleanup.go
+++ b/internal/webhook/cleanup.go
@@ -20,56 +20,92 @@ func CleanupResources(ctx context.Context, client kctrlclient.Client) []error {
 		mutatingWebhookConfiguration   = "subscription-mutating-webhook-configuration"
 		validatingWebhookConfiguration = "subscription-validating-webhook-configuration"
 	)
-	return []error{
-		deleteService(ctx, client, namespace, service),
-		deleteCronJob(ctx, client, namespace, cronjob),
-		deleteJob(ctx, client, namespace, job),
-		deleteMutatingWebhookConfiguration(ctx, client, namespace, mutatingWebhookConfiguration),
-		deleteValidatingWebhookConfiguration(ctx, client, namespace, validatingWebhookConfiguration),
-	}
+	var errList = make([]error, 0, 5)
+	appendIfError(errList, deleteService(ctx, client, namespace, service))
+	appendIfError(errList, deleteCronJob(ctx, client, namespace, cronjob))
+	appendIfError(errList, deleteJob(ctx, client, namespace, job))
+	appendIfError(errList, deleteMutatingWebhookConfiguration(ctx, client, namespace, mutatingWebhookConfiguration))
+	appendIfError(errList, deleteValidatingWebhookConfiguration(ctx, client, namespace, validatingWebhookConfiguration))
+	return errList
 }
 
 func deleteService(ctx context.Context, client kctrlclient.Client, namespace, name string) error {
-	return client.Delete(ctx, &kcorev1.Service{
-		ObjectMeta: kmetav1.ObjectMeta{
-			Namespace: namespace,
-			Name:      name,
-		},
-	})
+	return kctrlclient.IgnoreNotFound(
+		client.Delete(
+			ctx,
+			&kcorev1.Service{
+				ObjectMeta: kmetav1.ObjectMeta{
+					Namespace: namespace,
+					Name:      name,
+				},
+			},
+			kctrlclient.PropagationPolicy(kmetav1.DeletePropagationBackground),
+		),
+	)
 }
 
 func deleteCronJob(ctx context.Context, client kctrlclient.Client, namespace, name string) error {
-	return client.Delete(ctx, &kbatchv1.CronJob{
-		ObjectMeta: kmetav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-	})
+	return kctrlclient.IgnoreNotFound(
+		client.Delete(
+			ctx,
+			&kbatchv1.CronJob{
+				ObjectMeta: kmetav1.ObjectMeta{
+					Name:      name,
+					Namespace: namespace,
+				},
+			},
+			kctrlclient.PropagationPolicy(kmetav1.DeletePropagationBackground),
+		),
+	)
 }
 
 func deleteJob(ctx context.Context, client kctrlclient.Client, namespace, name string) error {
-	return client.Delete(ctx, &kbatchv1.Job{
-		ObjectMeta: kmetav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-	})
+	return kctrlclient.IgnoreNotFound(
+		client.Delete(
+			ctx,
+			&kbatchv1.Job{
+				ObjectMeta: kmetav1.ObjectMeta{
+					Name:      name,
+					Namespace: namespace,
+				},
+			},
+			kctrlclient.PropagationPolicy(kmetav1.DeletePropagationBackground),
+		),
+	)
 }
 
 func deleteMutatingWebhookConfiguration(ctx context.Context, client kctrlclient.Client, namespace, name string) error {
-	return client.Delete(ctx, &kadmissionregistrationv1.MutatingWebhookConfiguration{
-		ObjectMeta: kmetav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-	})
+	return kctrlclient.IgnoreNotFound(
+		client.Delete(
+			ctx,
+			&kadmissionregistrationv1.MutatingWebhookConfiguration{
+				ObjectMeta: kmetav1.ObjectMeta{
+					Name:      name,
+					Namespace: namespace,
+				},
+			},
+			kctrlclient.PropagationPolicy(kmetav1.DeletePropagationBackground),
+		),
+	)
 }
 
 func deleteValidatingWebhookConfiguration(ctx context.Context, client kctrlclient.Client, namespace, name string) error {
-	return client.Delete(ctx, &kadmissionregistrationv1.ValidatingWebhookConfiguration{
-		ObjectMeta: kmetav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-	})
+	return kctrlclient.IgnoreNotFound(
+		client.Delete(
+			ctx,
+			&kadmissionregistrationv1.ValidatingWebhookConfiguration{
+				ObjectMeta: kmetav1.ObjectMeta{
+					Name:      name,
+					Namespace: namespace,
+				},
+			},
+			kctrlclient.PropagationPolicy(kmetav1.DeletePropagationBackground),
+		),
+	)
+}
+
+func appendIfError(errList []error, err error) {
+	if err != nil {
+		errList = append(errList, err)
+	}
 }
diff --git a/pkg/backend/sink/validator.go b/pkg/backend/sink/validator.go
deleted file mode 100644
index 1682abeb..00000000
--- a/pkg/backend/sink/validator.go
+++ /dev/null
@@ -1,68 +0,0 @@
-package sink
-
-import (
-	"context"
-	"fmt"
-
-	"golang.org/x/xerrors"
-	kcorev1 "k8s.io/api/core/v1"
-	ktypes "k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/tools/record"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
-	"github.com/kyma-project/eventing-manager/internal/controller/events"
-	"github.com/kyma-project/eventing-manager/pkg/utils"
-)
-
-var ErrSinkValidationFailed = xerrors.New("Sink validation failed")
-
-type Validator interface {
-	Validate(ctx context.Context, subscription *v1alpha2.Subscription) error
-}
-
-// ValidatorFunc implements the Validator interface.
-type ValidatorFunc func(context.Context, *v1alpha2.Subscription) error
-
-func (vf ValidatorFunc) Validate(ctx context.Context, sub *v1alpha2.Subscription) error {
-	return vf(ctx, sub)
-}
-
-type defaultSinkValidator struct {
-	client   client.Client
-	recorder record.EventRecorder
-}
-
-// Perform a compile-time check.
-var _ Validator = &defaultSinkValidator{}
-
-func NewValidator(client client.Client, recorder record.EventRecorder) Validator {
-	return &defaultSinkValidator{client: client, recorder: recorder}
-}
-
-func (s defaultSinkValidator) Validate(ctx context.Context, subscription *v1alpha2.Subscription) error {
-	var svcNs, svcName string
-
-	if _, subDomains, err := utils.GetSinkData(subscription.Spec.Sink); err != nil {
-		return err
-	} else {
-		svcNs = subDomains[1]
-		svcName = subDomains[0]
-	}
-
-	if _, err := GetClusterLocalService(ctx, s.client, svcNs, svcName); err != nil {
-		events.Warn(s.recorder, subscription, events.ReasonValidationFailed, "Sink does not correspond to a valid cluster local svc")
-		return fmt.Errorf("%w: %w", ErrSinkValidationFailed, err)
-	}
-
-	return nil
-}
-
-func GetClusterLocalService(ctx context.Context, client client.Client, svcNs, svcName string) (*kcorev1.Service, error) {
-	svcLookupKey := ktypes.NamespacedName{Name: svcName, Namespace: svcNs}
-	svc := &kcorev1.Service{}
-	if err := client.Get(ctx, svcLookupKey, svc); err != nil {
-		return nil, err
-	}
-	return svc, nil
-}
diff --git a/pkg/backend/sink/validator_test.go b/pkg/backend/sink/validator_test.go
deleted file mode 100644
index a6ae8559..00000000
--- a/pkg/backend/sink/validator_test.go
+++ /dev/null
@@ -1,95 +0,0 @@
-package sink
-
-import (
-	"context"
-	"strings"
-	"testing"
-
-	"github.com/stretchr/testify/require"
-	kcorev1 "k8s.io/api/core/v1"
-	kmetav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/tools/record"
-	"sigs.k8s.io/controller-runtime/pkg/client/fake"
-
-	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
-	eventingtesting "github.com/kyma-project/eventing-manager/testing"
-)
-
-func TestSinkValidator(t *testing.T) {
-	// given
-	namespaceName := "test"
-	fakeClient := fake.NewClientBuilder().WithScheme(scheme.Scheme).Build()
-	ctx := context.Background()
-	recorder := &record.FakeRecorder{}
-	sinkValidator := NewValidator(fakeClient, recorder)
-
-	testCases := []struct {
-		name                  string
-		givenSubscriptionSink string
-		givenSvcNameToCreate  string
-		wantErrString         string
-	}{
-		{
-			name:                  "With invalid URL",
-			givenSubscriptionSink: "http://invalid Sink",
-			wantErrString:         "failed to parse subscription sink URL",
-		},
-		{
-			name:                  "With no existing svc in the cluster",
-			givenSubscriptionSink: "https://eventing-nats.test.svc.cluster.local:8080",
-			wantErrString:         "Sink validation failed",
-		},
-		{
-			name:                  "With no existing svc in the cluster, service has the wrong name",
-			givenSubscriptionSink: "https://eventing-nats.test.svc.cluster.local:8080",
-			givenSvcNameToCreate:  "test", // wrong name
-			wantErrString:         "Sink validation failed",
-		},
-		{
-			name:                  "With a valid sink",
-			givenSubscriptionSink: "https://eventing-nats.test.svc.cluster.local:8080",
-			givenSvcNameToCreate:  "eventing-nats",
-			wantErrString:         "",
-		},
-	}
-
-	for _, tC := range testCases {
-		testCase := tC
-		t.Run(testCase.name, func(t *testing.T) {
-			// given
-			sub := eventingtesting.NewSubscription(
-				"foo", namespaceName,
-				eventingtesting.WithConditions([]eventingv1alpha2.Condition{}),
-				eventingtesting.WithStatus(true),
-				eventingtesting.WithSink(testCase.givenSubscriptionSink),
-			)
-
-			// create the service if required for test
-			if testCase.givenSvcNameToCreate != "" {
-				svc := &kcorev1.Service{
-					ObjectMeta: kmetav1.ObjectMeta{
-						Name:      testCase.givenSvcNameToCreate,
-						Namespace: namespaceName,
-					},
-				}
-
-				err := fakeClient.Create(ctx, svc)
-				require.NoError(t, err)
-			}
-
-			// when
-			// call the defaultSinkValidator function
-			err := sinkValidator.Validate(ctx, sub)
-
-			// then
-			// given error should match expected error
-			if testCase.wantErrString == "" {
-				require.NoError(t, err)
-			} else {
-				substringResult := strings.Contains(err.Error(), testCase.wantErrString)
-				require.True(t, substringResult)
-			}
-		})
-	}
-}
diff --git a/pkg/subscriptionmanager/eventmesh/eventmesh.go b/pkg/subscriptionmanager/eventmesh/eventmesh.go
index 0065e579..c0398870 100644
--- a/pkg/subscriptionmanager/eventmesh/eventmesh.go
+++ b/pkg/subscriptionmanager/eventmesh/eventmesh.go
@@ -20,10 +20,10 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/eventmesh"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	backendeventmesh "github.com/kyma-project/eventing-manager/pkg/backend/eventmesh"
 	"github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	"github.com/kyma-project/eventing-manager/pkg/env"
 	"github.com/kyma-project/eventing-manager/pkg/logger"
@@ -116,6 +116,10 @@ func (c *SubscriptionManager) Start(_ env.DefaultSubscriptionConfig, params subm
 	client := c.mgr.GetClient()
 	recorder := c.mgr.GetEventRecorderFor("eventing-controller-beb")
 
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(client)
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
+
 	// Initialize v1alpha2 handler for EventMesh
 	eventMeshHandler := backendeventmesh.NewEventMesh(oauth2credential, nameMapper, c.logger)
 	eventMeshcleaner := cleaner.NewEventMeshCleaner(c.logger)
@@ -128,7 +132,7 @@ func (c *SubscriptionManager) Start(_ env.DefaultSubscriptionConfig, params subm
 		eventMeshHandler,
 		oauth2credential,
 		nameMapper,
-		sink.NewValidator(client, recorder),
+		subscriptionValidator,
 		c.collector,
 		c.domain,
 	)
diff --git a/pkg/subscriptionmanager/jetstream/jetstream.go b/pkg/subscriptionmanager/jetstream/jetstream.go
index 68795b97..50498d5b 100644
--- a/pkg/subscriptionmanager/jetstream/jetstream.go
+++ b/pkg/subscriptionmanager/jetstream/jetstream.go
@@ -18,10 +18,10 @@ import (
 
 	eventingv1alpha2 "github.com/kyma-project/eventing-manager/api/eventing/v1alpha2"
 	subscriptioncontrollerjetstream "github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/jetstream"
+	"github.com/kyma-project/eventing-manager/internal/controller/eventing/subscription/validator"
 	"github.com/kyma-project/eventing-manager/pkg/backend/cleaner"
 	backendjetstream "github.com/kyma-project/eventing-manager/pkg/backend/jetstream"
 	backendmetrics "github.com/kyma-project/eventing-manager/pkg/backend/metrics"
-	"github.com/kyma-project/eventing-manager/pkg/backend/sink"
 	backendutils "github.com/kyma-project/eventing-manager/pkg/backend/utils"
 	"github.com/kyma-project/eventing-manager/pkg/env"
 	"github.com/kyma-project/eventing-manager/pkg/logger"
@@ -91,6 +91,10 @@ func (sm *SubscriptionManager) Start(defaultSubsConfig env.DefaultSubscriptionCo
 	client := sm.mgr.GetClient()
 	recorder := sm.mgr.GetEventRecorderFor("eventing-controller-jetstream")
 
+	// Init the Subscription validator.
+	sinkValidator := validator.NewSinkValidator(client)
+	subscriptionValidator := validator.NewSubscriptionValidator(sinkValidator)
+
 	// Initialize v1alpha2 event type cleaner
 	jsCleaner := cleaner.NewJetStreamCleaner(sm.logger)
 	jetStreamHandler := backendjetstream.NewJetStream(sm.envCfg,
@@ -101,7 +105,7 @@ func (sm *SubscriptionManager) Start(defaultSubsConfig env.DefaultSubscriptionCo
 		sm.logger,
 		recorder,
 		jsCleaner,
-		sink.NewValidator(client, recorder),
+		subscriptionValidator,
 		sm.metricsCollector,
 	)
 	sm.backendv2 = jetStreamReconciler.Backend