ronantakizawa/codereview-bench huggingface dataset:
Repository: kubernetes/kubernetes
File: pkg/kubelet/podcertificate/podcertificatemanager_test.go (Go)

Code before:
/*
Copyright 2025 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package podcertificate

import (
	"context"
	"crypto/x509"
	"encoding/pem"
	"fmt"
	"slices"
	"strings"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"
	certsv1beta1 "k8s.io/api/certificates/v1beta1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes/fake"
	certlistersv1beta1 "k8s.io/client-go/listers/certificates/v1beta1"
	corelistersv1 "k8s.io/client-go/listers/core/v1"
	"k8s.io/client-go/tools/cache"
	"k8s.io/kubernetes/test/utils/hermeticpodcertificatesigner"
	"k8s.io/kubernetes/test/utils/ktesting"
	testclock "k8s.io/utils/clock/testing"
	"k8s.io/utils/ptr"
)

func TestTransitionInitialToWait(t *testing.T) {
	ctx, cancel := context.WithCancel(ktesting.Init(t))
	defer cancel()

Review comment (suggestion):
/hold
revert all changes to this file or provide a good reason about why they are valid in a separate commit.

Code after:
/*
Copyright 2025 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package podcertificate

import (
	"context"
	"crypto/x509"
	"encoding/pem"
	"fmt"
	"math/rand"
	"slices"
	"sync/atomic"
	"testing"
	"time"

	"github.com/google/go-cmp/cmp"
	certsv1beta1 "k8s.io/api/certificates/v1beta1"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/apimachinery/pkg/watch"
	"k8s.io/client-go/informers"
	"k8s.io/client-go/kubernetes/fake"
	certlistersv1beta1 "k8s.io/client-go/listers/certificates/v1beta1"
	corelistersv1 "k8s.io/client-go/listers/core/v1"
	k8stesting "k8s.io/client-go/testing"
	"k8s.io/client-go/tools/cache"
	"k8s.io/kubernetes/test/utils/hermeticpodcertificatesigner"
	"k8s.io/kubernetes/test/utils/ktesting"
	testclock "k8s.io/utils/clock/testing"
	"k8s.io/utils/ptr"
)
Repository: kubernetes/kubernetes
File: pkg/kubelet/kubelet_volumes_test.go (Go)

Code before:
				Volumes: []v1.Volume{
					{
						Name: "vol3",
						VolumeSource: v1.VolumeSource{
							RBD: &v1.RBDVolumeSource{
								RBDImage: "fake3",
							},
						},
					},
				},
			},
		},
	}

	tCtx := ktesting.Init(t)
	defer tCtx.Cancel("test has completed")
	go kubelet.volumeManager.Run(tCtx, kubelet.sourcesReady)

	kubelet.podManager.SetPods(pods)
	for _, pod := range pods {
		err := kubelet.volumeManager.WaitForAttachAndMount(context.Background(), pod)
		assert.NoError(t, err)
	}

	for _, pod := range pods {
		podVolumesExist := kubelet.podVolumesExist(klog.TODO(), pod.UID)
		assert.True(t, podVolumesExist, "pod %q", pod.UID)
	}
}

func TestPodVolumeDeadlineAttachAndMount(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping test in short mode.")
	}

	testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
		false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)

	defer testKubelet.Cleanup()
	kubelet := testKubelet.kubelet

	// any test cases added here should have volumes that fail to mount
	pods := []*v1.Pod{
		{
			ObjectMeta: metav1.ObjectMeta{
				Name: "pod1",
				UID:  "pod1uid",
			},
			Spec: v1.PodSpec{
				Containers: []v1.Container{
					{

Review comment (suggestion):
```suggestion
		podVolumesExist := kubelet.podVolumesExist(tCtx, pod.UID)
```

Code after:
					{
						Name: "vol3",
						VolumeSource: v1.VolumeSource{
							RBD: &v1.RBDVolumeSource{
								RBDImage: "fake3",
							},
						},
					},
				},
			},
		},
	}

	logger, tCtx := ktesting.NewTestContext(t)
	defer tCtx.Cancel("test has completed")
	go kubelet.volumeManager.Run(tCtx, kubelet.sourcesReady)

	kubelet.podManager.SetPods(pods)
	for _, pod := range pods {
		err := kubelet.volumeManager.WaitForAttachAndMount(tCtx, pod)
		assert.NoError(t, err)
	}

	for _, pod := range pods {
		podVolumesExist := kubelet.podVolumesExist(logger, pod.UID)
		assert.True(t, podVolumesExist, "pod %q", pod.UID)
	}
}

func TestPodVolumeDeadlineAttachAndMount(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping test in short mode.")
	}

	testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
		false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)

	defer testKubelet.Cleanup()
	kubelet := testKubelet.kubelet

	// any test cases added here should have volumes that fail to mount
	pods := []*v1.Pod{
		{
			ObjectMeta: metav1.ObjectMeta{
				Name: "pod1",
				UID:  "pod1uid",
			},
			Spec: v1.PodSpec{
				Containers: []v1.Container{
					{
						Name: "container1",
Repository: kubernetes/kubernetes
File: pkg/kubelet/kubelet_volumes_test.go (Go)

Code before:
				Volumes: []v1.Volume{
					{
						Name: "vol3",
						VolumeSource: v1.VolumeSource{
							RBD: &v1.RBDVolumeSource{
								RBDImage: "fake3",
							},
						},
					},
				},
			},
		},
	}

	tCtx := ktesting.Init(t)
	defer tCtx.Cancel("test has completed")
	go kubelet.volumeManager.Run(tCtx, kubelet.sourcesReady)

	kubelet.podManager.SetPods(pods)
	for _, pod := range pods {
		err := kubelet.volumeManager.WaitForAttachAndMount(context.Background(), pod)
		assert.NoError(t, err)
	}

	for _, pod := range pods {
		podVolumesExist := kubelet.podVolumesExist(klog.FromContext(tCtx), pod.UID)
		assert.True(t, podVolumesExist, "pod %q", pod.UID)
	}
}

func TestPodVolumeDeadlineAttachAndMount(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping test in short mode.")
	}

	testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
		false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)

	defer testKubelet.Cleanup()
	kubelet := testKubelet.kubelet

	// any test cases added here should have volumes that fail to mount
	pods := []*v1.Pod{
		{
			ObjectMeta: metav1.ObjectMeta{
				Name: "pod1",
				UID:  "pod1uid",
			},
			Spec: v1.PodSpec{
				Containers: []v1.Container{
					{

Review comment (suggestion):
```suggestion
		podVolumesExist := kubelet.podVolumesExist(logger, pod.UID)
```

Code after:
					{
						Name: "vol3",
						VolumeSource: v1.VolumeSource{
							RBD: &v1.RBDVolumeSource{
								RBDImage: "fake3",
							},
						},
					},
				},
			},
		},
	}

	logger, tCtx := ktesting.NewTestContext(t)
	defer tCtx.Cancel("test has completed")
	go kubelet.volumeManager.Run(tCtx, kubelet.sourcesReady)

	kubelet.podManager.SetPods(pods)
	for _, pod := range pods {
		err := kubelet.volumeManager.WaitForAttachAndMount(tCtx, pod)
		assert.NoError(t, err)
	}

	for _, pod := range pods {
		podVolumesExist := kubelet.podVolumesExist(logger, pod.UID)
		assert.True(t, podVolumesExist, "pod %q", pod.UID)
	}
}

func TestPodVolumeDeadlineAttachAndMount(t *testing.T) {
	if testing.Short() {
		t.Skip("skipping test in short mode.")
	}

	testKubelet := newTestKubeletWithImageList(t, nil /*imageList*/, false, /* controllerAttachDetachEnabled */
		false /*initFakeVolumePlugin*/, true /*localStorageCapacityIsolation*/, false /*excludePodAdmitHandlers*/, false /*enableResizing*/)

	defer testKubelet.Cleanup()
	kubelet := testKubelet.kubelet

	// any test cases added here should have volumes that fail to mount
	pods := []*v1.Pod{
		{
			ObjectMeta: metav1.ObjectMeta{
				Name: "pod1",
				UID:  "pod1uid",
			},
			Spec: v1.PodSpec{
				Containers: []v1.Container{
					{
						Name: "container1",
Repository: kubernetes/kubernetes
File: test/images/agnhost/mounttest/mt_utils.go (Go)

Code before:
*/

package mounttest

import (
	"fmt"
	"os"
	"syscall"
)

func umask(mask int) int {
	return syscall.Umask(mask)
}

func fsType(path string) error {
	if path == "" {
		return nil
	}

	buf := syscall.Statfs_t{}
	if err := syscall.Statfs(path, &buf); err != nil {
		fmt.Printf("error from statfs(%q): %v\n", path, err)
		return err
	}

	fmt.Print(formatFsType(path, int64(buf.Type)))

	return nil
}

func fileMode(path string) error {
	if path == "" {
		return nil
	}

	fileinfo, err := os.Stat(path)
	if err != nil {
		fmt.Printf("error from Stat(%q): %v\n", path, err)
		return err
	}

	fmt.Printf("mode of file %q: %v\n", path, fileinfo.Mode())
	return nil
}

func filePerm(path string) error {
	if path == "" {
		return nil
	}

	fileinfo, err := os.Stat(path)

Review comment (suggestion):
I'd suggest making the platform-specific bit just turn `int64 → string` to avoid repeating the message a bunch of times

```
fmt.Printf("mount type of %q: %s\n", path, formatFsType(int64(buf.Type)))
```

Code after:
*/

package mounttest

import (
	"fmt"
	"os"
	"syscall"
)

func umask(mask int) int {
	return syscall.Umask(mask)
}

func fsType(path string) error {
	if path == "" {
		return nil
	}

	buf := syscall.Statfs_t{}
	if err := syscall.Statfs(path, &buf); err != nil {
		fmt.Printf("error from statfs(%q): %v\n", path, err)
		return err
	}

	fmt.Printf("mount type of %q: %s\n", path, formatFsType(int64(buf.Type)))

	return nil
}

func fileMode(path string) error {
	if path == "" {
		return nil
	}

	fileinfo, err := os.Stat(path)
	if err != nil {
		fmt.Printf("error from Stat(%q): %v\n", path, err)
		return err
	}

	fmt.Printf("mode of file %q: %v\n", path, fileinfo.Mode())
	return nil
}

func filePerm(path string) error {
	if path == "" {
		return nil
	}

	fileinfo, err := os.Stat(path)
Repository: kubernetes/kubernetes
File: test/images/agnhost/mounttest/mt_utils_linux.go (Go)

Code before:
//go:build linux

/*
Copyright 2019 The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package mounttest

import (
	"strconv"

	"golang.org/x/sys/unix"
)

// formatFsType returns a string representation of the filesystem type.
// It translates known filesystem magic numbers to human-readable names.
func formatFsType(fsType int64) string {

Review comment (security):
nit: drop the year from all the new files

```suggestion
Copyright The Kubernetes Authors.
```

Code after:
//go:build linux

/*
Copyright The Kubernetes Authors.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package mounttest

import (
	"strconv"

	"golang.org/x/sys/unix"
)

// formatFsType returns a string representation of the filesystem type.
// It translates known filesystem magic numbers to human-readable names.
func formatFsType(fsType int64) string {
Repository: kubernetes/kubernetes
File: pkg/kubelet/cm/qos_container_manager_linux_test.go (Go)

Code before:
								},
							},
						},
					},
				}
			},
			expectedBurstableCPUShares: 1024,
		},
	}

	for _, testCase := range tests {

		t.Run(testCase.name, func(t *testing.T) {

			logger := logr.Discard()

			testContainerManager, err := createTestQOSContainerManager(logger)
			if err != nil {
				t.Fatalf("Unable to create Test Qos Container Manager: %s", err)
				return
			}

			fakecgroupManager := &fakeCgroupManager{}
			testContainerManager.cgroupManager = fakecgroupManager

			ctx, cancel := context.WithCancel(context.Background())
			defer cancel()

			err = testContainerManager.Start(ctx, func() v1.ResourceList { return v1.ResourceList{} }, testCase.testPods)

			if err != nil {
				t.Fatalf("Start() failed: %s", err)
			}

			// UpdateCgroups() is expected to update all QoS cgroups on each call
			// based on the current active pod set.
			err = testContainerManager.UpdateCgroups(logger)
			if err != nil {
				t.Fatalf("Error in UpdateCgroups(): %s", err)
			}
			cancel()

			// These flags will be used to check whether UpdateCgroups()
			// is updating the CPU shares for all QoS classes (cgroups)
			foundBurstable := false
			foundBestEffort := false
			foundGuaranteed := false

			// Start() initiates a background UpdateCgroups() goroutine which may
			// still be running. Take a snapshot to avoid observing updates from
			// the background UpdateCgroups() instead of the explicit UpdateCgroups() call.

Review comment (suggestion):
let's use `ktesting` which will give us both a context and a logger (sorry for late notice, realized thinking about `context.Background()`

Code after:
							},
						},
					},
				}
			},
			expectedBurstableCPUShares: 1024,
		},
	}

	for _, testCase := range tests {

		t.Run(testCase.name, func(t *testing.T) {

			logger, ctx := ktesting.NewTestContext(t)

			testContainerManager, err := createTestQOSContainerManager(logger)
			if err != nil {
				t.Fatalf("Unable to create Test Qos Container Manager: %s", err)
				return
			}

			fakecgroupManager := &fakeCgroupManager{}
			testContainerManager.cgroupManager = fakecgroupManager

			ctx, cancel := context.WithCancel(ctx)
			defer cancel()

			err = testContainerManager.Start(ctx, func() v1.ResourceList { return v1.ResourceList{} }, testCase.testPods)

			if err != nil {
				t.Fatalf("Start() failed: %s", err)
			}

			// UpdateCgroups() is expected to update all QoS cgroups on each call
			// based on the current active pod set.
			err = testContainerManager.UpdateCgroups(logger)
			if err != nil {
				t.Fatalf("Error in UpdateCgroups(): %s", err)
			}
			cancel()

			// UpdateCgroups() may also be running in the background reconciliation loop (because of Start()).
			// Retry until a consistent snapshot containing all QoS cgroup updates is observed.
			//
			// A small bounded retry count is sufficient here because UpdateCgroups()
			// is synchronous and expected to complete quickly. This avoids test flakiness
			// without risking an unbounded wait.
			maxRetryAttempts := 5

			var (
				foundBurstable  bool
Repository: kubernetes/kubernetes
File: pkg/scheduler/framework/interface.go (Go)

Code before:

	// HasScorePlugins returns true if at least one Score plugin is defined.
	HasScorePlugins() bool

	// ListPlugins returns a map of extension point name to list of configured Plugins.
	ListPlugins() *config.Plugins

	// PercentageOfNodesToScore returns percentageOfNodesToScore associated to a profile.
	PercentageOfNodesToScore() *int32

	// SetPodNominator sets the PodNominator
	SetPodNominator(nominator fwk.PodNominator)
	// SetPodActivator sets the PodActivator
	SetPodActivator(activator fwk.PodActivator)
	// SetAPICacher sets the APICacher
	SetAPICacher(apiCacher fwk.APICacher)

	// Close calls Close method of each plugin.
	Close() error
}

func NewPostFilterResultWithNominatedNode(nodeName string) *fwk.PostFilterResult {
	return NewPostFilterResult(nodeName, nil)
}

func NewPostFilterResult(nodeName string, pods []*v1.Pod) *fwk.PostFilterResult {
	return &fwk.PostFilterResult{
		NominatingInfo: &fwk.NominatingInfo{
			NominatedNodeName: nodeName,
			NominatingMode:    fwk.ModeOverride,
		},
		Victims: pods,
	}
}


Review comment (suggestion):
```suggestion
func NewPostFilterResultWithNominatedNode(nodeName string) *fwk.PostFilterResult {
	return NewPostFilterResult(nodeName, nil)
}

func NewPostFilterResult(nodeName string, victims []*v1.Pod) *fwk.PostFilterResult {
```

Code after:
	RunBindPlugins(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status

	// HasFilterPlugins returns true if at least one Filter plugin is defined.
	HasFilterPlugins() bool

	// HasPostFilterPlugins returns true if at least one PostFilter plugin is defined.
	HasPostFilterPlugins() bool

	// HasScorePlugins returns true if at least one Score plugin is defined.
	HasScorePlugins() bool

	// ListPlugins returns a map of extension point name to list of configured Plugins.
	ListPlugins() *config.Plugins

	// PercentageOfNodesToScore returns percentageOfNodesToScore associated to a profile.
	PercentageOfNodesToScore() *int32

	// SetPodNominator sets the PodNominator
	SetPodNominator(nominator fwk.PodNominator)
	// SetPodActivator sets the PodActivator
	SetPodActivator(activator fwk.PodActivator)
	// SetAPICacher sets the APICacher
	SetAPICacher(apiCacher fwk.APICacher)

	// Close calls Close method of each plugin.
	Close() error
}

// NewPostFilterResult creates PostFilterResult with a provided nominated node
// and a list of victims (if any) that need to be preempted to make the pod schedulable.
func NewPostFilterResult(nodeName string, victims []*v1.Pod) *fwk.PostFilterResult {
	return &fwk.PostFilterResult{
		NominatingInfo: &fwk.NominatingInfo{
			NominatedNodeName: nodeName,
			NominatingMode:    fwk.ModeOverride,
		},
		Victims: victims,
	}
}

// Deprecated: use NewPostFilterResult instead, even if no victims were identified.
// Deprecated in Kubernetes 1.36 and will be removed in 1.38.
func NewPostFilterResultWithNominatedNode(nodeName string) *fwk.PostFilterResult {
	return NewPostFilterResult(nodeName, nil)
}
Repository: kubernetes/kubernetes
File: pkg/scheduler/framework/interface.go (Go)

Code before:
	// SetAPICacher sets the APICacher
	SetAPICacher(apiCacher fwk.APICacher)

	// Close calls Close method of each plugin.
	Close() error
}

// NewPostFilterResult creates PostFilterResult with a provided nominated node
// and a list of victims that need to be preempted to make the pod schedulable.
func NewPostFilterResult(nodeName string, victims []*v1.Pod) *fwk.PostFilterResult {
	return &fwk.PostFilterResult{
		NominatingInfo: &fwk.NominatingInfo{
			NominatedNodeName: nodeName,
			NominatingMode:    fwk.ModeOverride,
		},
		Victims: victims,
	}
}

// NewPostFilterResultWithNominatedNode creates PostFilterResult with a provided
// nominated node. The list of victims is set to nil.
// This function is preserved for backward compatibility - PostFilter plugins
// should use the NewPostFilterResult function instead, however.
func NewPostFilterResultWithNominatedNode(nodeName string) *fwk.PostFilterResult {
	return NewPostFilterResult(nodeName, nil)
}


Review comment (suggestion):
// Deprecated: Use NewPostFilterResult instead even if no victims were identified.
// Deprecated in Kubernetes 1.36 and will be removed in 1.38.

Code after:
	// PercentageOfNodesToScore returns percentageOfNodesToScore associated to a profile.
	PercentageOfNodesToScore() *int32

	// SetPodNominator sets the PodNominator
	SetPodNominator(nominator fwk.PodNominator)
	// SetPodActivator sets the PodActivator
	SetPodActivator(activator fwk.PodActivator)
	// SetAPICacher sets the APICacher
	SetAPICacher(apiCacher fwk.APICacher)

	// Close calls Close method of each plugin.
	Close() error
}

// NewPostFilterResult creates PostFilterResult with a provided nominated node
// and a list of victims (if any) that need to be preempted to make the pod schedulable.
func NewPostFilterResult(nodeName string, victims []*v1.Pod) *fwk.PostFilterResult {
	return &fwk.PostFilterResult{
		NominatingInfo: &fwk.NominatingInfo{
			NominatedNodeName: nodeName,
			NominatingMode:    fwk.ModeOverride,
		},
		Victims: victims,
	}
}

// Deprecated: use NewPostFilterResult instead, even if no victims were identified.
// Deprecated in Kubernetes 1.36 and will be removed in 1.38.
func NewPostFilterResultWithNominatedNode(nodeName string) *fwk.PostFilterResult {
	return NewPostFilterResult(nodeName, nil)
}
Repository: kubernetes/kubernetes
File: pkg/scheduler/util/assumecache/assume_cache.go (Go)

Code before:
	c.eventHandlers = append(c.eventHandlers, handler)
	allObjs := c.listLocked(nil)
	for _, obj := range allObjs {
		c.eventQueue.WriteOne(func() {
			handler.OnAdd(obj, true)
		})
	}

	if c.handlerRegistration == nil {
		// No informer, so immediately synced.
		return syncedHandlerRegistration{}
	}

	return c.handlerRegistration
}

// emitEvents delivers all pending events that are in the queue, in the order
// in which they were stored there (FIFO). Only one goroutine at a time is
// delivering events, to ensure correct order.
func (c *AssumeCache) emitEvents() {
	c.rwMutex.Lock()
	for c.emittingEvents {
		// Wait for other caller of emitEvents to finish.
		// When it is done, it may or may not have drained
		// the events pushed by our caller.
		// We'll check below outselves.
		c.cond.Wait()
	}
	c.emittingEvents = true
	c.rwMutex.Unlock()

	defer func() {
		c.rwMutex.Lock()
		c.emittingEvents = false
		c.cond.Broadcast()
		c.rwMutex.Unlock()
	}()

	// When we get here, this instance of emitEvents is the active one.
	for {
		c.rwMutex.Lock()
		deliver, ok := c.eventQueue.ReadOne()
		c.rwMutex.Unlock()

		if !ok {
			return
		}
		func() {
			defer utilruntime.HandleCrash()
			deliver()
		}()

Review comment (suggestion):
```suggestion
		// We'll check below ourselves.
```

Code after:
		})
	}

	if c.handlerRegistration == nil {
		// No informer, so immediately synced.
		return syncedHandlerRegistration{}
	}

	return c.handlerRegistration
}

// emitEvents delivers all pending events that are in the queue, in the order
// in which they were stored there (FIFO). Only one goroutine at a time is
// delivering events, to ensure correct order.
func (c *AssumeCache) emitEvents() {
	c.rwMutex.Lock()
	for c.emittingEvents {
		// Wait for the active caller of emitEvents to finish.
		// When it is done, it may or may not have drained
		// the events pushed by our caller.
		// We'll check below ourselves.
		c.cond.Wait()
	}
	c.emittingEvents = true
	c.rwMutex.Unlock()

	defer func() {
		c.rwMutex.Lock()
		c.emittingEvents = false
		// Hand over the batton to one other goroutine, if there is one.
		// We don't need to wake up more than one because only one of
		// them would be able to grab the "emittingEvents" responsibility.
		c.cond.Signal()
		c.rwMutex.Unlock()
	}()

	// When we get here, this instance of emitEvents is the active one.
	for {
		c.rwMutex.Lock()
		deliver, ok := c.eventQueue.ReadOne()
		c.rwMutex.Unlock()

		if !ok {
			return
		}
		func() {
			defer utilruntime.HandleCrash()
			deliver()
		}()
	}
}