diff --git a/examples/builtins/main.go b/examples/builtins/main.go
index 8ea173b248..cc14c4e9f9 100644
--- a/examples/builtins/main.go
+++ b/examples/builtins/main.go
@@ -59,14 +59,14 @@ 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 1a115f2f7b..f99a22d0db 100644
--- a/pkg/builder/controller.go
+++ b/pkg/builder/controller.go
@@ -124,7 +124,6 @@ func (blder *Builder) Owns(object client.Object, opts ...OwnsOption) *Builder {
 // WatchesInput represents the information set by Watches method.
 type WatchesInput struct {
 	src              source.Source
-	eventHandler     handler.EventHandler
 	predicates       []predicate.Predicate
 	objectProjection objectProjection
 }
@@ -135,8 +134,13 @@ type WatchesInput struct {
 // 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...)
+	input := WatchesInput{}
+	for _, opt := range opts {
+		opt.ApplyToWatches(&input)
+	}
+	src := source.Kind(blder.mgr.GetCache(), object, eventHandler, input.predicates...)
+
+	return blder.WatchesRawSource(src, opts...)
 }
 
 // WatchesMetadata is the same as Watches, but forces the internal cache to only watch PartialObjectMetadata.
@@ -176,8 +180,8 @@ func (blder *Builder) WatchesMetadata(object client.Object, eventHandler handler
 //
 // 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}
+func (blder *Builder) WatchesRawSource(src source.Source, opts ...WatchesOption) *Builder {
+	input := WatchesInput{src: src}
 	for _, opt := range opts {
 		opt.ApplyToWatches(&input)
 	}
@@ -272,11 +276,11 @@ 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, hdler, allPredicates...)
+		if err := blder.ctrl.Watch(src); err != nil {
 			return err
 		}
 	}
@@ -290,7 +294,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())
@@ -302,7 +305,8 @@ 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, hdler, allPredicates...)
+		if err := blder.ctrl.Watch(src); err != nil {
 			return err
 		}
 	}
@@ -311,18 +315,19 @@ func (blder *Builder) doWatch() error {
 	if len(blder.watchesInput) == 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")
 	}
