Repository: kubernetes-sigs/controller-runtime

Issue:
Switch to use `events.EventRecorder` instead of `record.EventRecorder`
Controller runtime uses older interface, would be good to utilize new one

Hints from issue comments:
@k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-1605722589):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>@alculquicondor: /reopen

@alvaroaleman are there any considerations with regards to backwards-compatibility? Should we support both?
@k8s-ci-robot: @alculquicondor: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-1834281397):

>/reopen
>
>@alvaroaleman are there any considerations with regards to backwards-compatibility? Should we support both?
>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>@alvaroaleman: No, lets not support both. Is the old interface deprecated? Just for my understand as I have zero background here, what are reasons to use the `events.EventRecorder` over the `record.EventRecorder`?@alculquicondor: The one in `events` allows to provide extra information.

For example, you can have a secondary object and an action https://pkg.go.dev/k8s.io/client-go@v0.28.4/tools/events#EventRecorder

As a result, by migrating, we are forcing any users of `GetEventRecorderFor` to change their code.@alvaroaleman: Ah. So generally we allow breaking changes in minor version, so if we trade it for some benefit, that seems fine to me.@alculquicondor: @stuton, if we can't do https://github.com/kubernetes-sigs/kueue/issues/1387, maybe you can do the change in controller-runtime.@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-1902056396):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>@alculquicondor: /reopen@k8s-ci-robot: @alculquicondor: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-1904035195):

>/reopen


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-1956803871):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>@mimowo: /reopen 
To revisit the discussion and see if we could consider supporting both interfaces? This way applications using the controller-runtime would not need to change code on upgrade, while some applications could start using the new API. For example. the new interface could be exposed by a new function, such as `NewEventRecorderFor`.

WDYT @alvaroaleman ?@k8s-ci-robot: @mimowo: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2258326988):

>/reopen 
>To revisit the discussion and see if we could consider supporting both interfaces? This way applications using the controller-runtime would not need to change code on upgrade, while some applications could start using the new API. For example. the new interface could be exposed by a new function, such as `NewEventRecorderFor`.
>
>WDYT @alvaroaleman ?


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@alvaroaleman: I am in favor of breaking it rather than supporting both interfaces@mimowo: I see, does it get significantly more work to support both?

If both are supported, then the upgrade of `controller-runtime` could be friction-less for apps, and would not require a bump of the major version of controller-runtime (I imagine). Then, we could deprecate the old recorder only when the old type is deprecated upstream k8s.@vladikkuzn: > I see, does it get significantly more work to support both?

There'll be a need to have two methods in [interface of cluster](https://github.com/kubernetes-sigs/controller-runtime/blob/main/pkg/cluster/cluster.go#L63)
eg.
```
// Old recorder

// GetEventRecorderFor returns a deprecated version of record.EventRecorder for the provided name
GetEventRecorderFor(name string) record.EventRecorder

// New recorder

// GetNewEventRecorderFor returns a new events.EventRecorder for the provided name
GetNewEventRecorderFor(name string) events.EventRecorder
```@mimowo: This sounds innocent to me, is the underneath much implementation that would need to be repeated?@sbueringer: >  and would not require a bump of the major version of controller-runtime

controller-runtime is still using 0., so this will go into a minor version in any case.

> This sounds innocent to me, is the underneath much implementation that would need to be repeated?

Good question, someone will have to do some research :)

Regarding names. We could just drop the "For" which seems pretty redundant anyway

> GetEventRecorder(name string) events.EventRecorder@vladikkuzn: /assign@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2333601762):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@mimowo: /reopen
We have an assignee per https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2272974128@k8s-ci-robot: @mimowo: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2333628467):

>/reopen
>We have an assignee per https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2272974128


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@sbueringer: /remove lifecycle/rotten@troy0820: If we are going to move to the new interface (this doesn't look like it's planned for the next release) we can deprecate the old interface `recorder.EventRecorder` and by release-0.20 remove it in favor of the new `events.EventRecorder`.  

This would give time to people needing to migrate to the new interface. The breaking change is inevitable, but it's not really a drop in replacement with the function signature being different.  @k8s-triage-robot: The Kubernetes project currently lacks enough contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle stale`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle stale

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues.

This bot triages un-triaged issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Close this issue with `/close`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/lifecycle rotten

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-triage-robot: The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.

This bot triages issues according to the following rules:
- After 90d of inactivity, `lifecycle/stale` is applied
- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed

You can:
- Reopen this issue with `/reopen`
- Mark this issue as fresh with `/remove-lifecycle rotten`
- Offer to help out with [Issue Triage][1]

Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).

/close not-planned

[1]: https://www.kubernetes.dev/docs/guide/issue-triage/@k8s-ci-robot: @k8s-triage-robot: Closing this issue, marking it as "Not Planned".

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2746279077):

