Addressed comments. Change OnPodDeleted to only populate the name, namespace and UID of the pod object to avoid sending confusing, stale end state.

pkg/kubelet/apis/pods/server.go

@@ -26,6 +26,7 @@ import (
 	"google.golang.org/grpc/status"
 
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/klog/v2"
@@ -151,7 +152,14 @@ func (s *PodsServer) OnPodUpdated(pod *v1.Pod, status v1.PodStatus, eventType wa
 
 // OnPodRemoved is called when a pod is removed.
 func (s *PodsServer) OnPodRemoved(pod *v1.Pod) {
-	s.broadcaster.Broadcast(PodWatchEvent{Type: watch.Deleted, UID: pod.UID, Pod: pod})
+	minimalPod := &v1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      pod.Name,
+			Namespace: pod.Namespace,
+			UID:       pod.UID,
+		},
+	}
+	s.broadcaster.Broadcast(PodWatchEvent{Type: watch.Deleted, UID: pod.UID, Pod: minimalPod})
 	logger := klog.FromContext(context.Background())
 	logger.Info("Pod removed broadcasted", "podUID", pod.UID)
 }

pkg/kubelet/apis/pods/server_unit_test.go

@@ -72,7 +72,11 @@ func TestStartEventLoop(t *testing.T) {
 	event = <-clientChannel
 	assert.Equal(t, watch.Deleted, event.Type)
 	assert.Equal(t, pod1.UID, event.UID)
-	assert.Equal(t, pod1, event.Pod)
+	require.NotNil(t, event.Pod)
+	assert.Equal(t, pod1.Name, event.Pod.Name)
+	assert.Equal(t, pod1.Namespace, event.Pod.Namespace)
+	assert.Equal(t, pod1.UID, event.Pod.UID)
+	assert.Empty(t, event.Pod.Spec.Containers)
 
 	select {
 	case event := <-clientChannel:
@@ -144,6 +148,20 @@ func TestWatchPods(t *testing.T) {
 	err = podOut.Unmarshal(event.Pod)
 	require.NoError(t, err)
 	assert.Equal(t, "pod1", podOut.Name)
+
+	// Trigger a removal
+	server.OnPodRemoved(pod1)
+
+	// Verify DELETED event
+	event = <-mockStream.EventCh
+	assert.Equal(t, podsv1alpha1.EventType_DELETED, event.Type)
+	podOut = &v1.Pod{}
+	err = podOut.Unmarshal(event.Pod)
+	require.NoError(t, err)
+	assert.Equal(t, "pod1", podOut.Name)
+	assert.Equal(t, "ns1", podOut.Namespace)
+	assert.Equal(t, pod1.UID, podOut.UID)
+	assert.Empty(t, podOut.Spec.Containers)
 }
 
 func TestListPods(t *testing.T) {

pkg/kubelet/pod_workers.go

@@ -1749,7 +1749,6 @@ func (p *podWorkers) requeueLastPodUpdate(podUID types.UID, status *podSyncStatu
 		return
 	}
 	copied := *status.activeUpdate
-	copied.UpdateType = kubetypes.SyncPodSync
 	status.pendingUpdate = &copied
 
 	// notify the pod worker

staging/src/k8s.io/kubelet/pkg/apis/pods/v1alpha1/api.pb.go

@@ -99,7 +99,8 @@ type WatchPodsEvent struct {
 	state protoimpl.MessageState `protogen:"open.v1"`
 	// ADDED, MODIFIED, DELETED, INITIAL_SYNC_COMPLETE
 	Type EventType `protobuf:"varint,1,opt,name=type,proto3,enum=v1alpha1.EventType" json:"type,omitempty"`
-	// pod is nil for INITIAL_SYNC_COMPLETE
+	// pod is nil for INITIAL_SYNC_COMPLETE.
+	// For DELETED events, only the pod name, namespace, and UID are guaranteed to be set.
 	Pod           []byte `protobuf:"bytes,2,opt,name=pod,proto3" json:"pod,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache

staging/src/k8s.io/kubelet/pkg/apis/pods/v1alpha1/api.proto

@@ -31,7 +31,8 @@ enum EventType {
 message WatchPodsEvent {
     // ADDED, MODIFIED, DELETED, INITIAL_SYNC_COMPLETE
     EventType type = 1;
-    // pod is nil for INITIAL_SYNC_COMPLETE
+    // pod is nil for INITIAL_SYNC_COMPLETE.
+    // For DELETED events, only the pod name, namespace, and UID are guaranteed to be set.
     bytes pod = 2;
 }