+	allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
 	for _, w := range blder.watchesInput {
 		// If the source of this watch is of type Kind, project it.
 		if srcKind, ok := w.src.(*internalsource.Kind); ok {
+			allPredicates := append(allPredicates, w.predicates...)
 			typeForSrc, err := blder.project(srcKind.Type, w.objectProjection)
 			if err != nil {
 				return err
 			}
 			srcKind.Type = typeForSrc
+			srcKind.Predicates = append(srcKind.Predicates, allPredicates...)
 		}
-		allPredicates := append([]predicate.Predicate(nil), blder.globalPredicates...)
-		allPredicates = append(allPredicates, w.predicates...)
-		if err := blder.ctrl.Watch(w.src, w.eventHandler, allPredicates...); err != nil {
+		if err := blder.ctrl.Watch(w.src); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
index 4d6f6b8955..5c9e48beae 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"
@@ -84,13 +82,8 @@ 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 watches the provided Source.
+	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..2e38f77a7d 100644
--- a/pkg/controller/controller_integration_test.go
+++ b/pkg/controller/controller_integration_test.go
@@ -65,12 +65,13 @@ var _ = Describe("controller", func() {
 
 			By("Watching Resources")
 			err = instance.Watch(
-				source.Kind(cm.GetCache(), &appsv1.ReplicaSet{}),
-				handler.EnqueueRequestForOwner(cm.GetScheme(), cm.GetRESTMapper(), &appsv1.Deployment{}),
+				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 f1197827c5..415f03a5b4 100644
--- a/pkg/controller/controller_test.go
+++ b/pkg/controller/controller_test.go
@@ -79,7 +79,7 @@ var _ = Describe("controller.Controller", func() {
 
 			ctx, cancel := context.WithCancel(context.Background())
 			watchChan := make(chan event.GenericEvent, 1)
-			watch := &source.Channel{Source: watchChan}
+			watch := &source.Channel{Source: watchChan, Handler: &handler.EnqueueRequestForObject{}}
 			watchChan <- event.GenericEvent{Object: &corev1.Pod{}}
 
 			reconcileStarted := make(chan struct{})
@@ -101,7 +101,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..3c4dbd9f50 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
@@ -55,8 +54,9 @@ func ExampleEnqueueRequestForObject() {
 func ExampleEnqueueRequestForOwner() {
 	// controller is a controller.controller
 	err := c.Watch(
-		source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{}),
-		handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.Deployment{}, handler.OnlyControllerOwner()),
+		source.Kind(mgr.GetCache(), &appsv1.ReplicaSet{},
+			handler.EnqueueRequestForOwner(mgr.GetScheme(), mgr.GetRESTMapper(), &appsv1.Deployment{}, handler.OnlyControllerOwner()),
+		),
 	)
 	if err != nil {
 		// handle it
@@ -68,19 +68,20 @@ func ExampleEnqueueRequestForOwner() {
 func ExampleEnqueueRequestsFromMapFunc() {
 	// controller is a controller.controller
 	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{
-					Name:      a.GetName() + "-1",
-					Namespace: a.GetNamespace(),
-				}},
-				{NamespacedName: types.NamespacedName{
-					Name:      a.GetName() + "-2",
-					Namespace: a.GetNamespace(),
-				}},
-			}
-		}),
+		source.Kind(mgr.GetCache(), &appsv1.Deployment{},
+			handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, a client.Object) []reconcile.Request {
+				return []reconcile.Request{
+					{NamespacedName: types.NamespacedName{
+						Name:      a.GetName() + "-1",
+						Namespace: a.GetNamespace(),
+					}},
+					{NamespacedName: types.NamespacedName{
+						Name:      a.GetName() + "-2",
+						Namespace: a.GetNamespace(),
+					}},
+				}
+			}),
+		),
 	)
 	if err != nil {
 		// handle it
@@ -91,33 +92,34 @@ func ExampleEnqueueRequestsFromMapFunc() {
 func ExampleFuncs() {
 	// controller is a controller.controller
 	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{
-					Name:      e.Object.GetName(),
-					Namespace: e.Object.GetNamespace(),
-				}})
+		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{
+						Name:      e.Object.GetName(),
+						Namespace: e.Object.GetNamespace(),
+					}})
+				},
+				UpdateFunc: func(ctx context.Context, e event.UpdateEvent, q workqueue.RateLimitingInterface) {
+					q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+						Name:      e.ObjectNew.GetName(),
+						Namespace: e.ObjectNew.GetNamespace(),
+					}})
+				},
+				DeleteFunc: func(ctx context.Context, e event.DeleteEvent, q workqueue.RateLimitingInterface) {
+					q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+						Name:      e.Object.GetName(),
+						Namespace: e.Object.GetNamespace(),
+					}})
+				},
+				GenericFunc: func(ctx context.Context, e event.GenericEvent, q workqueue.RateLimitingInterface) {
+					q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
+						Name:      e.Object.GetName(),
+						Namespace: e.Object.GetNamespace(),
+					}})
+				},
 			},
-			UpdateFunc: func(ctx context.Context, e event.UpdateEvent, q workqueue.RateLimitingInterface) {
-				q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-					Name:      e.ObjectNew.GetName(),
-					Namespace: e.ObjectNew.GetNamespace(),
-				}})
-			},
-			DeleteFunc: func(ctx context.Context, e event.DeleteEvent, q workqueue.RateLimitingInterface) {
-				q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-					Name:      e.Object.GetName(),
-					Namespace: e.Object.GetNamespace(),
-				}})
-			},
-			GenericFunc: func(ctx context.Context, e event.GenericEvent, q workqueue.RateLimitingInterface) {
-				q.Add(reconcile.Request{NamespacedName: types.NamespacedName{
-					Name:      e.Object.GetName(),
-					Namespace: e.Object.GetNamespace(),
-				}})
-			},
-		},
+		),
 	)
 	if err != nil {
 		// handle it
diff --git a/pkg/internal/controller/controller.go b/pkg/internal/controller/controller.go
index 40ba0685d0..f888494a84 100644
--- a/pkg/internal/controller/controller.go
+++ b/pkg/internal/controller/controller.go
@@ -29,10 +29,8 @@ import (
 	"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/ratelimiter"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -98,9 +96,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.
@@ -124,7 +120,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()
 
@@ -132,12 +128,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.
@@ -181,7 +177,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 96cd27e1e3..a10c9dcd3b 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/ratelimiter"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
@@ -128,7 +127,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{}, nil),
 			}}
 			ctrl.Name = "foo"
 			ctx, cancel := context.WithCancel(context.Background())