>The Kubernetes project currently lacks enough active contributors to adequately respond to all issues and PRs.
>
>This bot triages issues according to the following rules:
>- After 90d of inactivity, `lifecycle/stale` is applied
>- After 30d of inactivity since `lifecycle/stale` was applied, `lifecycle/rotten` is applied
>- After 30d of inactivity since `lifecycle/rotten` was applied, the issue is closed
>
>You can:
>- Reopen this issue with `/reopen`
>- Mark this issue as fresh with `/remove-lifecycle rotten`
>- Offer to help out with [Issue Triage][1]
>
>Please send feedback to sig-contributor-experience at [kubernetes/community](https://github.com/kubernetes/community).
>
>/close not-planned
>
>[1]: https://www.kubernetes.dev/docs/guide/issue-triage/


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@sbueringer: /remove-lifecycle rotten
/reopen@k8s-ci-robot: @sbueringer: Reopened this issue.

<details>

In response to [this](https://github.com/kubernetes-sigs/controller-runtime/issues/2141#issuecomment-2788397613):

>/remove-lifecycle rotten
>/reopen


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>@clebs: Hi folks,

I have been investigating how to make this migration possible. From the `EventRecorder` perspective it is faily simple.
The tricky part is migrating to the new `EventBroadcaster`, which is quite different from the old one.

Since this change affects the entire codebase, I would also argue against keeping both APIs at the same time because it would mean we need duplication across the board and on tests.

I have a draft PR with most of the changes done to see if this is the right direction or we should go for a completely different approach: #3262

I would appreciate feedback especially in how broadcasters are injected for testing and the general plumbing throughout the codebase.
I tried to make the injection more straightforward and with less indirection now that the constructor has changed for broadcasters.

There are also some open challenges where I left comments (e.g. `resourcelock.ResourceLockConfig` still uses the old API).@clebs: The fact that client-go's [ResourceLockConfig](https://github.com/kubernetes/client-go/blob/06fcc8a0753eec9e896c5f96985fc6595ef74391/tools/leaderelection/resourcelock/interface.go#L68) (used in leader election) is still using the old events API forces us to support both for now.
Therefore I have added a way to somewhat still support the old API with minimal duplication.@clebs: /assign

Solution patch:
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
index 6e3b7734..d9b9f394 100644
--- a/.github/workflows/release.yaml
+++ b/.github/workflows/release.yaml
@@ -29,7 +29,7 @@ jobs:
       run: |
         make release
     - name: Release
-      uses: softprops/action-gh-release@62c96d0c4e8a889135c1f3a25910db8dbe0e85f7 # tag=v2.3.4
+      uses: softprops/action-gh-release@6da8fa9354ddfdc4aeace5fc48d7f679b5214090 # tag=v2.4.1
       with:
         draft: false
         files: tools/setup-envtest/out/*
diff --git a/pkg/cluster/cluster.go b/pkg/cluster/cluster.go
index 0603f4cd..ee14638c 100644
--- a/pkg/cluster/cluster.go
+++ b/pkg/cluster/cluster.go
@@ -19,13 +19,16 @@ package cluster
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/http"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
@@ -33,10 +36,13 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 	logf "sigs.k8s.io/controller-runtime/pkg/internal/log"
 	intrec "sigs.k8s.io/controller-runtime/pkg/internal/recorder"
+	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
 // Cluster provides various methods to interact with a cluster.
 type Cluster interface {
+	recorder.Provider
+
 	// GetHTTPClient returns an HTTP client that can be used to talk to the apiserver
 	GetHTTPClient() *http.Client
 
@@ -58,9 +64,6 @@ type Cluster interface {
 	// GetFieldIndexer returns a client.FieldIndexer configured with the client
 	GetFieldIndexer() client.FieldIndexer
 
-	// GetEventRecorderFor returns a new EventRecorder for the provided name
-	GetEventRecorderFor(name string) record.EventRecorder
-
 	// GetRESTMapper returns a RESTMapper
 	GetRESTMapper() meta.RESTMapper
 
@@ -160,8 +163,7 @@ func New(config *rest.Config, opts ...Option) (Cluster, error) {
 	}
 	options, err := setOptionsDefaults(options, config)
 	if err != nil {
-		options.Logger.Error(err, "Failed to set defaults")
-		return nil, err
+		return nil, fmt.Errorf("failed setting cluster default options: %w", err)
 	}
 
 	// Create the mapper provider
@@ -281,16 +283,24 @@ func setOptionsDefaults(options Options, config *rest.Config) (Options, error) {
 		options.newRecorderProvider = intrec.NewProvider
 	}
 
+	// This is duplicated with pkg/manager, we need it here to provide
+	// the user with an EventBroadcaster and there for the Leader election
+	evtCl, err := eventsv1client.NewForConfigAndClient(config, options.HTTPClient)
+	if err != nil {
+		return options, err
+	}
+
 	// This is duplicated with pkg/manager, we need it here to provide
 	// the user with an EventBroadcaster and there for the Leader election
 	if options.EventBroadcaster == nil {
 		// defer initialization to avoid leaking by default
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return record.NewBroadcaster(), true
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
 		}
 	} else {
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return options.EventBroadcaster, false
+		// keep supporting the options.EventBroadcaster in the old API, but do not introduce it for the new one.
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return options.EventBroadcaster, events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), false
 		}
 	}
 
diff --git a/pkg/cluster/internal.go b/pkg/cluster/internal.go
index 27427642..755f83b5 100644
--- a/pkg/cluster/internal.go
+++ b/pkg/cluster/internal.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
@@ -87,6 +88,10 @@ func (c *cluster) GetEventRecorderFor(name string) record.EventRecorder {
 	return c.recorderProvider.GetEventRecorderFor(name)
 }
 
+func (c *cluster) GetEventRecorder(name string) events.EventRecorder {
+	return c.recorderProvider.GetEventRecorder(name)
+}
+
 func (c *cluster) GetRESTMapper() meta.RESTMapper {
 	return c.mapper
 }
diff --git a/pkg/internal/recorder/recorder.go b/pkg/internal/recorder/recorder.go
index 21f0146b..bbc16048 100644
--- a/pkg/internal/recorder/recorder.go
+++ b/pkg/internal/recorder/recorder.go
@@ -24,16 +24,19 @@ import (
 
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 )
 
 // EventBroadcasterProducer makes an event broadcaster, returning
 // whether or not the broadcaster should be stopped with the Provider,
 // or not (e.g. if it's shared, it shouldn't be stopped with the Provider).
-type EventBroadcasterProducer func() (caster record.EventBroadcaster, stopWithProvider bool)
+// This producer currently produces both an old API and a new API broadcaster.
+type EventBroadcasterProducer func() (deprecatedCaster record.EventBroadcaster, caster events.EventBroadcaster, stopWithProvider bool)
 
 // Provider is a recorder.Provider that records events to the k8s API server
 // and to a logr Logger.
@@ -48,9 +51,13 @@ type Provider struct {
 	evtClient       corev1client.EventInterface
 	makeBroadcaster EventBroadcasterProducer
 
-	broadcasterOnce sync.Once
-	broadcaster     record.EventBroadcaster
-	stopBroadcaster bool
+	broadcasterOnce         sync.Once
+	broadcaster             events.EventBroadcaster
+	cancelSinkRecordingFunc context.CancelFunc
+	stopWatcherFunc         func()
+	// Deprecated: will be removed in a future release. Use the broadcaster above instead.
+	deprecatedBroadcaster record.EventBroadcaster
+	stopBroadcaster       bool
 }
 
 // NB(directxman12): this manually implements Stop instead of Being a runnable because we need to
@@ -71,10 +78,13 @@ func (p *Provider) Stop(shutdownCtx context.Context) {
 		// almost certainly already been started (e.g. by leader election).  We
 		// need to invoke this to ensure that we don't inadvertently race with
 		// an invocation of getBroadcaster.
-		broadcaster := p.getBroadcaster()
+		deprecatedBroadcaster, broadcaster := p.getBroadcaster()
 		if p.stopBroadcaster {
 			p.lock.Lock()
 			broadcaster.Shutdown()
+			p.cancelSinkRecordingFunc()
+			p.stopWatcherFunc()
+			deprecatedBroadcaster.Shutdown()
 			p.stopped = true
 			p.lock.Unlock()
 		}
@@ -89,7 +99,7 @@ func (p *Provider) Stop(shutdownCtx context.Context) {
 
 // getBroadcaster ensures that a broadcaster is started for this
 // provider, and returns it.  It's threadsafe.
-func (p *Provider) getBroadcaster() record.EventBroadcaster {
+func (p *Provider) getBroadcaster() (record.EventBroadcaster, events.EventBroadcaster) {
 	// NB(directxman12): this can technically still leak if something calls
 	// "getBroadcaster" (i.e. Emits an Event) but never calls Start, but if we
 	// create the broadcaster in start, we could race with other things that
@@ -97,17 +107,37 @@ func (p *Provider) getBroadcaster() record.EventBroadcaster {
 	// silently swallowing events and more locking, but that seems suboptimal.
 
 	p.broadcasterOnce.Do(func() {
-		broadcaster, stop := p.makeBroadcaster()
-		broadcaster.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: p.evtClient})
-		broadcaster.StartEventWatcher(
+		p.deprecatedBroadcaster, p.broadcaster, p.stopBroadcaster = p.makeBroadcaster()
+
+		// init deprecated broadcaster
+		p.deprecatedBroadcaster.StartRecordingToSink(&corev1client.EventSinkImpl{Interface: p.evtClient})
+		p.deprecatedBroadcaster.StartEventWatcher(
 			func(e *corev1.Event) {
 				p.logger.V(1).Info(e.Message, "type", e.Type, "object", e.InvolvedObject, "reason", e.Reason)
 			})
-		p.broadcaster = broadcaster
-		p.stopBroadcaster = stop
+
+		// init new broadcaster
+		ctx, cancel := context.WithCancel(context.Background())
+		p.cancelSinkRecordingFunc = cancel
+		if err := p.broadcaster.StartRecordingToSinkWithContext(ctx); err != nil {
+			p.logger.Error(err, "error starting recording for broadcaster")
+			return
+		}
+
+		stopWatcher, err := p.broadcaster.StartEventWatcher(func(event runtime.Object) {
+			e, isEvt := event.(*eventsv1.Event)
+			if isEvt {
+				p.logger.V(1).Info(e.Note, "type", e.Type, "object", e.Related, "action", e.Action, "reason", e.Reason)
+			}
+		})
+		if err != nil {
+			p.logger.Error(err, "error starting event watcher for broadcaster")
+		}
+
+		p.stopWatcherFunc = stopWatcher
 	})
 
-	return p.broadcaster
+	return p.deprecatedBroadcaster, p.broadcaster
 }
 
 // NewProvider create a new Provider instance.
@@ -128,6 +158,15 @@ func NewProvider(config *rest.Config, httpClient *http.Client, scheme *runtime.S
 // GetEventRecorderFor returns an event recorder that broadcasts to this provider's
 // broadcaster.  All events will be associated with a component of the given name.
 func (p *Provider) GetEventRecorderFor(name string) record.EventRecorder {
+	return &deprecatedRecorder{
+		prov: p,
+		name: name,
+	}
+}
+
+// GetEventRecorder returns an event recorder that broadcasts to this provider's
+// broadcaster.  All events will be associated with a component of the given name.
+func (p *Provider) GetEventRecorder(name string) events.EventRecorder {
 	return &lazyRecorder{
 		prov: p,
 		name: name,
@@ -141,18 +180,47 @@ type lazyRecorder struct {
 	name string
 
 	recOnce sync.Once
-	rec     record.EventRecorder
+	rec     events.EventRecorder
 }
 
 // ensureRecording ensures that a concrete recorder is populated for this recorder.
 func (l *lazyRecorder) ensureRecording() {
 	l.recOnce.Do(func() {
-		broadcaster := l.prov.getBroadcaster()
-		l.rec = broadcaster.NewRecorder(l.prov.scheme, corev1.EventSource{Component: l.name})
+		_, broadcaster := l.prov.getBroadcaster()
+		l.rec = broadcaster.NewRecorder(l.prov.scheme, l.name)
 	})
 }
 
-func (l *lazyRecorder) Event(object runtime.Object, eventtype, reason, message string) {
+func (l *lazyRecorder) Eventf(regarding runtime.Object, related runtime.Object, eventtype, reason, action, note string, args ...any) {
+	l.ensureRecording()
+
+	l.prov.lock.RLock()
+	if !l.prov.stopped {
+		l.rec.Eventf(regarding, related, eventtype, reason, action, note, args...)
+	}
+	l.prov.lock.RUnlock()
+}
+
+// deprecatedRecorder implements the old events API during the tranisiton and will be removed in a future release.
+//
+// Deprecated: will be removed in a future release.
+type deprecatedRecorder struct {
+	prov *Provider
+	name string
+
+	recOnce sync.Once
+	rec     record.EventRecorder
+}
+
+// ensureRecording ensures that a concrete recorder is populated for this recorder.
+func (l *deprecatedRecorder) ensureRecording() {
+	l.recOnce.Do(func() {
+		deprecatedBroadcaster, _ := l.prov.getBroadcaster()
+		l.rec = deprecatedBroadcaster.NewRecorder(l.prov.scheme, corev1.EventSource{Component: l.name})
+	})
+}
+
+func (l *deprecatedRecorder) Event(object runtime.Object, eventtype, reason, message string) {
 	l.ensureRecording()
 
 	l.prov.lock.RLock()
@@ -161,7 +229,8 @@ func (l *lazyRecorder) Event(object runtime.Object, eventtype, reason, message s
 	}
 	l.prov.lock.RUnlock()
 }
-func (l *lazyRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...interface{}) {
+
+func (l *deprecatedRecorder) Eventf(object runtime.Object, eventtype, reason, messageFmt string, args ...any) {
 	l.ensureRecording()
 
 	l.prov.lock.RLock()
@@ -170,7 +239,8 @@ func (l *lazyRecorder) Eventf(object runtime.Object, eventtype, reason, messageF
 	}
 	l.prov.lock.RUnlock()
 }
-func (l *lazyRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...interface{}) {
+
+func (l *deprecatedRecorder) AnnotatedEventf(object runtime.Object, annotations map[string]string, eventtype, reason, messageFmt string, args ...any) {
 	l.ensureRecording()
 
 	l.prov.lock.RLock()
diff --git a/pkg/leaderelection/leader_election.go b/pkg/leaderelection/leader_election.go
index 6c013e79..63d875b4 100644
--- a/pkg/leaderelection/leader_election.go
+++ b/pkg/leaderelection/leader_election.go
@@ -127,8 +127,10 @@ func NewResourceLock(config *rest.Config, recorderProvider recorder.Provider, op
 		corev1Client,
 		coordinationClient,
 		resourcelock.ResourceLockConfig{
-			Identity:      id,
-			EventRecorder: recorderProvider.GetEventRecorderFor(id),
+			Identity: id,
+			// TODO(clebs): Replace with the new events API after leader election is updated upstream.
+			// REF: https://github.com/kubernetes/kubernetes/issues/82846
+			EventRecorder: recorderProvider.GetEventRecorderFor(id), //nolint:staticcheck
 		},
 		options.LeaderLabels,
 	)
diff --git a/pkg/manager/internal.go b/pkg/manager/internal.go
index a2c3e532..4362022b 100644
--- a/pkg/manager/internal.go
+++ b/pkg/manager/internal.go
@@ -32,6 +32,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
@@ -256,7 +257,11 @@ func (cm *controllerManager) GetCache() cache.Cache {
 }
 
 func (cm *controllerManager) GetEventRecorderFor(name string) record.EventRecorder {
-	return cm.cluster.GetEventRecorderFor(name)
+	return cm.cluster.GetEventRecorderFor(name) //nolint:staticcheck
+}
+
+func (cm *controllerManager) GetEventRecorder(name string) events.EventRecorder {
+	return cm.cluster.GetEventRecorder(name)
 }
 
 func (cm *controllerManager) GetRESTMapper() meta.RESTMapper {
diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go
index e0e94245..74983ddc 100644
--- a/pkg/manager/manager.go
+++ b/pkg/manager/manager.go
@@ -29,7 +29,9 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
@@ -337,7 +339,10 @@ func New(config *rest.Config, options Options) (Manager, error) {
 		return nil, errors.New("must specify Config")
 	}
 	// Set default values for options fields
-	options = setOptionsDefaults(options)
+	options, err := setOptionsDefaults(config, options)
+	if err != nil {
+		return nil, fmt.Errorf("failed setting manager default options: %w", err)
+	}
 
 	cluster, err := cluster.New(config, func(clusterOptions *cluster.Options) {
 		clusterOptions.Scheme = options.Scheme
@@ -493,7 +498,7 @@ func defaultBaseContext() context.Context {
 }
 
 // setOptionsDefaults set default values for Options fields.
-func setOptionsDefaults(options Options) Options {
+func setOptionsDefaults(config *rest.Config, options Options) (Options, error) {
 	// Allow newResourceLock to be mocked
 	if options.newResourceLock == nil {
 		options.newResourceLock = leaderelection.NewResourceLock
@@ -507,14 +512,25 @@ func setOptionsDefaults(options Options) Options {
 	// This is duplicated with pkg/cluster, we need it here
 	// for the leader election and there to provide the user with
 	// an EventBroadcaster
+	httpClient, err := rest.HTTPClientFor(config)
+	if err != nil {
+		return options, err
+	}
+
+	evtCl, err := eventsv1client.NewForConfigAndClient(config, httpClient)
+	if err != nil {
+		return options, err
+	}
+
 	if options.EventBroadcaster == nil {
 		// defer initialization to avoid leaking by default
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return record.NewBroadcaster(), true
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
 		}
 	} else {
-		options.makeBroadcaster = func() (record.EventBroadcaster, bool) {
-			return options.EventBroadcaster, false
+		// keep supporting the options.EventBroadcaster in the old API, but do not introduce it for the new one.
+		options.makeBroadcaster = func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+			return options.EventBroadcaster, events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), false
 		}
 	}
 
@@ -571,5 +587,5 @@ func setOptionsDefaults(options Options) Options {
 		options.WebhookServer = webhook.NewServer(webhook.Options{})
 	}
 
-	return options
+	return options, nil
 }
diff --git a/pkg/recorder/recorder.go b/pkg/recorder/recorder.go
index f093f0a7..b34fecb5 100644
--- a/pkg/recorder/recorder.go
+++ b/pkg/recorder/recorder.go
@@ -21,11 +21,16 @@ limitations under the License.
 package recorder
 
 import (
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 )
 
 // Provider knows how to generate new event recorders with given name.
 type Provider interface {
-	// NewRecorder returns an EventRecorder with given name.
+	// GetEventRecorderFor returns an EventRecorder for the old events API.
+	//
+	// Deprecated: this uses the old events API and will be removed in a future release. Please use GetEventRecorder instead.
 	GetEventRecorderFor(name string) record.EventRecorder
+	// GetEventRecorder returns a EventRecorder with given name.
+	GetEventRecorder(name string) events.EventRecorder
 }



Test patch:
diff --git a/pkg/client/namespaced_client_test.go b/pkg/client/namespaced_client_test.go
index 7060f323..6e963547 100644
--- a/pkg/client/namespaced_client_test.go
+++ b/pkg/client/namespaced_client_test.go
@@ -44,6 +44,7 @@ import (
 
 var _ = Describe("NamespacedClient", func() {
 	var dep *appsv1.Deployment
+	var nameSpace *corev1.Namespace
 	var acDep *appsv1applyconfigurations.DeploymentApplyConfiguration
 	var ns = "default"
 	var count uint64 = 0
@@ -83,6 +84,12 @@ var _ = Describe("NamespacedClient", func() {
 				},
 			},
 		}
+		nameSpace = &corev1.Namespace{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:   fmt.Sprintf("namespace-%v", count),
+				Labels: map[string]string{"name": fmt.Sprintf("namespace-%v", count)},
+			},
+		}
 		acDep = appsv1applyconfigurations.Deployment(dep.Name, "").
 			WithLabels(dep.Labels).
 			WithSpec(appsv1applyconfigurations.DeploymentSpec().
@@ -134,10 +141,13 @@ var _ = Describe("NamespacedClient", func() {
 			var err error
 			dep, err = clientset.AppsV1().Deployments(ns).Create(ctx, dep, metav1.CreateOptions{})
 			Expect(err).NotTo(HaveOccurred())
+			nameSpace, err = clientset.CoreV1().Namespaces().Create(ctx, nameSpace, metav1.CreateOptions{})
+			Expect(err).NotTo(HaveOccurred())
 		})
 
 		AfterEach(func(ctx SpecContext) {
 			deleteDeployment(ctx, dep, ns)
+			deleteNamespace(ctx, nameSpace)
 		})
 
 		It("should successfully List objects when namespace is not specified with the object", func(ctx SpecContext) {
@@ -150,7 +160,7 @@ var _ = Describe("NamespacedClient", func() {
 		})
 
 		It("should successfully List objects when object is not namespaced scoped", func(ctx SpecContext) {
-			result := &corev1.NodeList{}
+			result := &corev1.NamespaceList{}
 			opts := &client.ListOptions{}
 			Expect(getClient().List(ctx, result, opts)).NotTo(HaveOccurred())
 			Expect(result.Items).NotTo(BeEmpty())
diff --git a/pkg/cluster/cluster_test.go b/pkg/cluster/cluster_test.go
index c08a7424..c275ff0b 100644
--- a/pkg/cluster/cluster_test.go
+++ b/pkg/cluster/cluster_test.go
@@ -40,7 +40,6 @@ var _ = Describe("cluster.Cluster", func() {
 			c, err := New(nil)
 			Expect(c).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Config"))
-
 		})
 
 		It("should return an error if it can't create a RestMapper", func() {
@@ -50,7 +49,6 @@ var _ = Describe("cluster.Cluster", func() {
 			})
 			Expect(c).To(BeNil())
 			Expect(err).To(Equal(expected))
-
 		})
 
 		It("should return an error it can't create a client.Client", func() {
@@ -96,7 +94,6 @@ var _ = Describe("cluster.Cluster", func() {
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("expected error"))
 		})
-
 	})
 
 	Describe("Start", func() {
@@ -160,7 +157,13 @@ var _ = Describe("cluster.Cluster", func() {
 	It("should provide a function to get the EventRecorder", func() {
 		c, err := New(cfg)
 		Expect(err).NotTo(HaveOccurred())
-		Expect(c.GetEventRecorderFor("test")).NotTo(BeNil())
+		Expect(c.GetEventRecorder("test")).NotTo(BeNil())
+	})
+
+	It("should provide a function to get the deprecated EventRecorder", func() {
+		c, err := New(cfg)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(c.GetEventRecorderFor("test")).NotTo(BeNil()) //nolint:staticcheck
 	})
 	It("should provide a function to get the APIReader", func() {
 		c, err := New(cfg)
diff --git a/pkg/internal/recorder/recorder_integration_test.go b/pkg/internal/recorder/recorder_integration_test.go
index c278fbde..06107016 100644
--- a/pkg/internal/recorder/recorder_integration_test.go
+++ b/pkg/internal/recorder/recorder_integration_test.go
@@ -21,7 +21,9 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes/scheme"
 	ref "k8s.io/client-go/tools/reference"
@@ -36,20 +38,22 @@ import (
 )
 
 var _ = Describe("recorder", func() {
-	Describe("recorder", func() {
+	Describe("deprecated recorder", func() {
 		It("should publish events", func(ctx SpecContext) {
 			By("Creating the Manager")
 			cm, err := manager.New(cfg, manager.Options{})
 			Expect(err).NotTo(HaveOccurred())
 
 			By("Creating the Controller")
-			recorder := cm.GetEventRecorderFor("test-recorder")
+			deprecatedRecorder := cm.GetEventRecorderFor("test-deprecated-recorder") //nolint:staticcheck
+			recorder := cm.GetEventRecorder("test-recorder")
 			instance, err := controller.New("foo-controller", cm, controller.Options{
 				Reconciler: reconcile.Func(
 					func(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
 						dp, err := clientset.AppsV1().Deployments(request.Namespace).Get(ctx, request.Name, metav1.GetOptions{})
 						Expect(err).NotTo(HaveOccurred())
-						recorder.Event(dp, corev1.EventTypeNormal, "test-reason", "test-msg")
+						deprecatedRecorder.Event(dp, corev1.EventTypeNormal, "deprecated-test-reason", "deprecated-test-msg")
+						recorder.Eventf(dp, nil, corev1.EventTypeNormal, "test-reason", "test-action", "test-note")
 						return reconcile.Result{}, nil
 					}),
 			})
@@ -66,7 +70,7 @@ var _ = Describe("recorder", func() {
 			}()
 
 			deployment := &appsv1.Deployment{
-				ObjectMeta: metav1.ObjectMeta{Name: "deployment-name"},
+				ObjectMeta: metav1.ObjectMeta{Name: "deprecated-deployment-name"},
 				Spec: appsv1.DeploymentSpec{
 					Selector: &metav1.LabelSelector{
 						MatchLabels: map[string]string{"foo": "bar"},
@@ -89,23 +93,42 @@ var _ = Describe("recorder", func() {
 			deployment, err = clientset.AppsV1().Deployments("default").Create(ctx, deployment, metav1.CreateOptions{})
 			Expect(err).NotTo(HaveOccurred())
 
-			By("Validate event is published as expected")
-			evtWatcher, err := clientset.CoreV1().Events("default").Watch(ctx, metav1.ListOptions{})
+			// watch both deprecated and new events based on the reason
+			By("Validate deprecated event is published as expected")
+			deprecatedEvtWatcher, err := clientset.CoreV1().Events("default").Watch(ctx,
+				metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("reason", "deprecated-test-reason").String()})
 			Expect(err).NotTo(HaveOccurred())
 
-			resultEvent := <-evtWatcher.ResultChan()
+			resultEvent := <-deprecatedEvtWatcher.ResultChan()
 
 			Expect(resultEvent.Type).To(Equal(watch.Added))
-			evt, isEvent := resultEvent.Object.(*corev1.Event)
+			deprecatedEvt, isEvent := resultEvent.Object.(*corev1.Event)
 			Expect(isEvent).To(BeTrue())
 
 			dpRef, err := ref.GetReference(scheme.Scheme, deployment)
 			Expect(err).NotTo(HaveOccurred())
 
-			Expect(evt.InvolvedObject).To(Equal(*dpRef))
+			Expect(deprecatedEvt.InvolvedObject).To(Equal(*dpRef))
+			Expect(deprecatedEvt.Type).To(Equal(corev1.EventTypeNormal))
+			Expect(deprecatedEvt.Reason).To(Equal("deprecated-test-reason"))
+			Expect(deprecatedEvt.Message).To(Equal("deprecated-test-msg"))
+
+			By("Validate event is published as expected")
+			evtWatcher, err := clientset.EventsV1().Events("default").Watch(ctx,
+				metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("reason", "test-reason").String()})
+			Expect(err).NotTo(HaveOccurred())
+
+			resultEvent = <-evtWatcher.ResultChan()
+
+			Expect(resultEvent.Type).To(Equal(watch.Added))
+			evt, isEvent := resultEvent.Object.(*eventsv1.Event)
+			Expect(isEvent).To(BeTrue())
+
+			Expect(evt.Regarding).To(Equal(*dpRef))
 			Expect(evt.Type).To(Equal(corev1.EventTypeNormal))
 			Expect(evt.Reason).To(Equal("test-reason"))
-			Expect(evt.Message).To(Equal("test-msg"))
+			Expect(evt.Action).To(Equal("test-action"))
+			Expect(evt.Note).To(Equal("test-note"))
 		})
 	})
 })
diff --git a/pkg/internal/recorder/recorder_test.go b/pkg/internal/recorder/recorder_test.go
index e226e165..e592a1e1 100644
--- a/pkg/internal/recorder/recorder_test.go
+++ b/pkg/internal/recorder/recorder_test.go
@@ -21,15 +21,16 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"k8s.io/client-go/kubernetes/scheme"
+	eventsv1client "k8s.io/client-go/kubernetes/typed/events/v1"
+	"k8s.io/client-go/tools/events"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/internal/recorder"
 )
 
 var _ = Describe("recorder.Provider", func() {
-	makeBroadcaster := func() (record.EventBroadcaster, bool) { return record.NewBroadcaster(), true }
 	Describe("NewProvider", func() {
 		It("should return a provider instance and a nil error.", func() {
-			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(provider).NotTo(BeNil())
 			Expect(err).NotTo(HaveOccurred())
 		})
@@ -38,18 +39,36 @@ var _ = Describe("recorder.Provider", func() {
 			// Invalid the config
 			cfg1 := *cfg
 			cfg1.Host = "invalid host"
-			_, err := recorder.NewProvider(&cfg1, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+			_, err := recorder.NewProvider(&cfg1, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("failed to init client"))
 		})
 	})
+	Describe("GetEventRecorderFor", func() {
+		It("should return a deprecated recorder instance.", func() {
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
+			Expect(err).NotTo(HaveOccurred())
+
+			recorder := provider.GetEventRecorderFor("test")
+			Expect(recorder).NotTo(BeNil())
+		})
+	})
 	Describe("GetEventRecorder", func() {
 		It("should return a recorder instance.", func() {
-			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster)
+			provider, err := recorder.NewProvider(cfg, httpClient, scheme.Scheme, logr.Discard(), makeBroadcaster())
 			Expect(err).NotTo(HaveOccurred())
 
-			recorder := provider.GetEventRecorderFor("test")
+			recorder := provider.GetEventRecorder("test")
 			Expect(recorder).NotTo(BeNil())
 		})
 	})
 })
+
+func makeBroadcaster() func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+	evtCl, err := eventsv1client.NewForConfigAndClient(cfg, httpClient)
+	Expect(err).NotTo(HaveOccurred())
+
+	return func() (record.EventBroadcaster, events.EventBroadcaster, bool) {
+		return record.NewBroadcaster(), events.NewBroadcaster(&events.EventSinkImpl{Interface: evtCl}), true
+	}
+}
diff --git a/pkg/manager/manager_test.go b/pkg/manager/manager_test.go
index 4363d62f..4bf55357 100644
--- a/pkg/manager/manager_test.go
+++ b/pkg/manager/manager_test.go
@@ -36,7 +36,10 @@ import (
 	"go.uber.org/goleak"
 	coordinationv1 "k8s.io/api/coordination/v1"
 	corev1 "k8s.io/api/core/v1"
+	eventsv1 "k8s.io/api/events/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/rest"
@@ -60,7 +63,6 @@ var _ = Describe("manger.Manager", func() {
 			m, err := New(nil, Options{})
 			Expect(m).To(BeNil())
 			Expect(err.Error()).To(ContainSubstring("must specify Config"))
-
 		})
 
 		It("should return an error if it can't create a RestMapper", func() {
@@ -70,7 +72,6 @@ var _ = Describe("manger.Manager", func() {
 			})
 			Expect(m).To(BeNil())
 			Expect(err).To(Equal(expected))
-
 		})
 
 		It("should return an error it can't create a client.Client", func() {
@@ -207,7 +208,6 @@ var _ = Describe("manger.Manager", func() {
 				}
 				// Don't leak routines
 				<-mgrDone
-
 			})
 			It("should disable gracefulShutdown when stopping to lead", func(ctx SpecContext) {
 				m, err := New(cfg, Options{
@@ -443,7 +443,6 @@ var _ = Describe("manger.Manager", func() {
 				Expect(ok).To(BeTrue())
 				_, isLeaseLock := cm.resourceLock.(*resourcelock.LeaseLock)
 				Expect(isLeaseLock).To(BeTrue())
-
 			})
 			It("should use the specified ResourceLock", func() {
 				m, err := New(cfg, Options{
@@ -671,7 +670,7 @@ var _ = Describe("manger.Manager", func() {
 	})
 
 	Describe("Start", func() {
-		var startSuite = func(options Options, callbacks ...func(Manager)) {
+		startSuite := func(options Options, callbacks ...func(Manager)) {
 			It("should Start each Component", func(ctx SpecContext) {
 				m, err := New(cfg, options)
 				Expect(err).NotTo(HaveOccurred())
@@ -1256,7 +1255,6 @@ var _ = Describe("manger.Manager", func() {
 				<-managerStopDone
 				Expect(time.Since(beforeDone)).To(BeNumerically(">=", 1500*time.Millisecond))
 			})
-
 		}
 
 		Context("with defaults", func() {
@@ -1790,7 +1788,6 @@ var _ = Describe("manger.Manager", func() {
 			err = m.Start(ctx)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(Equal("manager already started"))
-
 		})
 	})
 
@@ -1810,7 +1807,7 @@ var _ = Describe("manger.Manager", func() {
 		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
 	})
 
-	It("should not leak goroutines if the default event broadcaster is used & events are emitted", func(specCtx SpecContext) {
+	It("should not leak goroutines if the deprecated event broadcaster is used & events are emitted", func(specCtx SpecContext) {
 		currentGRs := goleak.IgnoreCurrent()
 
 		m, err := New(cfg, Options{ /* implicit: default setting for EventBroadcaster */ })
