diff --git a/examples/builtins/main.go b/examples/builtins/main.go
index 8ea173b248..c791b0bf10 100644
--- a/examples/builtins/main.go
+++ b/examples/builtins/main.go
@@ -59,14 +59,17 @@ func main() {
 	}
 
 	// Watch ReplicaSets and enqueue ReplicaSet object key
-	if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}), &handler.EnqueueRequestForObject{}); err != nil {
+	if err := c.Watch(source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}, &handler.EnqueueRequestForObject{})); err != nil {
 		entryLog.Error(err, "unable to watch ReplicaSets")
 		os.Exit(1)
 	}
 
 	// Watch Pods and enqueue owning ReplicaSet key
-	if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}),
-		handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner())); err != nil {
+	if err := c.Watch(source.Kind(
+		mgr.GetCache(),
+		&corev1.Pod{},
+		handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.ReplicaSet{}, handler.OnlyControllerOwner()),
+	)); err != nil {
 		entryLog.Error(err, "unable to watch Pods")
 		os.Exit(1)
 	}
diff --git a/pkg/builder/controller.go b/pkg/builder/controller.go
index a649f47071..8c51fa5ca7 100644
--- a/pkg/builder/controller.go
+++ b/pkg/builder/controller.go
@@ -53,14 +53,15 @@ const (
 
 // Builder builds a Controller.
 type Builder struct {
-	forInput         ForInput
-	ownsInput        []OwnsInput
-	watchesInput     []WatchesInput
-	mgr              manager.Manager
-	globalPredicates []predicate.Predicate
-	ctrl             controller.Controller
-	ctrlOptions      controller.Options
-	name             string
+	forInput           ForInput
+	ownsInput          []OwnsInput
+	watchesObjectInput []WatchesObjectInput
+	watchesSourceInput []WatchesSourceInput
+	mgr                manager.Manager
+	globalPredicates   []predicate.Predicate
+	ctrl               controller.Controller
+	ctrlOptions        controller.Options
+	name               string
 }
 
 // ControllerManagedBy returns a new controller builder that will be started by the provided Manager.
@@ -79,7 +80,7 @@ type ForInput struct {
 // For defines the type of Object being *reconciled*, and configures the ControllerManagedBy to respond to create / delete /
 // update events by *reconciling the object*.
 // This is the equivalent of calling
-// Watches(&source.Kind{Type: apiType}, &handler.EnqueueRequestForObject{}).
+// Watches(source.Kind(mgr.GetCache(), apiType, &handler.EnqueueRequestForObject{})).
 func (blder *Builder) For(object client.Object, opts ...ForOption) *Builder {
 	if blder.forInput.object != nil {
 		blder.forInput.err = fmt.Errorf("For(...) should only be called once, could not assign multiple objects for reconciliation")
@@ -120,10 +121,10 @@ func (blder *Builder) Owns(object client.Object, opts ...OwnsOption) *Builder {
 	return blder
 }
 
-// WatchesInput represents the information set by Watches method.
-type WatchesInput struct {
-	src              source.Source
-	eventHandler     handler.EventHandler
+// WatchesObjectInput represents the information set by Watches method.
+type WatchesObjectInput struct {
+	object           client.Object
+	eventhandler     handler.EventHandler
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
 }
@@ -132,10 +133,15 @@ type WatchesInput struct {
 // update events by *reconciling the object* with the given EventHandler.
 //
 // This is the equivalent of calling
-// WatchesRawSource(source.Kind(cache, object), eventHandler, opts...).
-func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
-	src := source.Kind(blder.mgr.GetCache(), object)
-	return blder.WatchesRawSource(src, eventHandler, opts...)
+// WatchesRawSource(source.Kind(scheme, object, eventhandler, opts...)).
+func (blder *Builder) Watches(object client.Object, eventhandler handler.EventHandler, opts ...WatchesObjectOption) *Builder {
+	input := WatchesObjectInput{object: object, eventhandler: eventhandler}
+	for _, opt := range opts {
+		opt.ApplyToWatchesObject(&input)
+	}
+
+	blder.watchesObjectInput = append(blder.watchesObjectInput, input)
+	return blder
 }
 
 // WatchesMetadata is the same as Watches, but forces the internal cache to only watch PartialObjectMetadata.
@@ -165,29 +171,30 @@ func (blder *Builder) Watches(object client.Object, eventHandler handler.EventHa
 // In the first case, controller-runtime will create another cache for the
 // concrete type on top of the metadata cache; this increases memory
 // consumption and leads to race conditions as caches are not in sync.
-func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
+func (blder *Builder) WatchesMetadata(object client.Object, eventhandler handler.EventHandler, opts ...WatchesObjectOption) *Builder {
 	opts = append(opts, OnlyMetadata)
-	return blder.Watches(object, eventHandler, opts...)
+	return blder.Watches(object, eventhandler, opts...)
+}
+
+// WatchesSourceInput represents the information set by Watches method.
+type WatchesSourceInput struct {
+	src source.Source
 }
 
 // WatchesRawSource exposes the lower-level ControllerManagedBy Watches functions through the builder.
-// Specified predicates are registered only for given source.
 //
 // STOP! Consider using For(...), Owns(...), Watches(...), WatchesMetadata(...) instead.
-// This method is only exposed for more advanced use cases, most users should use one of the higher level functions.
-func (blder *Builder) WatchesRawSource(src source.Source, eventHandler handler.EventHandler, opts ...WatchesOption) *Builder {
-	input := WatchesInput{src: src, eventHandler: eventHandler}
-	for _, opt := range opts {
-		opt.ApplyToWatches(&input)
-	}
-
-	blder.watchesInput = append(blder.watchesInput, input)
+// This method is only exposed for more advanced use cases, most users should use higher level functions.
+// This method does generally disregard all the global configuration set by the builder.
+func (blder *Builder) WatchesRawSource(src source.Source) *Builder {
+	blder.watchesSourceInput = append(blder.watchesSourceInput, WatchesSourceInput{src: src})
 	return blder
 }
 
 // WithEventFilter sets the event filters, to filter which create/update/delete/generic events eventually
 // trigger reconciliations. For example, filtering on whether the resource version has changed.
 // Given predicate is added for all watched objects.
+// The predicates are not applied to sources watched with WatchesRawSource(...).
 // Defaults to the empty list.
 func (blder *Builder) WithEventFilter(p predicate.Predicate) *Builder {
 	blder.globalPredicates = append(blder.globalPredicates, p)
@@ -271,11 +278,14 @@ func (blder *Builder) doWatch() error {
 		if err != nil {
 			return err
 		}
-		src := source.Kind(blder.mgr.GetCache(), obj)
-		hdler := &handler.EnqueueRequestForObject{}
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, blder.forInput.predicates...)
-		if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
+		src := source.Kind(
+			blder.mgr.GetCache(),
+			obj,
+			handler.WithPredicates(&handler.EnqueueRequestForObject{}, allPredicates...),
+		)
+		if err := blder.ctrl.Watch(src); err != nil {
 			return err
 		}
 	}
@@ -289,7 +299,6 @@ func (blder *Builder) doWatch() error {
 		if err != nil {
 			return err
 		}
-		src := source.Kind(blder.mgr.GetCache(), obj)
 		opts := []handler.OwnerOption{}
 		if !own.matchEveryOwner {
 			opts = append(opts, handler.OnlyControllerOwner())
@@ -301,32 +310,43 @@ func (blder *Builder) doWatch() error {
 		)
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, own.predicates...)
-		if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
+		src := source.Kind(
+			blder.mgr.GetCache(),
+			obj,
+			handler.WithPredicates(hdler, allPredicates...),
+		)
+		if err := blder.ctrl.Watch(src); err != nil {
 			return err
 		}
 	}
 
 	// Do the watch requests
-	if len(blder.watchesInput) == 0 && blder.forInput.object == nil {
+	if len(blder.watchesObjectInput) == 0 && len(blder.watchesSourceInput) == 0 && blder.forInput.object == nil {
 		return errors.New("there are no watches configured, controller will never get triggered. Use For(), Owns() or Watches() to set them up")
 	}
-	for _, w := range blder.watchesInput {
-		// If the source of this watch is of type Kind, project it.
-		if srcKind, ok := w.src.(interface {
-			ProjectObject(func(client.Object) (client.Object, error)) error
-		}); ok {
-			if err := srcKind.ProjectObject(func(o client.Object) (client.Object, error) {
-				return blder.project(o, w.objectProjection)
-			}); err != nil {
-				return err
-			}
+	for _, w := range blder.watchesObjectInput {
+		obj, err := blder.project(w.object, w.objectProjection)
+		if err != nil {
+			return err
 		}
+
 		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 		allPredicates = append(allPredicates, w.predicates...)
-		if err := blder.ctrl.Watch(w.src, w.eventHandler, allPredicates...); err != nil {
+		src := source.Kind(
+			blder.mgr.GetCache(), obj,
+			handler.WithPredicates(w.eventhandler, allPredicates...),
+		)
+		if err := blder.ctrl.Watch(src); err != nil {
 			return err
 		}
 	}
+
+	for _, w := range blder.watchesSourceInput {
+		if err := blder.ctrl.Watch(w.src); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
diff --git a/pkg/builder/options.go b/pkg/builder/options.go
index 15f66b2a82..1ffc7c5287 100644
--- a/pkg/builder/options.go
+++ b/pkg/builder/options.go
@@ -34,10 +34,10 @@ type OwnsOption interface {
 	ApplyToOwns(*OwnsInput)
 }
 
-// WatchesOption is some configuration that modifies options for a watches request.
-type WatchesOption interface {
-	// ApplyToWatches applies this configuration to the given watches options.
-	ApplyToWatches(*WatchesInput)
+// WatchesObjectOption is some configuration that modifies options for a watches request.
+type WatchesObjectOption interface {
+	// ApplyToWatchesObject applies this configuration to the given watches options.
+	ApplyToWatchesObject(*WatchesObjectInput)
 }
 
 // }}}
@@ -66,14 +66,14 @@ func (w Predicates) ApplyToOwns(opts *OwnsInput) {
 	opts.predicates = w.predicates
 }
 
-// ApplyToWatches applies this configuration to the given WatchesInput options.
-func (w Predicates) ApplyToWatches(opts *WatchesInput) {
+// ApplyToWatchesObject applies this configuration to the given WatchesInput options.
+func (w Predicates) ApplyToWatchesObject(opts *WatchesObjectInput) {
 	opts.predicates = w.predicates
 }
 
 var _ ForOption = &Predicates{}
 var _ OwnsOption = &Predicates{}
-var _ WatchesOption = &Predicates{}
+var _ WatchesObjectOption = &Predicates{}
 
 // }}}
 
@@ -94,8 +94,8 @@ func (p projectAs) ApplyToOwns(opts *OwnsInput) {
 	opts.objectProjection = objectProjection(p)
 }
 
-// ApplyToWatches applies this configuration to the given WatchesInput options.
-func (p projectAs) ApplyToWatches(opts *WatchesInput) {
+// ApplyToWatchesObject applies this configuration to the given WatchesObjectInput options.
+func (p projectAs) ApplyToWatchesObject(opts *WatchesObjectInput) {
 	opts.objectProjection = objectProjection(p)
 }
 
@@ -132,9 +132,9 @@ var (
 	// consumption and leads to race conditions as caches are not in sync.
 	OnlyMetadata = projectAs(projectAsMetadata)
 
-	_ ForOption     = OnlyMetadata
-	_ OwnsOption    = OnlyMetadata
-	_ WatchesOption = OnlyMetadata
+	_ ForOption           = OnlyMetadata
+	_ OwnsOption          = OnlyMetadata
+	_ WatchesObjectOption = OnlyMetadata
 )
 
 // }}}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index e48db41f94..4edf01ed85 100644
--- a/pkg/controller/controller.go
+++ b/pkg/controller/controller.go
@@ -25,10 +25,8 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/internal/controller"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/ratelimiter"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -72,13 +70,9 @@ type Controller interface {
 	// Reconciler is called to reconcile an object by Namespace/Name
 	reconcile.Reconciler
 
-	// Watch takes events provided by a Source and uses the EventHandler to
-	// enqueue reconcile.Requests in response to the events.
-	//
-	// Watch may be provided one or more Predicates to filter events before
-	// they are given to the EventHandler.  Events will be passed to the
-	// EventHandler if all provided Predicates evaluate to true.
-	Watch(src source.Source, eventhandler handler.EventHandler, predicates ...predicate.Predicate) error
+	// Watch takes events provided by a Source and enqueues reconcile.Requests
+	// in response to the events.
+	Watch(src source.Source) error
 
 	// Start starts the controller.  Start blocks until the context is closed or a
 	// controller has an error starting.
diff --git a/pkg/controller/controller_integration_test.go b/pkg/controller/controller_integration_test.go
index 48facf1e94..73e5e807e4 100644
--- a/pkg/controller/controller_integration_test.go
+++ b/pkg/controller/controller_integration_test.go
@@ -64,13 +64,14 @@ var _ = Describe("controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Watching Resources")
-			err = instance.Watch(
-				source.Kind(cm.GetCache(), &appsv1.ReplicaSet{}),
+			err = instance.Watch(source.Kind(
+				cm.GetCache(),
+				&appsv1.ReplicaSet{},
 				handler.EnqueueRequestForOwner(cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}),
-			)
+			))
 			Expect(err).NotTo(HaveOccurred())
 
-			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}), &handler.EnqueueRequestForObject{})
+			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}))
 			Expect(err).NotTo(HaveOccurred())
 
 			err = cm.GetClient().Get(ctx, types.NamespacedName{Name: "foo"}, &corev1.Namespace{})
diff --git a/pkg/controller/controller_test.go b/pkg/controller/controller_test.go
index 3568ab5d89..5ad4a3de8b 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -77,7 +77,7 @@ var _ = Describe("controller.Controller", func() {
 
 			ctx, cancel := context.WithCancel(context.Background())
 			watchChan := make(chan event.GenericEvent, 1)
-			watch := source.Channel(source.NewChannelBroadcaster(watchChan))
+			watch := source.Channel(source.NewChannelBroadcaster(watchChan), &handler.EnqueueRequestForObject{})
 			watchChan <- event.GenericEvent{Object: &corev1.Pod{}}
 
 			reconcileStarted := make(chan struct{})
@@ -99,7 +99,7 @@ var _ = Describe("controller.Controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			c, err := controller.New("new-controller", m, controller.Options{Reconciler: rec})
-			Expect(c.Watch(watch, &handler.EnqueueRequestForObject{})).To(Succeed())
+			Expect(c.Watch(watch)).To(Succeed())
 			Expect(err).NotTo(HaveOccurred())
 
 			go func() {
diff --git a/pkg/controller/example_test.go b/pkg/controller/example_test.go
index d4fa1aef0b..85e9b9737b 100644
--- a/pkg/controller/example_test.go
+++ b/pkg/controller/example_test.go
@@ -71,7 +71,7 @@ func ExampleController() {
 	}
 
 	// Watch for Pod create / update / delete events and call Reconcile
-	err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{})
+	err = c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{}))
 	if err != nil {
 		log.Error(err, "unable to watch pods")
 		os.Exit(1)
@@ -108,7 +108,7 @@ func ExampleController_unstructured() {
 		Version: "v1",
 	})
 	// Watch for Pod create / update / delete events and call Reconcile
-	err = c.Watch(source.Kind(mgr.GetCache(), u), &handler.EnqueueRequestForObject{})
+	err = c.Watch(source.Kind(mgr.GetCache(), u, &handler.EnqueueRequestForObject{}))
 	if err != nil {
 		log.Error(err, "unable to watch pods")
 		os.Exit(1)
@@ -139,7 +139,7 @@ func ExampleNewUnmanaged() {
 		os.Exit(1)
 	}
 
-	if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{}); err != nil {
+	if err := c.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{})); err != nil {
 		log.Error(err, "unable to watch pods")
 		os.Exit(1)
 	}
diff --git a/pkg/handler/example_test.go b/pkg/handler/example_test.go
index 575ea05fca..cd73fb53fc 100644
--- a/pkg/handler/example_test.go
+++ b/pkg/handler/example_test.go
@@ -42,8 +42,7 @@ var (
 func ExampleEnqueueRequestForObject() {
 	// controller is a controller.controller
 	err := c.Watch(
-		source.Kind(mgr.GetCache(), &corev1.Pod{}),
-		&handler.EnqueueRequestForObject{},
+		source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{}),
 	)
 	if err != nil {
 		// handle it
@@ -54,10 +53,11 @@ func ExampleEnqueueRequestForObject() {
 // owning (direct) Deployment responsible for the creation of the ReplicaSet.
 func ExampleEnqueueRequestForOwner() {
 	// controller is a controller.controller
-	err := c.Watch(
-		source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}),
+	err := c.Watch(source.Kind(
+		mgr.GetCache(),
+		&appsv1.ReplicaSet{},
 		handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.Deployment{}, handler.OnlyControllerOwner()),
-	)
+	))
 	if err != nil {
 		// handle it
 	}
@@ -67,8 +67,9 @@ func ExampleEnqueueRequestForOwner() {
 // objects (of Type: MyKind) using a mapping function defined by the user.
 func ExampleEnqueueRequestsFromMapFunc() {
 	// controller is a controller.controller
-	err := c.Watch(
-		source.Kind(mgr.GetCache(), &appsv1.Deployment{}),
+	err := c.Watch(source.Kind(
+		mgr.GetCache(),
+		&appsv1.Deployment{},
 		handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []reconcile.Request {
 			return []reconcile.Request{
 				{NamespacedName: types.NamespacedName{
@@ -81,7 +82,7 @@ func ExampleEnqueueRequestsFromMapFunc() {
 				}},
 			}
 		}),
-	)
+	))
 	if err != nil {
 		// handle it
 	}
@@ -90,8 +91,9 @@ func ExampleEnqueueRequestsFromMapFunc() {
 // This example implements handler.EnqueueRequestForObject.
 func ExampleFuncs() {
 	// controller is a controller.controller
-	err := c.Watch(
-		source.Kind(mgr.GetCache(), &corev1.Pod{}),
+	err := c.Watch(source.Kind(
+		mgr.GetCache(),
+		&corev1.Pod{},
 		handler.Funcs{
 			CreateFunc: func(ctx context.Context, e event.CreateEvent, q workqueue.RateLimitingInterface) {
 				q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
@@ -118,7 +120,7 @@ func ExampleFuncs() {
 				}})
 			},
 		},
-	)
+	))
 	if err != nil {
 		// handle it
 	}
diff --git a/pkg/handler/predicates.go b/pkg/handler/predicates.go
new file mode 100644
index 0000000000..81175e755b
--- /dev/null
+++ b/pkg/handler/predicates.go
@@ -0,0 +1,47 @@
+package handler
+
+import (
+	"context"
+
+	"k8s.io/client-go/util/workqueue"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+// WithPredicates returns an EventHandler that only calls the underlying handler if all the given predicates return true.
+func WithPredicates(handler EventHandler, predicates ...predicate.Predicate) EventHandler {
+	return &Funcs{
+		CreateFunc: func(ctx context.Context, createEvent event.CreateEvent, queue workqueue.RateLimitingInterface) {
+			for _, predicate := range predicates {
+				if !predicate.Create(createEvent) {
+					return
+				}
+			}
+			handler.Create(ctx, createEvent, queue)
+		},
+		UpdateFunc: func(ctx context.Context, updateEvent event.UpdateEvent, queue workqueue.RateLimitingInterface) {
+			for _, predicate := range predicates {
+				if !predicate.Update(updateEvent) {
+					return
+				}
+			}
+			handler.Update(ctx, updateEvent, queue)
+		},
+		DeleteFunc: func(ctx context.Context, deleteEvent event.DeleteEvent, queue workqueue.RateLimitingInterface) {
+			for _, predicate := range predicates {
+				if !predicate.Delete(deleteEvent) {
+					return
+				}
+			}
+			handler.Delete(ctx, deleteEvent, queue)
+		},
+		GenericFunc: func(ctx context.Context, genericEvent event.GenericEvent, queue workqueue.RateLimitingInterface) {
+			for _, predicate := range predicates {
+				if !predicate.Generic(genericEvent) {
+					return
+				}
+			}
+			handler.Generic(ctx, genericEvent, queue)
+		},
+	}
+}
diff --git a/pkg/internal/controller/controller.go b/pkg/internal/controller/controller.go
index 33883647b9..270951dbad 100644
--- a/pkg/internal/controller/controller.go
+++ b/pkg/internal/controller/controller.go
@@ -28,11 +28,8 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/util/workqueue"
-
-	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -94,9 +91,7 @@ type Controller struct {
 
 // watchDescription contains all the information necessary to start a watch.
 type watchDescription struct {
-	src        source.Source
-	handler    handler.EventHandler
-	predicates []predicate.Predicate
+	src source.Source
 }
 
 // Reconcile implements reconcile.Reconciler.
@@ -120,7 +115,7 @@ func (c *Controller) Reconcile(ctx context.Context, req reconcile.Request) (_ re
 }
 
 // Watch implements controller.Controller.
-func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prct ...predicate.Predicate) error {
+func (c *Controller) Watch(src source.Source) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
@@ -128,12 +123,12 @@ func (c *Controller) Watch(src source.Source, evthdler handler.EventHandler, prc
 	//
 	// These watches are going to be held on the controller struct until the manager or user calls Start(...).
 	if !c.Started {
-		c.startWatches = append(c.startWatches, watchDescription{src: src, handler: evthdler, predicates: prct})
+		c.startWatches = append(c.startWatches, watchDescription{src: src})
 		return nil
 	}
 
 	c.LogConstructor(nil).Info("Starting EventSource", "source", src)
-	return src.Start(c.ctx, evthdler, c.Queue, prct...)
+	return src.Start(c.ctx, c.Queue)
 }
 
 // NeedLeaderElection implements the manager.LeaderElectionRunnable interface.
@@ -177,7 +172,7 @@ func (c *Controller) Start(ctx context.Context) error {
 		for _, watch := range c.startWatches {
 			c.LogConstructor(nil).Info("Starting EventSource", "source", fmt.Sprintf("%s", watch.src))
 
-			if err := watch.src.Start(ctx, watch.handler, c.Queue, watch.predicates...); err != nil {
+			if err := watch.src.Start(ctx, c.Queue); err != nil {
 				return err
 			}
 		}
diff --git a/pkg/internal/controller/controller_test.go b/pkg/internal/controller/controller_test.go
index 39cecdc292..f1a27852e2 100644
--- a/pkg/internal/controller/controller_test.go
+++ b/pkg/internal/controller/controller_test.go
@@ -42,7 +42,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/internal/controller/metrics"
 	"sigs.k8s.io/controller-runtime/pkg/internal/log"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 )
@@ -127,7 +126,7 @@ var _ = Describe("controller", func() {
 		It("should return an error if there is an error waiting for the informers", func() {
 			f := false
 			ctrl.startWatches = []watchDescription{{
-				src: source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}),
+				src: source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{}),
 			}}
 			ctrl.Name = "foo"
 			ctx, cancel := context.WithCancel(context.Background())
@@ -145,7 +144,7 @@ var _ = Describe("controller", func() {
 			c = &cacheWithIndefinitelyBlockingGetInformer{c}
 
 			ctrl.startWatches = []watchDescription{{
-				src: source.Kind(c, &appsv1.Deployment{}),
+				src: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}),
 			}}
 			ctrl.Name = "testcontroller"
 
@@ -163,7 +162,7 @@ var _ = Describe("controller", func() {
 			c = &cacheWithIndefinitelyBlockingGetInformer{c}
 			ctrl.startWatches = []watchDescription{{
 				src: &singnallingSourceWrapper{
-					SyncingSource: source.Kind(c, &appsv1.Deployment{}),
+					SyncingSource: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}),
 					cacheSyncDone: sourceSynced,
 				},
 			}}
@@ -191,7 +190,7 @@ var _ = Describe("controller", func() {
 			Expect(err).NotTo(HaveOccurred())
 			ctrl.startWatches = []watchDescription{{
 				src: &singnallingSourceWrapper{
-					SyncingSource: source.Kind(c, &appsv1.Deployment{}),
+					SyncingSource: source.Kind(c, &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}),
 					cacheSyncDone: sourceSynced,
 				},
 			}}
@@ -226,19 +225,21 @@ var _ = Describe("controller", func() {
 				Object: p,
 			}
 
-			ins := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1))
+			ins := source.Channel(
+				source.NewChannelBroadcaster(ch),
+				handler.Funcs{
+					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						close(processed)
+					},
+				},
+			)
 
 			// send the event to the channel
 			ch <- evt
 
 			ctrl.startWatches = []watchDescription{{
 				src: ins,
-				handler: handler.Funcs{
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						close(processed)
-					},
-				},
 			}}
 
 			go func() {
@@ -252,7 +253,10 @@ var _ = Describe("controller", func() {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 
-			ins := source.Channel(nil)
+			ins := source.Channel(
+				nil,
+				&handler.EnqueueRequestForObject{},
+			)
 			ctrl.startWatches = []watchDescription{{
 				src: ins,
 			}}
@@ -268,10 +272,10 @@ var _ = Describe("controller", func() {
 
 			ins := source.Channel(
 				source.NewChannelBroadcaster(make(<-chan event.GenericEvent)),
+				nil,
 			)
 			ctrl.startWatches = []watchDescription{{
-				src:     ins,
-				handler: nil,
+				src: ins,
 			}}
 
 			e := ctrl.Start(ctx)
@@ -280,20 +284,15 @@ var _ = Describe("controller", func() {
 		})
 
 		It("should call Start on sources with the appropriate EventHandler, Queue, and Predicates", func() {
-			pr1 := &predicate.Funcs{}
-			pr2 := &predicate.Funcs{}
-			evthdl := &handler.EnqueueRequestForObject{}
 			started := false
-			src := source.Func(func(ctx context.Context, e handler.EventHandler, q workqueue.RateLimitingInterface, p ...predicate.Predicate) error {
+			src := source.Func(func(ctx context.Context, q workqueue.RateLimitingInterface) error {
 				defer GinkgoRecover()
-				Expect(e).To(Equal(evthdl))
 				Expect(q).To(Equal(ctrl.Queue))
-				Expect(p).To(ConsistOf(pr1, pr2))
 
 				started = true
 				return nil
 			})
-			Expect(ctrl.Watch(src, evthdl, pr1, pr2)).NotTo(HaveOccurred())
+			Expect(ctrl.Watch(src)).NotTo(HaveOccurred())
 
 			// Use a cancelled context so Start doesn't block
 			ctx, cancel := context.WithCancel(context.Background())
@@ -304,13 +303,11 @@ var _ = Describe("controller", func() {
 
 		It("should return an error if there is an error starting sources", func() {
 			err := fmt.Errorf("Expected Error: could not start source")
-			src := source.Func(func(context.Context, handler.EventHandler,
-				workqueue.RateLimitingInterface,
-				...predicate.Predicate) error {
+			src := source.Func(func(context.Context, workqueue.RateLimitingInterface) error {
 				defer GinkgoRecover()
 				return err
 			})
-			Expect(ctrl.Watch(src, &handler.EnqueueRequestForObject{})).To(Succeed())
+			Expect(ctrl.Watch(src)).To(Succeed())
 
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
diff --git a/pkg/internal/recorder/recorder_integration_test.go b/pkg/internal/recorder/recorder_integration_test.go
index 130a306053..f9a20737fc 100644
--- a/pkg/internal/recorder/recorder_integration_test.go
+++ b/pkg/internal/recorder/recorder_integration_test.go
@@ -56,7 +56,7 @@ var _ = Describe("recorder", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Watching Resources")
-			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}), &handler.EnqueueRequestForObject{})
+			err = instance.Watch(source.Kind(cm.GetCache(), &appsv1.Deployment{}, &handler.EnqueueRequestForObject{}))
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Starting the Manager")
diff --git a/pkg/predicate/predicate.go b/pkg/predicate/predicate.go
index 314635875e..b50c0c529e 100644
--- a/pkg/predicate/predicate.go
+++ b/pkg/predicate/predicate.go
@@ -177,9 +177,13 @@ func (GenerationChangedPredicate) Update(e event.UpdateEvent) bool {
 // It is intended to be used in conjunction with the GenerationChangedPredicate, as in the following example:
 //
 //	Controller.Watch(
-//		&source.Kind{Type: v1.MyCustomKind},
-//		&handler.EnqueueRequestForObject{},
-//		predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{}))
+//		source.Kind(
+//			mgr.GetCache(),
+//			v1.MyCustomKind,
+//			&handler.EnqueueRequestForObject{},
+//			predicate.Or(predicate.GenerationChangedPredicate{}, predicate.AnnotationChangedPredicate{}))
+//		),
+//	)
 //
 // This is mostly useful for controllers that needs to trigger both when the resource's generation is incremented
 // (i.e., when the resource' .spec changes), or an annotation changes (e.g., for a staging/alpha API).
@@ -206,11 +210,14 @@ func (AnnotationChangedPredicate) Update(e event.UpdateEvent) bool {
 // This predicate will skip update events that have no change in the object's label.
 // It is intended to be used in conjunction with the GenerationChangedPredicate, as in the following example:
 //
-// Controller.Watch(
-//
-//	&source.Kind{Type: v1.MyCustomKind},
-//	&handler.EnqueueRequestForObject{},
-//	predicate.Or(predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{}))
+//	Controller.Watch(
+//		source.Kind(
+//			mgr.GetCache(),
+//			v1.MyCustomKind,
+//			&handler.EnqueueRequestForObject{},
+//			predicate.Or(predicate.GenerationChangedPredicate{}, predicate.LabelChangedPredicate{}))
+//		),
+//	)
 //
 // This will be helpful when object's labels is carrying some extra specification information beyond object's spec,
 // and the controller will be triggered if any valid spec change (not only in spec, but also in labels) happens.
diff --git a/pkg/source/example_test.go b/pkg/source/example_test.go
index fc96dc4d8d..4a668f7158 100644
--- a/pkg/source/example_test.go
+++ b/pkg/source/example_test.go
@@ -31,7 +31,7 @@ var ctrl controller.Controller
 // This example Watches for Pod Events (e.g. Create / Update / Delete) and enqueues a reconcile.Request
 // with the Name and Namespace of the Pod.
 func ExampleKind() {
-	err := ctrl.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}), &handler.EnqueueRequestForObject{})
+	err := ctrl.Watch(source.Kind(mgr.GetCache(), &corev1.Pod{}, &handler.EnqueueRequestForObject{}))
 	if err != nil {
 		// handle it
 	}
@@ -45,8 +45,8 @@ func ExampleChannel() {
 	err := ctrl.Watch(
 		source.Channel(
 			source.NewChannelBroadcaster(events),
+			&handler.EnqueueRequestForObject{},
 		),
-		&handler.EnqueueRequestForObject{},
 	)
 	if err != nil {
 		// handle it
diff --git a/pkg/source/internal/channel.go b/pkg/source/internal/channel.go
index cccb980feb..d1ccae51e7 100644
--- a/pkg/source/internal/channel.go
+++ b/pkg/source/internal/channel.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // ChannelOptions contains the options for the Channel source.
@@ -43,6 +42,9 @@ type Channel struct {
 	// Broadcaster contains the source channel for events.
 	Broadcaster *ChannelBroadcaster
 
+	// EventHandler is the handler to call when events are received.
+	EventHandler handler.EventHandler
+
 	mu sync.Mutex
 	// isStarted is true if the source has been started. A source can only be started once.
 	isStarted bool
@@ -53,16 +55,11 @@ func (cs *Channel) String() string {
 }
 
 // Start implements Source and should only be called by the Controller.
-func (cs *Channel) Start(
-	ctx context.Context,
-	handler handler.EventHandler,
-	queue workqueue.RateLimitingInterface,
-	prct ...predicate.Predicate) error {
-	// Broadcaster should have been specified by the user.
+func (cs *Channel) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error {
 	if cs.Broadcaster == nil {
 		return fmt.Errorf("must create Channel with a non-nil Broadcaster")
 	}
-	if handler == nil {
+	if cs.EventHandler == nil {
 		return fmt.Errorf("must create Channel with a non-nil EventHandler")
 	}
 
@@ -87,8 +84,6 @@ func (cs *Channel) Start(
 			ctx,
 			destination,
 			queue,
-			handler,
-			prct,
 		)
 	}()
 
@@ -99,10 +94,7 @@ func (cs *Channel) processReceivedEvents(
 	ctx context.Context,
 	destination <-chan event.GenericEvent,
 	queue workqueue.RateLimitingInterface,
-	eventHandler handler.EventHandler,
-	predicates []predicate.Predicate,
 ) {
-eventloop:
 	for {
 		select {
 		case <-ctx.Done():
@@ -112,16 +104,8 @@ eventloop:
 				return
 			}
 
-			// Check predicates against the event first
-			// and continue the outer loop if any of them fail.
-			for _, p := range predicates {
-				if !p.Generic(event) {
-					continue eventloop
-				}
-			}
-
 			// Call the event handler with the event.
-			eventHandler.Generic(ctx, event, queue)
+			cs.EventHandler.Generic(ctx, event, queue)
 		}
 	}
 }
diff --git a/pkg/source/internal/event_handler.go b/pkg/source/internal/event_handler.go
index ae8404a1fa..43edadb977 100644
--- a/pkg/source/internal/event_handler.go
+++ b/pkg/source/internal/event_handler.go
@@ -26,19 +26,16 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
-
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 var log = logf.RuntimeLog.WithName("source").WithName("EventHandler")
 
 // NewEventHandler creates a new EventHandler.
-func NewEventHandler(ctx context.Context, queue workqueue.RateLimitingInterface, handler handler.EventHandler, predicates []predicate.Predicate) *EventHandler {
+func NewEventHandler(ctx context.Context, queue workqueue.RateLimitingInterface, handler handler.EventHandler) *EventHandler {
 	return &EventHandler{
-		ctx:        ctx,
-		handler:    handler,
-		queue:      queue,
-		predicates: predicates,
+		ctx:     ctx,
+		handler: handler,
+		queue:   queue,
 	}
 }
 
@@ -48,9 +45,8 @@ type EventHandler struct {
 	// that is used to propagate cancellation signals to each handler function.
 	ctx context.Context
 
-	handler    handler.EventHandler
-	queue      workqueue.RateLimitingInterface
-	predicates []predicate.Predicate
+	handler handler.EventHandler
+	queue   workqueue.RateLimitingInterface
 }
 
 // HandlerFuncs converts EventHandler to a ResourceEventHandlerFuncs
@@ -76,12 +72,6 @@ func (e *EventHandler) OnAdd(obj interface{}) {
 		return
 	}
 
-	for _, p := range e.predicates {
-		if !p.Create(c) {
-			return
-		}
-	}
-
 	// Invoke create handler
 	ctx, cancel := context.WithCancel(e.ctx)
 	defer cancel()
@@ -109,12 +99,6 @@ func (e *EventHandler) OnUpdate(oldObj, newObj interface{}) {
 		return
 	}
 
-	for _, p := range e.predicates {
-		if !p.Update(u) {
-			return
-		}
-	}
-
 	// Invoke update handler
 	ctx, cancel := context.WithCancel(e.ctx)
 	defer cancel()
@@ -157,12 +141,6 @@ func (e *EventHandler) OnDelete(obj interface{}) {
 		return
 	}
 
-	for _, p := range e.predicates {
-		if !p.Delete(d) {
-			return
-		}
-	}
-
 	// Invoke delete handler
 	ctx, cancel := context.WithCancel(e.ctx)
 	defer cancel()
diff --git a/pkg/source/internal/func.go b/pkg/source/internal/func.go
index 9a0e43bb09..3db10b54fc 100644
--- a/pkg/source/internal/func.go
+++ b/pkg/source/internal/func.go
@@ -21,17 +21,14 @@ import (
 	"fmt"
 
 	"k8s.io/client-go/util/workqueue"
-	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // Func is a function that implements Source.
-type Func func(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
+type Func func(context.Context, workqueue.RateLimitingInterface) error
 
 // Start implements Source.
-func (f Func) Start(ctx context.Context, evt handler.EventHandler, queue workqueue.RateLimitingInterface,
-	pr ...predicate.Predicate) error {
-	return f(ctx, evt, queue, pr...)
+func (f Func) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error {
+	return f(ctx, queue)
 }
 
 func (f Func) String() string {
diff --git a/pkg/source/internal/informer.go b/pkg/source/internal/informer.go
index 13350a7cff..65b321e9d9 100644
--- a/pkg/source/internal/informer.go
+++ b/pkg/source/internal/informer.go
@@ -24,7 +24,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create).
@@ -32,6 +31,9 @@ type Informer struct {
 	// Informer is the controller-runtime Informer
 	Informer cache.Informer
 
+	// EventHandler is the handler to call when events are received.
+	EventHandler handler.EventHandler
+
 	mu sync.Mutex
 	// isStarted is true if the source has been started. A source can only be started once.
 	isStarted bool
@@ -39,12 +41,11 @@ type Informer struct {
 
 // Start is internal and should be called only by the Controller to register an EventHandler with the Informer
 // to enqueue reconcile.Requests.
-func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
-	prct ...predicate.Predicate) error {
+func (is *Informer) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error {
 	if is.Informer == nil {
 		return fmt.Errorf("must create Informer with a non-nil Informer")
 	}
-	if handler == nil {
+	if is.EventHandler == nil {
 		return fmt.Errorf("must create Informer with a non-nil EventHandler")
 	}
 
@@ -55,7 +56,7 @@ func (is *Informer) Start(ctx context.Context, handler handler.EventHandler, que
 	}
 	is.isStarted = true
 
-	_, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs())
+	_, err := is.Informer.AddEventHandler(NewEventHandler(ctx, queue, is.EventHandler).HandlerFuncs())
 	if err != nil {
 		return err
 	}
diff --git a/pkg/source/internal/internal_test.go b/pkg/source/internal/internal_test.go
index c7b7aa4847..6446fc3d86 100644
--- a/pkg/source/internal/internal_test.go
+++ b/pkg/source/internal/internal_test.go
@@ -74,7 +74,7 @@ var _ = Describe("Internal", func() {
 				set = true
 			},
 		}
-		instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, funcs, nil)
+		instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, funcs)
 	})
 
 	Describe("EventHandler", func() {
@@ -99,41 +99,47 @@ var _ = Describe("Internal", func() {
 		})
 
 		It("should used Predicates to filter CreateEvents", func() {
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }},
-			})
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{},
+				handler.WithPredicates(setfuncs, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }}),
+			)
 			set = false
 			instance.OnAdd(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-			})
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{},
+				handler.WithPredicates(setfuncs, predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }}),
+			)
 			instance.OnAdd(pod)
 			Expect(set).To(BeTrue())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }},
-			})
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{},
+				handler.WithPredicates(setfuncs,
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }},
+				),
+			)
 			instance.OnAdd(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }},
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-			})
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{},
+				handler.WithPredicates(setfuncs,
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return false }},
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
+				),
+			)
 			instance.OnAdd(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-			})
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{},
+				handler.WithPredicates(setfuncs,
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
+					predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
+				),
+			)
 			instance.OnAdd(pod)
 			Expect(set).To(BeTrue())
 		})
@@ -157,40 +163,40 @@ var _ = Describe("Internal", func() {
 
 		It("should used Predicates to filter UpdateEvents", func() {
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{UpdateFunc: func(updateEvent event.UpdateEvent) bool { return false }},
-			})
+			))
 			instance.OnUpdate(pod, newPod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }},
-			})
+			))
 			instance.OnUpdate(pod, newPod)
 			Expect(set).To(BeTrue())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }},
 				predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return false }},
-			})
+			))
 			instance.OnUpdate(pod, newPod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return false }},
 				predicate.Funcs{UpdateFunc: func(event.UpdateEvent) bool { return true }},