@@ -146,7 +145,7 @@ var _ = Describe("controller", func() {
 			c = &cacheWithIndefinitelyBlockingGetInformer{c}
 
 			ctrl.startWatches = []watchDescription{{
-				src: source.Kind(c, &appsv1.Deployment{}),
+				src: source.Kind(c, &appsv1.Deployment{}, nil),
 			}}
 			ctrl.Name = "testcontroller"
 
@@ -164,7 +163,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{}, nil),
 					cacheSyncDone: sourceSynced,
 				},
 			}}
@@ -192,7 +191,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{}, nil),
 					cacheSyncDone: sourceSynced,
 				},
 			}}
@@ -227,7 +226,15 @@ var _ = Describe("controller", func() {
 				Object: p,
 			}
 
-			ins := &source.Channel{Source: ch}
+			ins := &source.Channel{
+				Source: ch,
+				Handler: handler.Funcs{
+					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) {
+						defer GinkgoRecover()
+						close(processed)
+					},
+				},
+			}
 			ins.DestBufferSize = 1
 
 			// send the event to the channel
@@ -235,12 +242,6 @@ var _ = Describe("controller", func() {
 
 			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() {
@@ -265,20 +266,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())
@@ -289,13 +285,13 @@ 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,
+			src := source.Func(func(context.Context,
 				workqueue.RateLimitingInterface,
-				...predicate.Predicate) error {
+			) 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/internal/source/kind.go b/pkg/internal/source/kind.go
index b3a8227125..c56f423cf1 100644
--- a/pkg/internal/source/kind.go
+++ b/pkg/internal/source/kind.go
@@ -24,6 +24,10 @@ type Kind struct {
 	// Cache used to watch APIs
 	Cache cache.Cache
 
+	Handler handler.EventHandler
+
+	Predicates []predicate.Predicate
+
 	// 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
@@ -32,8 +36,7 @@ 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 object")
 	}
@@ -79,7 +82,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.Handler, ks.Predicates).HandlerFuncs())
 		if err != nil {
 			ks.started <- err
 			return
diff --git a/pkg/source/example_test.go b/pkg/source/example_test.go
index 77857729de..7940a225d1 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
 	}
@@ -43,8 +43,10 @@ func ExampleChannel() {
 	events := make(chan event.GenericEvent)
 
 	err := ctrl.Watch(
-		&source.Channel{Source: events},
-		&handler.EnqueueRequestForObject{},
+		&source.Channel{
+			Source:  events,
+			Handler: &handler.EnqueueRequestForObject{},
+		},
 	)
 	if err != nil {
 		// handle it
diff --git a/pkg/source/source.go b/pkg/source/source.go
index c0b9b1d9da..e36450d88d 100644
--- a/pkg/source/source.go
+++ b/pkg/source/source.go
@@ -18,6 +18,7 @@ package source
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 
@@ -47,7 +48,7 @@ const (
 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
@@ -58,8 +59,13 @@ type SyncingSource interface {
 }
 
 // Kind creates a KindSource with the given cache provider.
-func Kind(cache cache.Cache, object client.Object) SyncingSource {
-	return &internal.Kind{Type: object, Cache: cache}
+func Kind(cache cache.Cache, object client.Object, handler handler.EventHandler, predicates ...predicate.Predicate) SyncingSource {
+	return &internal.Kind{
+		Type:       object,
+		Cache:      cache,
+		Handler:    handler,
+		Predicates: predicates,
+	}
 }
 
 var _ Source = &Channel{}
@@ -74,6 +80,10 @@ type Channel struct {
 	// Source is the source channel to fetch GenericEvents
 	Source <-chan event.GenericEvent
 
+	Handler handler.EventHandler
+
+	Predicates []predicate.Predicate
+
 	// dest is the destination channels of the added event handlers
 	dest []chan event.GenericEvent
 
@@ -92,13 +102,15 @@ 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 {
+) error {
 	// Source should have been specified by the user.
 	if cs.Source == nil {
 		return fmt.Errorf("must specify Channel.Source")
 	}
+	if cs.Handler == nil {
+		return errors.New("must specify Channel.Handler")
+	}
 
 	// use default value if DestBufferSize not specified
 	if cs.DestBufferSize == 0 {
@@ -119,7 +131,7 @@ func (cs *Channel) Start(
 	go func() {
 		for evt := range dst {
 			shouldHandle := true
-			for _, p := range prct {
+			for _, p := range cs.Predicates {
 				if !p.Generic(evt) {
 					shouldHandle = false
 					break
@@ -130,7 +142,7 @@ func (cs *Channel) Start(
 				func() {
 					ctx, cancel := context.WithCancel(ctx)
 					defer cancel()
-					handler.Generic(ctx, evt, queue)
+					cs.Handler.Generic(ctx, evt, queue)
 				}()
 			}
 		}
@@ -184,21 +196,25 @@ func (cs *Channel) syncLoop(ctx context.Context) {
 // Informer is used to provide a source of events originating inside the cluster from Watches (e.g. Pod Create).
 type Informer struct {
 	// Informer is the controller-runtime Informer
-	Informer cache.Informer
+	Informer   cache.Informer
+	Handler    handler.EventHandler
+	Predicates []predicate.Predicate
 }
 
 var _ Source = &Informer{}
 
 // 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 {
 	// Informer should have been specified by the user.
 	if is.Informer == nil {
 		return fmt.Errorf("must specify Informer.Informer")
 	}
+	if is.Handler == nil {
+		return errors.New("must specify Informer.Handler")
+	}
 
-	_, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, handler, prct).HandlerFuncs())
+	_, err := is.Informer.AddEventHandler(internal.NewEventHandler(ctx, queue, is.Handler, is.Predicates).HandlerFuncs())
 	if err != nil {
 		return err
 	}
@@ -212,12 +228,11 @@ func (is *Informer) String() string {
 var _ Source = Func(nil)
 
 // 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/source_integration_test.go b/pkg/source/source_integration_test.go
index 594d3c9a9c..f6b2948874 100644
--- a/pkg/source/source_integration_test.go
+++ b/pkg/source/source_integration_test.go
@@ -58,11 +58,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())
@@ -124,8 +119,10 @@ var _ = Describe("Source", func() {
 				handler2 := newHandler(c2)
 
 				// Create 2 instances
-				Expect(instance1.Start(ctx, handler1, q)).To(Succeed())
-				Expect(instance2.Start(ctx, handler2, q)).To(Succeed())
+				instance1 = source.Kind(icache, obj, handler1)
+				instance2 = source.Kind(icache, obj, handler2)
+				Expect(instance1.Start(ctx, q)).To(Succeed())
+				Expect(instance2.Start(ctx, q)).To(Succeed())
 
 				By("Creating a Deployment and expecting the CreateEvent.")
 				created, err = client.Create(ctx, deployment, metav1.CreateOptions{})
@@ -241,31 +238,34 @@ var _ = Describe("Source", func() {
 				c := make(chan struct{})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := &source.Informer{Informer: depInformer}
-				err := instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						var err error
-						rs, err := clientset.AppsV1().ReplicaSets("default").Get(ctx, rs.Name, metav1.GetOptions{})
-						Expect(err).NotTo(HaveOccurred())
-
-						Expect(q2).To(BeIdenticalTo(q))
-						Expect(evt.Object).To(Equal(rs))
-						close(c)
-					},
-					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(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected GenericEvent")
+				instance := &source.Informer{
+					Informer: depInformer,
+					Handler: handler.Funcs{
+						CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							var err error
+							rs, err := clientset.AppsV1().ReplicaSets("default").Get(ctx, rs.Name, metav1.GetOptions{})
+							Expect(err).NotTo(HaveOccurred())
+
+							Expect(q2).To(BeIdenticalTo(q))
+							Expect(evt.Object).To(Equal(rs))
+							close(c)
+						},
+						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(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
+							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,32 +282,35 @@ var _ = Describe("Source", func() {
 				rs2.SetLabels(map[string]string{"biz": "baz"})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := &source.Informer{Informer: depInformer}
-				err = instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
-					},
-					UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						var err error
-						rs2, err := clientset.AppsV1().ReplicaSets("default").Get(ctx, rs.Name, metav1.GetOptions{})
-						Expect(err).NotTo(HaveOccurred())
+				instance := &source.Informer{
+					Informer: depInformer,
+					Handler: handler.Funcs{
+						CreateFunc: func(ctx context.Context, evt event.CreateEvent, q2 workqueue.RateLimitingInterface) {
+						},
+						UpdateFunc: func(ctx context.Context, evt event.UpdateEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							var err error
+							rs2, err := clientset.AppsV1().ReplicaSets("default").Get(ctx, rs.Name, metav1.GetOptions{})
+							Expect(err).NotTo(HaveOccurred())
 
-						Expect(q2).To(Equal(q))
-						Expect(evt.ObjectOld).To(Equal(rs))
+							Expect(q2).To(Equal(q))
+							Expect(evt.ObjectOld).To(Equal(rs))
 
-						Expect(evt.ObjectNew).To(Equal(rs2))
+							Expect(evt.ObjectNew).To(Equal(rs2))
 
-						close(c)
-					},
-					DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected DeleteEvent")
-					},
-					GenericFunc: func(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected GenericEvent")
+							close(c)
+						},
+						DeleteFunc: func(context.Context, event.DeleteEvent, workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Fail("Unexpected DeleteEvent")
+						},
+						GenericFunc: func(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
+							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,23 +322,26 @@ var _ = Describe("Source", func() {
 				c := make(chan struct{})
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := &source.Informer{Informer: depInformer}
-				err := instance.Start(ctx, handler.Funcs{
-					CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
-					},
-					UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
-					},
-					DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Expect(q2).To(Equal(q))
-						Expect(evt.Object.GetName()).To(Equal(rs.Name))
-						close(c)
-					},
-					GenericFunc: func(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
-						Fail("Unexpected GenericEvent")
+				instance := &source.Informer{
+					Informer: depInformer,
+					Handler: handler.Funcs{
+						CreateFunc: func(context.Context, event.CreateEvent, workqueue.RateLimitingInterface) {
+						},
+						UpdateFunc: func(context.Context, event.UpdateEvent, workqueue.RateLimitingInterface) {
+						},
+						DeleteFunc: func(ctx context.Context, evt event.DeleteEvent, q2 workqueue.RateLimitingInterface) {
+							defer GinkgoRecover()
+							Expect(q2).To(Equal(q))
+							Expect(evt.Object.GetName()).To(Equal(rs.Name))
+							close(c)
+						},
+						GenericFunc: func(context.Context, event.GenericEvent, workqueue.RateLimitingInterface) {
+							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 16c365e8a2..84b0624afa 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())
 
@@ -180,23 +180,23 @@ var _ = Describe("Source", func() {
 		})
 
 		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{}, nil)
+			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, nil)
+			err := instance.Start(ctx, nil)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("must create Kind with a non-nil object"))
 		})
 
 		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{}, nil)
+			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 +211,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 +220,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{}, nil)
+			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"))
@@ -234,22 +234,20 @@ var _ = Describe("Source", func() {
 			run := false
 			instance := source.Func(func(
 				context.Context,
-				handler.EventHandler,
-				workqueue.RateLimitingInterface, ...predicate.Predicate) error {
+				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 {
+				workqueue.RateLimitingInterface) error {
 				return expected
 			})
-			Expect(instance.Start(ctx, nil, nil)).To(Equal(expected))
+			Expect(instance.Start(ctx, nil)).To(Equal(expected))
 		})
 	})
 
@@ -289,29 +287,33 @@ var _ = Describe("Source", func() {
 				}
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := &source.Channel{Source: 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)
+				instance := &source.Channel{
+					Source: ch,
+					Handler: 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)
+					Predicates: []predicate.Predicate{prct},
+				}
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				ch <- invalidEvt
@@ -327,34 +329,37 @@ var _ = Describe("Source", func() {
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				// Add a handler to get distribution blocked
-				instance := &source.Channel{Source: ch}
-				instance.DestBufferSize = 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: ch,
+					Handler: 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)
+				}
+				instance.DestBufferSize = 1
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				// Write 3 events into the source channel.
@@ -383,28 +388,31 @@ var _ = Describe("Source", func() {
 
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				// Add a handler to get distribution blocked
-				instance := &source.Channel{Source: ch}
-				instance.DestBufferSize = 1
+				instance := &source.Channel{
+					Source: ch,
+					Handler: 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()
+				}
+				instance.DestBufferSize = 1
 
-						close(processed)
-					},
-				}, q)
+				err := instance.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				<-processed
@@ -422,30 +430,32 @@ var _ = Describe("Source", func() {
 				close(ch)
 
 				By("feeding that channel to a channel source")
-				src := &source.Channel{Source: ch}
-
 				processed := make(chan struct{})
 				defer close(processed)
+				src := &source.Channel{
+					Source: ch,
+					Handler: 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 := 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")
+							processed <- struct{}{}
+						},
 					},
-					GenericFunc: func(ctx context.Context, evt event.GenericEvent, q2 workqueue.RateLimitingInterface) {
-						defer GinkgoRecover()
+				}
 
-						processed <- struct{}{}
-					},
-				}, q)
+				err := src.Start(ctx, q)
 				Expect(err).NotTo(HaveOccurred())
 
 				By("expecting to only get one event")
@@ -455,79 +465,9 @@ var _ = Describe("Source", func() {
 			It("should get error if no source specified", func() {
 				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
 				instance := &source.Channel{ /*no source specified*/ }
-				err := instance.Start(ctx, handler.Funcs{}, q)
+				err := instance.Start(ctx, q)
 				Expect(err).To(Equal(fmt.Errorf("must specify Channel.Source")))
 			})
 		})
-		Context("for multi sources (handlers)", func() {
-			It("should provide GenericEvents for all handlers", func() {
-				ch := make(chan event.GenericEvent)
-				p := &corev1.Pod{
-					ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "bar"},
-				}
-				evt := event.GenericEvent{
-					Object: p,
-				}
-
-				var resEvent1, resEvent2 event.GenericEvent
-				c1 := make(chan struct{})
-				c2 := make(chan struct{})
-
-				q := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "test")
-				instance := &source.Channel{Source: 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()
-						Expect(q2).To(BeIdenticalTo(q))
-						Expect(evt.Object).To(Equal(p))
-						resEvent1 = evt
-						close(c1)
-					},
-				}, q)
-				Expect(err).NotTo(HaveOccurred())
-
-				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()
-						Expect(q2).To(BeIdenticalTo(q))
-						Expect(evt.Object).To(Equal(p))
-						resEvent2 = evt
-						close(c2)
-					},
-				}, q)
-				Expect(err).NotTo(HaveOccurred())
-
-				ch <- evt
-				<-c1
-				<-c2
-
-				// Validate the two handlers received same event
-				Expect(resEvent1).To(Equal(resEvent2))
-			})
-		})
 	})
 })