Clean shutdown of garbagecollector integration tests

pkg/controller/garbagecollector/garbagecollector.go

@@ -77,6 +77,9 @@ type GarbageCollector struct {
 	// GC caches the owners that do not exist according to the API server.
 	absentOwnerCache *ReferenceCache
 
+	kubeClient       clientset.Interface
+	eventBroadcaster record.EventBroadcaster
+
 	workerLock sync.RWMutex
 }
 
@@ -94,8 +97,6 @@ func NewGarbageCollector(
 ) (*GarbageCollector, error) {
 
 	eventBroadcaster := record.NewBroadcaster()
-	eventBroadcaster.StartStructuredLogging(0)
-	eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: kubeClient.CoreV1().Events("")})
 	eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "garbage-collector-controller"})
 
 	attemptToDelete := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "garbage_collector_attempt_to_delete")
@@ -107,6 +108,8 @@ func NewGarbageCollector(
 		attemptToDelete:  attemptToDelete,
 		attemptToOrphan:  attemptToOrphan,
 		absentOwnerCache: absentOwnerCache,
+		kubeClient:       kubeClient,
+		eventBroadcaster: eventBroadcaster,
 	}
 	gc.dependencyGraphBuilder = &GraphBuilder{
 		eventRecorder:    eventRecorder,
@@ -146,6 +149,11 @@ func (gc *GarbageCollector) Run(ctx context.Context, workers int) {
 	defer gc.attemptToOrphan.ShutDown()
 	defer gc.dependencyGraphBuilder.graphChanges.ShutDown()
 
+	// Start events processing pipeline.
+	gc.eventBroadcaster.StartStructuredLogging(0)
+	gc.eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: gc.kubeClient.CoreV1().Events("")})
+	defer gc.eventBroadcaster.Shutdown()
+
 	klog.Infof("Starting garbage collector controller")
 	defer klog.Infof("Shutting down garbage collector controller")