-			})
+			))
 			instance.OnUpdate(pod, newPod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
 				predicate.Funcs{CreateFunc: func(event.CreateEvent) bool { return true }},
-			})
+			))
 			instance.OnUpdate(pod, newPod)
 			Expect(set).To(BeTrue())
 		})
@@ -215,40 +221,40 @@ var _ = Describe("Internal", func() {
 
 		It("should used Predicates to filter DeleteEvents", func() {
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }},
-			})
+			))
 			instance.OnDelete(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }},
-			})
+			))
 			instance.OnDelete(pod)
 			Expect(set).To(BeTrue())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }},
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }},
-			})
+			))
 			instance.OnDelete(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return false }},
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }},
-			})
+			))
 			instance.OnDelete(pod)
 			Expect(set).To(BeFalse())
 
 			set = false
-			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, setfuncs, []predicate.Predicate{
+			instance = internal.NewEventHandler(ctx, &controllertest.Queue{}, handler.WithPredicates(setfuncs,
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }},
 				predicate.Funcs{DeleteFunc: func(event.DeleteEvent) bool { return true }},
-			})
+			))
 			instance.OnDelete(pod)
 			Expect(set).To(BeTrue())
 		})
diff --git a/pkg/source/internal/kind.go b/pkg/source/internal/kind.go
index 3406a09b5d..ba3931ae99 100644
--- a/pkg/source/internal/kind.go
+++ b/pkg/source/internal/kind.go
@@ -29,7 +29,6 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // Kind is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create).
@@ -39,6 +38,9 @@ type Kind struct {
 	// Cache used to watch APIs
 	Cache cache.Cache
 
+	// EventHandler is the handler to call when events are received.
+	EventHandler handler.EventHandler
+
 	// started may contain an error if one was encountered during startup. If its closed and does not
 	// contain an error, startup and syncing finished.
 	started     chan error
@@ -47,15 +49,14 @@ type Kind struct {
 
 // Start is internal and should be called only by the Controller to register an EventHandler with the Informer
 // to enqueue reconcile.Requests.
-func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue workqueue.RateLimitingInterface,
-	prct ...predicate.Predicate) error {
+func (ks *Kind) Start(ctx context.Context, queue workqueue.RateLimitingInterface) error {
 	if ks.Type == nil {
 		return fmt.Errorf("must create Kind with a non-nil Type")
 	}
 	if ks.Cache == nil {
 		return fmt.Errorf("must create Kind with a non-nil Cache")
 	}
-	if handler == nil {
+	if ks.EventHandler == nil {
 		return fmt.Errorf("must create Kind with a non-nil EventHandler")
 	}
 
@@ -101,7 +102,7 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w
 			return
 		}
 
-		_, err := i.AddEventHandler(NewEventHandler(ctx, queue, handler, prct).HandlerFuncs())
+		_, err := i.AddEventHandler(NewEventHandler(ctx, queue, ks.EventHandler).HandlerFuncs())
 		if err != nil {
 			ks.started <- err
 			return
@@ -116,23 +117,6 @@ func (ks *Kind) Start(ctx context.Context, handler handler.EventHandler, queue w
 	return nil
 }
 
-// ProjectObject sets the Kind's object to the given object.
-// This function should only be called by the Controller builder.
-// NOTE: make sure to update pkg/builder/controller.go if you change this function.
-func (ks *Kind) ProjectObject(fn func(client.Object) (client.Object, error)) error {
-	if ks.startCancel != nil {
-		return fmt.Errorf("cannot project object after Start has been called")
-	}
-
-	newType, err := fn(ks.Type)
-	if err != nil {
-		return err
-	}
-
-	ks.Type = newType
-	return nil
-}
-
 func (ks *Kind) String() string {
 	if ks.Type != nil {
 		return fmt.Sprintf("kind source: %T", ks.Type)
diff --git a/pkg/source/source.go b/pkg/source/source.go
index 74796e315c..2b2fcb8d43 100644
--- a/pkg/source/source.go
+++ b/pkg/source/source.go
@@ -22,7 +22,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
-	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	internal "sigs.k8s.io/controller-runtime/pkg/source/internal"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
@@ -39,7 +38,7 @@ import (
 type Source interface {
 	// Start is internal and should be called only by the Controller to register an EventHandler with the Informer
 	// to enqueue reconcile.Requests.
-	Start(context.Context, handler.EventHandler, workqueue.RateLimitingInterface, ...predicate.Predicate) error
+	Start(context.Context, workqueue.RateLimitingInterface) error
 }
 
 // SyncingSource is a source that needs syncing prior to being usable. The controller
@@ -56,8 +55,8 @@ var _ Source = &internal.Channel{}
 var _ Source = internal.Func(nil)
 
 // Kind creates a KindSource with the given cache provider.
-func Kind(cache cache.Cache, object client.Object) SyncingSource {
-	return &internal.Kind{Cache: cache, Type: object}
+func Kind(cache cache.Cache, object client.Object, eventhandler handler.EventHandler) SyncingSource {
+	return &internal.Kind{Cache: cache, Type: object, EventHandler: eventhandler}
 }
 
 // NewChannelBroadcaster creates a new ChannelBroadcaster for the given channel.
@@ -80,7 +79,7 @@ func WithDestBufferSize(destBufferSize int) ChannelOption {
 }
 
 // Channel creates a ChannelSource with the given buffer size.
-func Channel(broadcaster *internal.ChannelBroadcaster, options ...ChannelOption) Source {
+func Channel(broadcaster *internal.ChannelBroadcaster, eventhandler handler.EventHandler, options ...ChannelOption) Source {
 	opts := internal.ChannelOptions{
 		// 1024 is the default number of event notifications that can be buffered.
 		DestBufferSize: 1024,
@@ -92,12 +91,12 @@ func Channel(broadcaster *internal.ChannelBroadcaster, options ...ChannelOption)
 		o(&opts)
 	}
 
-	return &internal.Channel{Options: opts, Broadcaster: broadcaster}
+	return &internal.Channel{Options: opts, Broadcaster: broadcaster, EventHandler: eventhandler}
 }
 
 // Informer creates an InformerSource with the given cache provider.
-func Informer(informer cache.Informer) Source {
-	return &internal.Informer{Informer: informer}
+func Informer(informer cache.Informer, eventhandler handler.EventHandler) Source {
+	return &internal.Informer{Informer: informer, EventHandler: eventhandler}
 }
 
 // Func creates a FuncSource with the given function.
diff --git a/pkg/source/source_integration_test.go b/pkg/source/source_integration_test.go
index cfb92cb88f..6d6515d8aa 100644
--- a/pkg/source/source_integration_test.go
+++ b/pkg/source/source_integration_test.go
@@ -37,7 +37,6 @@ import (
 )
 
 var _ = Describe("Source", func() {
-	var instance1, instance2 source.Source
 	var obj client.Object
 	var q workqueue.RateLimitingInterface
 	var c1, c2 chan interface{}
@@ -58,11 +57,6 @@ var _ = Describe("Source", func() {
 		c2 = make(chan interface{})
 	})
 
-	JustBeforeEach(func() {
-		instance1 = source.Kind(icache, obj)
-		instance2 = source.Kind(icache, obj)
-	})
-
 	AfterEach(func() {
 		err := clientset.CoreV1().Namespaces().Delete(ctx, ns, metav1.DeleteOptions{})
 		Expect(err).NotTo(HaveOccurred())
@@ -101,31 +95,33 @@ var _ = Describe("Source", func() {
 				}
 
 				// Create an event handler to verify the events
-				newHandler := func(c chan interface{}) handler.Funcs {
-					return handler.Funcs{
-						CreateFunc: func(ctx context.Context, evt event.CreateEvent, rli workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Expect(rli).To(Equal(q))
-							c <- evt
-						},
-						UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, rli workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Expect(rli).To(Equal(q))
-							c <- evt
-						},
-						DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, rli workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Expect(rli).To(Equal(q))
-							c <- evt
+				newSource := func(c chan interface{}) source.Source {
+					return source.Kind(
+						icache,
+						obj,
+						handler.Funcs{
+							CreateFunc: func(ctx context.Context, evt event.CreateEvent, rli workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Expect(rli).To(Equal(q))
+								c <- evt
+							},
+							UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, rli workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Expect(rli).To(Equal(q))
+								c <- evt
+							},
+							DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, rli workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Expect(rli).To(Equal(q))
+								c <- evt
+							},
 						},
-					}
+					)
 				}
-				handler1 := newHandler(c1)
-				handler2 := newHandler(c2)
 
 				// Create 2 instances
-				Expect(instance1.Start(ctx, handler1, q)).To(Succeed())
-				Expect(instance2.Start(ctx, handler2, q)).To(Succeed())
+				Expect(newSource(c1).Start(ctx, q)).To(Succeed())
+				Expect(newSource(c2).Start(ctx, q)).To(Succeed())
 
 				By("Creating a Deployment and expecting the CreateEvent.")
 				created, err = client.Create(ctx, deployment, metav1.CreateOptions{})
@@ -241,8 +237,7 @@ var _ = Describe("Source", func() {
 				c := make(chan struct{})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Informer(depInformer)
-				err := instance.Start(ctx, handler.Funcs{
+				instance := source.Informer(depInformer, handler.Funcs{
 					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
 						var err error
@@ -265,7 +260,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				_, err = clientset.AppsV1().ReplicaSets("default").Create(ctx, rs, metav1.CreateOptions{})
@@ -282,8 +278,7 @@ var _ = Describe("Source", func() {
 				rs2.SetLabels(map[string]string{"biz": "baz"})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Informer(depInformer)
-				err = instance.Start(ctx, handler.Funcs{
+				instance := source.Informer(depInformer, handler.Funcs{
 					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
 					},
 					UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) {
@@ -307,7 +302,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err = instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				_, err = clientset.AppsV1().ReplicaSets("default").Update(ctx, rs2, metav1.UpdateOptions{})
@@ -319,8 +315,7 @@ var _ = Describe("Source", func() {
 				c := make(chan struct{})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Informer(depInformer)
-				err := instance.Start(ctx, handler.Funcs{
+				instance := source.Informer(depInformer, handler.Funcs{
 					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
 					},
 					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
@@ -335,7 +330,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				err = clientset.AppsV1().ReplicaSets("default").Delete(ctx, rs.Name, metav1.DeleteOptions{})
diff --git a/pkg/source/source_test.go b/pkg/source/source_test.go
index 551d77b9c7..6e2d723cb8 100644
--- a/pkg/source/source_test.go
+++ b/pkg/source/source_test.go
@@ -65,8 +65,7 @@ var _ = Describe("Source", func() {
 				}
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Kind(ic, &corev1.Pod{})
-				err := instance.Start(ctx, handler.Funcs{
+				instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{
 					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
 						Expect(q2).To(Equal(q))
@@ -85,7 +84,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred())
 
@@ -102,8 +102,7 @@ var _ = Describe("Source", func() {
 
 				ic := &informertest.FakeInformers{}
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Kind(ic, &corev1.Pod{})
-				err := instance.Start(ctx, handler.Funcs{
+				instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{
 					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
 						Fail("Unexpected CreateEvent")
@@ -125,7 +124,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred())
 
@@ -147,8 +147,7 @@ var _ = Describe("Source", func() {
 				}
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Kind(ic, &corev1.Pod{})
-				err := instance.Start(ctx, handler.Funcs{
+				instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{
 					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
 						defer GinkgoRecover()
 						Fail("Unexpected DeleteEvent")
@@ -167,7 +166,8 @@ var _ = Describe("Source", func() {
 						defer GinkgoRecover()
 						Fail("Unexpected GenericEvent")
 					},
-				}, q)
+				})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 				Expect(instance.WaitForSync(context.Background())).NotTo(HaveOccurred())
 
@@ -179,24 +179,31 @@ var _ = Describe("Source", func() {
 			})
 		})
 
+		It("should return an error from Start eventhandler was not provided", func() {
+			instance := source.Kind(ic, &corev1.Pod{}, nil)
+			err := instance.Start(ctx, nil)
+			Expect(err).To(HaveOccurred())
+			Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil EventHandler"))
+		})
+
 		It("should return an error from Start cache was not provided", func() {
-			instance := source.Kind(nil, &corev1.Pod{})
-			err := instance.Start(ctx, nil, nil)
+			instance := source.Kind(nil, &corev1.Pod{}, &handler.EnqueueRequestForObject{})
+			err := instance.Start(ctx, nil)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil Cache"))
 		})
 
 		It("should return an error from Start if a type was not provided", func() {
-			instance := source.Kind(ic, nil)
-			err := instance.Start(ctx, nil, nil)
+			instance := source.Kind(ic, nil, &handler.EnqueueRequestForObject{})
+			err := instance.Start(ctx, nil)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil Type"))
 		})
 
 		It("should return an error if syncing fails", func() {
 			f := false
-			instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{})
-			Expect(instance.Start(context.Background(), nil, nil)).NotTo(HaveOccurred())
+			instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{})
+			Expect(instance.Start(context.Background(), nil)).NotTo(HaveOccurred())
 			err := instance.WaitForSync(context.Background())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(Equal("cache did not sync"))
@@ -211,8 +218,8 @@ var _ = Describe("Source", func() {
 				ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
 				defer cancel()
 
-				instance := source.Kind(ic, &corev1.Pod{})
-				err := instance.Start(ctx, handler.Funcs{}, q)
+				instance := source.Kind(ic, &corev1.Pod{}, handler.Funcs{})
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 				Eventually(instance.WaitForSync).WithArguments(context.Background()).Should(HaveOccurred())
 			})
@@ -220,8 +227,8 @@ var _ = Describe("Source", func() {
 
 		It("should return an error if syncing fails", func() {
 			f := false
-			instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{})
-			Expect(instance.Start(context.Background(), nil, nil)).NotTo(HaveOccurred())
+			instance := source.Kind(&informertest.FakeInformers{Synced: &f}, &corev1.Pod{}, &handler.EnqueueRequestForObject{})
+			Expect(instance.Start(context.Background(), nil)).NotTo(HaveOccurred())
 			err := instance.WaitForSync(context.Background())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(Equal("cache did not sync"))
@@ -232,24 +239,18 @@ var _ = Describe("Source", func() {
 	Describe("Func", func() {
 		It("should be called from Start", func() {
 			run := false
-			instance := source.Func(func(
-				context.Context,
-				handler.EventHandler,
-				workqueue.RateLimitingInterface, ...predicate.Predicate) error {
+			instance := source.Func(func(context.Context, workqueue.RateLimitingInterface) error {
 				run = true
 				return nil
 			})
-			Expect(instance.Start(ctx, nil, nil)).NotTo(HaveOccurred())
+			Expect(instance.Start(ctx, nil)).NotTo(HaveOccurred())
 			Expect(run).To(BeTrue())
 
 			expected := fmt.Errorf("expected error: Func")
-			instance = source.Func(func(
-				context.Context,
-				handler.EventHandler,
-				workqueue.RateLimitingInterface, ...predicate.Predicate) error {
+			instance = source.Func(func(context.Context, workqueue.RateLimitingInterface) error {
 				return expected
 			})
-			Expect(instance.Start(ctx, nil, nil)).To(Equal(expected))
+			Expect(instance.Start(ctx, nil)).To(Equal(expected))
 		})
 	})
 
@@ -289,29 +290,35 @@ var _ = Describe("Source", func() {
 				}
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Channel(source.NewChannelBroadcaster(ch))
-				err := instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected CreateEvent")
-					},
-					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected UpdateEvent")
-					},
-					DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected DeleteEvent")
-					},
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						// The empty event should have been filtered out by the predicates,
-						// and will not be passed to the handler.
-						Expect(q2).To(BeIdenticalTo(q))
-						Expect(evt.Object).To(Equal(p))
-						close(c)
-					},
-				}, q, prct)
+				instance := source.Channel(
+					source.NewChannelBroadcaster(ch),
+					handler.WithPredicates(
+						handler.Funcs{
+							CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected CreateEvent")
+							},
+							UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected UpdateEvent")
+							},
+							DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected DeleteEvent")
+							},
+							GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								// The empty event should have been filtered out by the predicates,
+								// and will not be passed to the handler.
+								Expect(q2).To(BeIdenticalTo(q))
+								Expect(evt.Object).To(Equal(p))
+								close(c)
+							},
+						},
+						prct,
+					),
+				)
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				ch <- invalidEvt
@@ -327,33 +334,37 @@ var _ = Describe("Source", func() {
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				// Add a handler to get distribution blocked
-				instance := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1))
-				err := instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected CreateEvent")
-					},
-					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected UpdateEvent")
-					},
-					DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected DeleteEvent")
-					},
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						// Block for the first time
-						if eventCount == 0 {
-							<-unblock
-						}
-						eventCount++
-
-						if eventCount == 3 {
-							close(processed)
-						}
+				instance := source.Channel(
+					source.NewChannelBroadcaster(ch),
+					handler.Funcs{
+						CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected CreateEvent")
+						},
+						UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected UpdateEvent")
+						},
+						DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected DeleteEvent")
+						},
+						GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							// Block for the first time
+							if eventCount == 0 {
+								<-unblock
+							}
+							eventCount++
+
+							if eventCount == 3 {
+								close(processed)
+							}
+						},
 					},
-				}, q)
+					source.WithDestBufferSize(1),
+				)
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				// Write 3 events into the source channel.
@@ -382,27 +393,31 @@ var _ = Describe("Source", func() {
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				// Add a handler to get distribution blocked
-				instance := source.Channel(source.NewChannelBroadcaster(ch), source.WithDestBufferSize(1))
+				instance := source.Channel(
+					source.NewChannelBroadcaster(ch),
+					handler.Funcs{
+						CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected CreateEvent")
+						},
+						UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected UpdateEvent")
+						},
+						DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected DeleteEvent")
+						},
+						GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
 
-				err := instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected CreateEvent")
-					},
-					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected UpdateEvent")
-					},
-					DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected DeleteEvent")
+							close(processed)
+						},
 					},
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
+					source.WithDestBufferSize(1),
+				)
 
-						close(processed)
-					},
-				}, q)
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				<-processed
@@ -419,31 +434,34 @@ var _ = Describe("Source", func() {
 				ch <- evt
 				close(ch)
 
-				By("feeding that channel to a channel source")
-				src := source.Channel(source.NewChannelBroadcaster(ch))
-
 				processed := make(chan struct{})
 				defer close(processed)
 
-				err := src.Start(ctx, handler.Funcs{
-					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected CreateEvent")
-					},
-					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected UpdateEvent")
-					},
-					DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected DeleteEvent")
-					},
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
+				By("feeding that channel to a channel source")
+				src := source.Channel(
+					source.NewChannelBroadcaster(ch),
+					handler.Funcs{
+						CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected CreateEvent")
+						},
+						UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected UpdateEvent")
+						},
+						DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected DeleteEvent")
+						},
+						GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
 
-						processed <- struct{}{}
+							processed <- struct{}{}
+						},
 					},
-				}, q)
+				)
+
+				err := src.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				By("expecting to only get one event")
@@ -452,8 +470,8 @@ var _ = Describe("Source", func() {
 			})
 			It("should get error if no source specified", func() {
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := source.Channel(nil)
-				err := instance.Start(ctx, handler.Funcs{}, q)
+				instance := source.Channel(nil, handler.Funcs{})
+				err := instance.Start(ctx, q)
 				Expect(err).To(Equal(fmt.Errorf("must create Channel with a non-nil Broadcaster")))
 			})
 		})
@@ -470,36 +488,39 @@ var _ = Describe("Source", func() {
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				channelSource := source.NewChannelBroadcaster(ch)
 
-				createHandler := func(c chan event.GenericEvent) handler.Funcs {
-					return handler.Funcs{
-						CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Fail("Unexpected CreateEvent")
-						},
-						UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Fail("Unexpected UpdateEvent")
-						},
-						DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Fail("Unexpected DeleteEvent")
-						},
-						GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-							defer GinkgoRecover()
-							Expect(q2).To(BeIdenticalTo(q))
-							Expect(evt.Object).To(Equal(p))
-							c <- evt
+				createInstance := func(c chan event.GenericEvent) source.Source {
+					return source.Channel(
+						channelSource,
+						handler.Funcs{
+							CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected CreateEvent")
+							},
+							UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected UpdateEvent")
+							},
+							DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Fail("Unexpected DeleteEvent")
+							},
+							GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
+								defer GinkgoRecover()
+								Expect(q2).To(BeIdenticalTo(q))
+								Expect(evt.Object).To(Equal(p))
+								c <- evt
+							},
 						},
-					}
+					)
 				}
 
 				c1 := make(chan event.GenericEvent)
 				c2 := make(chan event.GenericEvent)
 
-				err := source.Channel(channelSource).Start(ctx, createHandler(c1), q)
+				err := createInstance(c1).Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
-				err = source.Channel(channelSource).Start(ctx, createHandler(c2), q)
+				err = createInstance(c2).Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				ch <- evt