@@ -1820,7 +1817,7 @@ var _ = Describe("manger.Manager", func() {
 		ns := corev1.Namespace{}
 		ns.Name = "default"
 
-		recorder := m.GetEventRecorderFor("rock-and-roll")
+		recorder := m.GetEventRecorderFor("rock-and-roll") //nolint:staticcheck
 		Expect(m.Add(RunnableFunc(func(_ context.Context) error {
 			recorder.Event(&ns, "Warning", "BallroomBlitz", "yeah, yeah, yeah-yeah-yeah")
 			return nil
@@ -1858,6 +1855,60 @@ var _ = Describe("manger.Manager", func() {
 		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
 	})
 
+	It("should not leak goroutines if the default event broadcaster is used & events are emitted", func(specCtx SpecContext) {
+		currentGRs := goleak.IgnoreCurrent()
+
+		m, err := New(cfg, Options{ /* implicit: default setting for EventBroadcaster */ })
+		Expect(err).NotTo(HaveOccurred())
+
+		By("adding a runnable that emits an event")
+		ns := corev1.Namespace{}
+		ns.Name = "default"
+
+		recorder := m.GetEventRecorder("rock-and-roll")
+		Expect(m.Add(RunnableFunc(func(_ context.Context) error {
+			recorder.Eventf(&ns, nil, "Warning", "BallroomBlitz", "dance action", "yeah, yeah, yeah-yeah-yeah")
+			return nil
+		}))).To(Succeed())
+
+		By("starting the manager & waiting till we've sent our event")
+		ctx, cancel := context.WithCancel(specCtx)
+		doneCh := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			defer close(doneCh)
+			Expect(m.Start(ctx)).To(Succeed())
+		}()
+		<-m.Elected()
+
+		Eventually(func() *eventsv1.Event {
+			evts, err := clientset.EventsV1().Events("").List(ctx,
+				metav1.ListOptions{FieldSelector: fields.OneTermEqualSelector("regarding.name", ns.Name).String()})
+			Expect(err).NotTo(HaveOccurred())
+
+			for i, evt := range evts.Items {
+				if evt.Reason == "BallroomBlitz" {
+					return &evts.Items[i]
+				}
+			}
+			return nil
+		}).ShouldNot(BeNil())
+
+		// Sleep between broadcasting start and shutdown to prevent a race condition
+		// that causes goroutines to leak.
+		// See pkg/internal/recorder/recorder.go:103 for more info.
+		time.Sleep(3 * time.Second)
+
+		By("making sure there's no extra go routines still running after we stop")
+		cancel()
+		<-doneCh
+
+		// force-close keep-alive connections.  These'll time anyway (after
+		// like 30s or so) but force it to speed up the tests.
+		clientTransport.CloseIdleConnections()
+		Eventually(func() error { return goleak.Find(currentGRs) }).Should(Succeed())
+	})
+
 	It("should not leak goroutines when a runnable returns error slowly after being signaled to stop", func(specCtx SpecContext) {
 		// This test reproduces the race condition where the manager's Start method
 		// exits due to context cancellation, leaving no one to drain errChan
@@ -1872,7 +1923,7 @@ var _ = Describe("manger.Manager", func() {
 		Expect(err).NotTo(HaveOccurred())
 
 		// Add the slow runnable that will return an error after some delay
-		for i := 0; i < 3; i++ {
+		for range 3 {
 			slowRunnable := RunnableFunc(func(c context.Context) error {
 				<-c.Done()
 
@@ -1938,10 +1989,15 @@ var _ = Describe("manger.Manager", func() {
 		Expect(m.GetFieldIndexer()).To(Equal(mgr.cluster.GetFieldIndexer()))
 	})
 
+	It("should provide a function to get the deprecated EventRecorder", func() {
+		m, err := New(cfg, Options{})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(m.GetEventRecorderFor("test")).NotTo(BeNil()) //nolint:staticcheck
+	})
 	It("should provide a function to get the EventRecorder", func() {
 		m, err := New(cfg, Options{})
 		Expect(err).NotTo(HaveOccurred())
-		Expect(m.GetEventRecorderFor("test")).NotTo(BeNil())
+		Expect(m.GetEventRecorder("test")).NotTo(BeNil())
 	})
 	It("should provide a function to get the APIReader", func() {
 		m, err := New(cfg, Options{})
@@ -2020,8 +2076,7 @@ var _ = Describe("manger.Manager", func() {
 	})
 })
 
-type runnableError struct {
-}
+type runnableError struct{}
 
 func (runnableError) Error() string {
 	return "not feeling like that"
diff --git a/pkg/recorder/example_test.go b/pkg/recorder/example_test.go
index 969420d8..47f14ff7 100644
--- a/pkg/recorder/example_test.go
+++ b/pkg/recorder/example_test.go
@@ -18,31 +18,39 @@ package recorder_test
 
 import (
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 
 	_ "github.com/onsi/ginkgo/v2"
 	"sigs.k8s.io/controller-runtime/pkg/recorder"
 )
 
 var (
-	recorderProvider recorder.Provider
-	somePod          *corev1.Pod // the object you're reconciling, for example
+	recorderProvider  recorder.Provider
+	somePod           *corev1.Pod    // the object you're reconciling, for example
+	someRelatedObject runtime.Object // another object related to the reconciled object and the event.
 )
 
 func Example_event() {
 	// recorderProvider is a recorder.Provider
-	recorder := recorderProvider.GetEventRecorderFor("my-controller")
+	deprecatedRecorder := recorderProvider.GetEventRecorderFor("my-controller")
 
 	// emit an event with a fixed message
-	recorder.Event(somePod, corev1.EventTypeWarning,
+	deprecatedRecorder.Event(somePod, corev1.EventTypeWarning,
 		"WrongTrousers", "It's the wrong trousers, Gromit!")
 }
 
 func Example_eventf() {
 	// recorderProvider is a recorder.Provider
-	recorder := recorderProvider.GetEventRecorderFor("my-controller")
+	deprecatedRecorder := recorderProvider.GetEventRecorderFor("my-controller")
 
 	// emit an event with a variable message
 	mildCheese := "Wensleydale"
-	recorder.Eventf(somePod, corev1.EventTypeNormal,
+	deprecatedRecorder.Eventf(somePod, corev1.EventTypeNormal,
 		"DislikesCheese", "Not even %s?", mildCheese)
+
+	recorder := recorderProvider.GetEventRecorder("my-controller")
+
+	// emit an event with a fixed message
+	recorder.Eventf(somePod, someRelatedObject, corev1.EventTypeWarning,
+		"WrongTrousers", "getting dressed", "It's the wrong trousers, Gromit!")
 }