Repository: karmada-io/karmada

Issue:
APIServer compatibility  is failing
<!-- Please use this template while reporting a bug and provide as much info as possible. Not doing so may result in your bug not being addressed in a timely manner. Thanks!
-->


**What happened**:

https://github.com/karmada-io/karmada/actions/runs/15241477746/job/42862086548

KEP-3998 brings JobSuccessPolicy, it means that the feature-gate is enabled, JobSuccessCriteriaMet status needs to be added when the job is completed .  

this feature-gate introduced as :
- alpha in 1.30 (false by default)
- beta in 1.31 (true by default)
- GA in 1.33 (lock to true )

workflow is failing with karmada-apiserver 1.32 and not 1.31, let me check it.

**What you expected to happen**:

**How to reproduce it (as minimally and precisely as possible)**:

**Anything else we need to know?**:

**Environment**:
- Karmada version:
- kubectl-karmada or karmadactl version (the result of `kubectl-karmada version` or `karmadactl version`):
- Others:


Discussion:
@liangyuanpeng: Another feature-gate related to "job" is JobManagedBy:

```yaml
- name: JobManagedBy
  versionedSpecs:
  - default: false
    lockToDefault: false
    preRelease: Alpha
    version: "1.30"
  - default: true
    lockToDefault: false
    preRelease: Beta
    version: "1.32"
```

If set JobManagedBy to true, We will get the same error on karmada-apiserver 1.31:
```
Invalid value: cannot set Complete=True condition without the SuccessCriteriaMet=true condition"
```
@RainbowMango: > https://github.com/karmada-io/karmada/actions/runs/15241477746/job/42862086548

This test is against Kubernetes v1.30? As you mentioned the feature became enabled default since v1.31.@zhzhuang-zju: > If set JobManagedBy to true, We will get the same error on karmada-apiserver 1.31:

@liangyuanpeng Sorry, I don't understand why setting `JobManagedBy` to true triggers Invalid value error. Can you explain it in more detail? Thanks@zhzhuang-zju: https://github.com/kubernetes/kubernetes/blob/ce9ba818025ef28555bac9e25da917acec2ccb1d/pkg/registry/batch/job/strategy.go#L345-L380
```go
	if utilfeature.DefaultFeatureGate.Enabled(features.JobManagedBy) {
		// A strengthened validation of the Job status transitions is needed since the
		// Job managedBy field let's the Job object be controlled by external
		// controllers. We want to make sure the transitions done by the external
		// controllers meet the expectations of the clients of the Job API.
		// For example, we verify that a Job in terminal state (Failed or Complete)
		// does not flip to a non-terminal state.
```
The introduction of FeatureGate JobManagedBy has strengthened the validation of job status in Kubernetes. Since JobManagedBy defaults to true starting from v1.32.0, our APIServer compatibility testing began to fail at this version.@RainbowMango: I'm going to move this from v1.15 to v1.16. @RainbowMango: I can see the [job status collection testing](https://github.com/karmada-io/karmada/blob/70aa89b6df05c1f09ee77d023bbd048fe6eb69d2/test/e2e/suites/base/resource_test.go#L381-L399) in [APIServer compatibility](https://github.com/karmada-io/karmada/blob/master/.github/workflows/ci-schedule-compatibility.yaml) workflow is failing when the `karmada-apiserver` version is greater than or equal to v1.32.0, the log as follows:
```
  I1124 00:49:55.784973   56382 resource_test.go:390] job(karmadatest-lrp2q/job-f2nvc) succeedPods: 2, wanted succeedPods: 3
  I1124 00:50:00.805228   56382 resource_test.go:390] job(karmadatest-lrp2q/job-f2nvc) succeedPods: 2, wanted succeedPods: 3
  I1124 00:50:05.894340   56382 resource_test.go:390] job(karmadatest-lrp2q/job-f2nvc) succeedPods: 2, wanted succeedPods: 3
  I1124 00:50:10.897962   56382 resource_test.go:390] job(karmadatest-lrp2q/job-f2nvc) succeedPods: 2, wanted succeedPods: 3
  I1124 00:50:15.904566   56382 resource_test.go:390] job(karmadatest-lrp2q/job-f2nvc) succeedPods: 2, wanted succeedPods: 3
  << Captured StdOut/StdErr Output

  Timeline >>
  STEP: Creating Job(karmadatest-lrp2q/job-f2nvc) @ 11/24/25 00:45:20.286
  STEP: Creating PropagationPolicy(karmadatest-lrp2q/job-f2nvc) @ 11/24/25 00:45:20.293
  STEP: check whether the job status can be correctly collected @ 11/24/25 00:45:20.312
  [FAILED] in [It] - /home/runner/work/karmada/karmada/test/e2e/suites/base/resource_test.go:396 @ 11/24/25 00:50:20.313
  STEP: Removing PropagationPolicy(karmadatest-lrp2q/job-f2nvc) @ 11/24/25 00:50:20.393
  STEP: Removing Job(karmadatest-lrp2q/job-f2nvc) @ 11/24/25 00:50:20.399
  << Timeline

  [FAILED] Timed out after 300.001s.
  Expected
      <bool>: false
  to equal
      <bool>: true
  In [It] at: /home/runner/work/karmada/karmada/test/e2e/suites/base/resource_test.go:396 @ 11/24/25 00:50:20.313
```

The test times out when waiting for the federated job's expected status.

I reproduced the situation with a similar Job and PropagationPolicy:
```yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: pi
spec:
  template:
    spec:
      containers:
      - name: pi
        image: busybox:latest
        command: ["sleep",  "30s"]
      restartPolicy: Never
  backoffLimit: 4
```
```yaml
apiVersion: policy.karmada.io/v1alpha1
kind: PropagationPolicy
metadata:
  name: job
spec:
  resourceSelectors:
    - apiVersion: batch/v1
      kind: Job
      name: pi
  placement:
    clusterAffinity:
      clusterNames:
        - member1
        - member2
        - member3
```
Note: to reproduce, you have to set the `KARMADA_APISERVER_VERSION` environment to `v1.32.0` (v1.32+ will work as well) before launching the environment by `hack/local-up-karmada.sh`.

The logs of `karmada-controller-manager` explain the reason why it can't aggregate the status:
```
{"ts":1764236607602.941,"caller":"controller/controller.go:474","msg":"Reconciler error","controller":"resource-binding-status-controller","controllerGroup":"work.karmada.io","controllerKind":"ResourceBinding","ResourceBinding":{"name":"pi-job","namespace":"default"},"namespace":"default","name":"pi-job","reconcileID":"b45e20a4-2b63-4472-b2eb-e632d7cfc602","err":"Job.batch \"pi\" is invalid: status.conditions: Invalid value: cannot set Complete=True condition without the SuccessCriteriaMet=true condition"}
```
The logs say that when setting Job's condition `Complete=True`, the `SuccessCriteriaMet=true` condition must be set at the same time.

That's the reason why the `karmada-apiserver` (actually `kube-apiserver`) rejected the update request.@RainbowMango: The [validation rule](https://github.com/kubernetes/kubernetes/pull/125510/files#diff-67fdcc10f97b3bbec1288558a1575efd01474516066db5332a141592ee5dd4e4R526-R530) was added by https://github.com/kubernetes/kubernetes/pull/125510 in Kubernetes v1.31, but it only takes effect when the feature gate `JobManagedBy` is enabled. The feature [has been promoted to beta](https://github.com/kubernetes/kubernetes/blob/4bed36e03e7bd699b089d33da6f7d7c9ef9eb661/pkg/features/kube_features.go#L1405-L1409) since Kubernetes v1.32, so this issue arises when we are testing against kube-apiserver v1.32 and v1.33. @RainbowMango: /assign 
Sent https://github.com/karmada-io/karmada/pull/6964 for this and replace #6415.
@liangyuanpeng Hope you don't mind. :)

Solution patch:
diff --git a/pkg/clusterdiscovery/clusterapi/clusterapi.go b/pkg/clusterdiscovery/clusterapi/clusterapi.go
index 218409db8..76976f469 100644
--- a/pkg/clusterdiscovery/clusterapi/clusterapi.go
+++ b/pkg/clusterdiscovery/clusterapi/clusterapi.go
@@ -34,6 +34,7 @@ import (
 	secretutil "sigs.k8s.io/cluster-api/util/secret"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/karmadactl/join"
 	"github.com/karmada-io/karmada/pkg/karmadactl/options"
 	"github.com/karmada-io/karmada/pkg/karmadactl/unjoin"
@@ -62,7 +63,7 @@ type ClusterDetector struct {
 	ClusterAPIClient      client.Client
 	InformerManager       genericmanager.SingleClusterInformerManager
 	EventHandler          cache.ResourceEventHandler
-	Processor             util.AsyncWorker
+	Processor             util.AsyncPriorityWorker
 	ConcurrentReconciles  int
 }
 
@@ -72,9 +73,10 @@ func (d *ClusterDetector) Start(ctx context.Context) error {
 
 	d.EventHandler = fedinformer.NewHandlerOnEvents(d.OnAdd, d.OnUpdate, d.OnDelete)
 	workerOptions := util.Options{
-		Name:          "cluster-api cluster detector",
-		KeyFunc:       ClusterWideKeyFunc,
-		ReconcileFunc: d.Reconcile,
+		Name:             "cluster-api cluster detector",
+		KeyFunc:          ClusterWideKeyFunc,
+		ReconcileFunc:    d.Reconcile,
+		UsePriorityQueue: features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	d.Processor = util.NewAsyncWorker(workerOptions)
 	d.Processor.Run(ctx, d.ConcurrentReconciles)
@@ -97,17 +99,18 @@ func (d *ClusterDetector) discoveryCluster() {
 }
 
 // OnAdd handles object add event and push the object to queue.
-func (d *ClusterDetector) OnAdd(obj interface{}) {
+func (d *ClusterDetector) OnAdd(obj interface{}, isInInitialList bool) {
 	runtimeObj, ok := obj.(runtime.Object)
 	if !ok {
 		return
 	}
-	d.Processor.Enqueue(runtimeObj)
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
+	d.Processor.EnqueueWithOpts(util.AddOpts{Priority: priority}, runtimeObj)
 }
 
 // OnUpdate handles object update event and push the object to queue.
 func (d *ClusterDetector) OnUpdate(_, newObj interface{}) {
-	d.OnAdd(newObj)
+	d.OnAdd(newObj, false)
 }
 
 // OnDelete handles object delete event and push the object to queue.
@@ -119,7 +122,7 @@ func (d *ClusterDetector) OnDelete(obj interface{}) {
 			return
 		}
 	}
-	d.OnAdd(obj)
+	d.OnAdd(obj, false)
 }
 
 // Reconcile performs a full reconciliation for the object referred to by the key.
diff --git a/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_controller.go b/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_controller.go
index 35d5157ff..e29731f47 100644
--- a/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_controller.go
+++ b/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_controller.go
@@ -26,6 +26,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
 	"github.com/karmada-io/karmada/pkg/util"
 )
@@ -42,15 +43,16 @@ type HpaScaleTargetMarker struct {
 	DynamicClient dynamic.Interface
 	RESTMapper    meta.RESTMapper
 
-	scaleTargetWorker  util.AsyncWorker
+	scaleTargetWorker  util.AsyncPriorityWorker
 	RateLimiterOptions ratelimiterflag.Options
 }
 
 // SetupWithManager creates a controller and register to controller manager.
 func (r *HpaScaleTargetMarker) SetupWithManager(mgr controllerruntime.Manager) error {
 	scaleTargetWorkerOptions := util.Options{
-		Name:          "scale target worker",
-		ReconcileFunc: r.reconcileScaleRef,
+		Name:             "scale target worker",
+		ReconcileFunc:    r.reconcileScaleRef,
+		UsePriorityQueue: features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	r.scaleTargetWorker = util.NewAsyncWorker(scaleTargetWorkerOptions)
 	r.scaleTargetWorker.Run(context.Background(), scaleTargetWorkerNum)
diff --git a/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_predicate.go b/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_predicate.go
index 124d50447..46c6e6505 100644
--- a/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_predicate.go
+++ b/pkg/controllers/hpascaletargetmarker/hpa_scale_target_marker_predicate.go
@@ -25,6 +25,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
+	"github.com/karmada-io/karmada/pkg/util"
 )
 
 var _ predicate.Predicate = &HpaScaleTargetMarker{}
@@ -40,7 +41,8 @@ func (r *HpaScaleTargetMarker) Create(e event.CreateEvent) bool {
 
 	// if hpa exist and has been propagated, add label to its scale ref resource
 	if hasBeenPropagated(hpa) {
-		r.scaleTargetWorker.Add(labelEvent{addLabelEvent, hpa})
+		priority := util.ItemPriorityIfInInitialList(e.IsInInitialList)
+		r.scaleTargetWorker.AddWithOpts(util.AddOpts{Priority: priority}, labelEvent{addLabelEvent, hpa})
 	}
 
 	return false
diff --git a/pkg/controllers/mcs/service_export_controller.go b/pkg/controllers/mcs/service_export_controller.go
index fea9c6367..364bdfe5c 100644
--- a/pkg/controllers/mcs/service_export_controller.go
+++ b/pkg/controllers/mcs/service_export_controller.go
@@ -50,6 +50,7 @@ import (
 	clusterv1alpha1 "github.com/karmada-io/karmada/pkg/apis/cluster/v1alpha1"
 	workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
 	"github.com/karmada-io/karmada/pkg/controllers/ctrlutil"
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
 	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer"
@@ -84,7 +85,7 @@ type ServiceExportController struct {
 	// "member1": instance of ResourceEventHandler
 	eventHandlers sync.Map
 	// worker process resources periodic from rateLimitingQueue.
-	worker             util.AsyncWorker
+	worker             util.AsyncPriorityWorker
 	RateLimiterOptions ratelimiterflag.Options
 
 	// epsWorkListFunc is used to mock the work list provider to return a specific work list,
@@ -161,9 +162,10 @@ func (c *ServiceExportController) SetupWithManager(mgr controllerruntime.Manager
 // RunWorkQueue initializes worker and run it, worker will process resource asynchronously.
 func (c *ServiceExportController) RunWorkQueue() {
 	workerOptions := util.Options{
-		Name:          "service-export",
-		KeyFunc:       nil,
-		ReconcileFunc: c.syncServiceExportOrEndpointSlice,
+		Name:             "service-export",
+		KeyFunc:          nil,
+		ReconcileFunc:    c.syncServiceExportOrEndpointSlice,
+		UsePriorityQueue: features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	c.worker = util.NewAsyncWorker(workerOptions)
 	c.worker.Run(c.Context, c.WorkerNumber)
@@ -318,15 +320,16 @@ func (c *ServiceExportController) getEventHandler(clusterName string) cache.Reso
 	return eventHandler
 }
 
-func (c *ServiceExportController) genHandlerAddFunc(clusterName string) func(obj interface{}) {
-	return func(obj interface{}) {
+func (c *ServiceExportController) genHandlerAddFunc(clusterName string) func(obj interface{}, isInInitialList bool) {
+	return func(obj interface{}, isInInitialList bool) {
 		curObj := obj.(runtime.Object)
 		key, err := keys.FederatedKeyFunc(clusterName, curObj)
 		if err != nil {
 			klog.ErrorS(err, "Failed to generate key for obj", "gvk", curObj.GetObjectKind().GroupVersionKind())
 			return
 		}
-		c.worker.Add(key)
+		priority := util.ItemPriorityIfInInitialList(isInInitialList)
+		c.worker.AddWithOpts(util.AddOpts{Priority: priority}, key)
 	}
 }
 
diff --git a/pkg/controllers/multiclusterservice/endpointslice_collect_controller.go b/pkg/controllers/multiclusterservice/endpointslice_collect_controller.go
index 307cd17ba..eb735f57b 100644
--- a/pkg/controllers/multiclusterservice/endpointslice_collect_controller.go
+++ b/pkg/controllers/multiclusterservice/endpointslice_collect_controller.go
@@ -46,6 +46,7 @@ import (
 	networkingv1alpha1 "github.com/karmada-io/karmada/pkg/apis/networking/v1alpha1"
 	workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
 	"github.com/karmada-io/karmada/pkg/controllers/ctrlutil"
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
 	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer"
@@ -69,7 +70,7 @@ type EndpointSliceCollectController struct {
 	// Each handler takes the cluster name as key and takes the handler function as the value, e.g.
 	// "member1": instance of ResourceEventHandler
 	eventHandlers sync.Map
-	worker        util.AsyncWorker // worker process resources periodic from rateLimitingQueue.
+	worker        util.AsyncPriorityWorker // worker process resources periodic from rateLimitingQueue.
 
 	ClusterCacheSyncTimeout metav1.Duration
 	RateLimiterOptions      ratelimiterflag.Options
@@ -133,9 +134,10 @@ func (c *EndpointSliceCollectController) SetupWithManager(mgr controllerruntime.
 // RunWorkQueue initializes worker and run it, worker will process resource asynchronously.
 func (c *EndpointSliceCollectController) RunWorkQueue() {
 	workerOptions := util.Options{
-		Name:          "endpointslice-collect",
-		KeyFunc:       nil,
-		ReconcileFunc: c.collectEndpointSlice,
+		Name:             "endpointslice-collect",
+		KeyFunc:          nil,
+		ReconcileFunc:    c.collectEndpointSlice,
+		UsePriorityQueue: features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	c.worker = util.NewAsyncWorker(workerOptions)
 	c.worker.Run(c.Context, c.WorkerNumber)
@@ -242,15 +244,16 @@ func (c *EndpointSliceCollectController) getEventHandler(clusterName string) cac
 	return eventHandler
 }
 
-func (c *EndpointSliceCollectController) genHandlerAddFunc(clusterName string) func(obj interface{}) {
-	return func(obj interface{}) {
+func (c *EndpointSliceCollectController) genHandlerAddFunc(clusterName string) func(obj interface{}, isInInitialList bool) {
+	return func(obj interface{}, isInInitialList bool) {
 		curObj := obj.(runtime.Object)
 		key, err := keys.FederatedKeyFunc(clusterName, curObj)
 		if err != nil {
 			klog.ErrorS(err, "Failed to generate key for obj", "gvk", curObj.GetObjectKind().GroupVersionKind())
 			return
 		}
-		c.worker.Add(key)
+		priority := util.ItemPriorityIfInInitialList(isInInitialList)
+		c.worker.AddWithOpts(util.AddOpts{Priority: priority}, key)
 	}
 }
 
diff --git a/pkg/controllers/status/work_status_controller.go b/pkg/controllers/status/work_status_controller.go
index f7174cbc5..c9640a64f 100644
--- a/pkg/controllers/status/work_status_controller.go
+++ b/pkg/controllers/status/work_status_controller.go
@@ -26,6 +26,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
@@ -42,6 +43,7 @@ import (
 	workv1alpha1 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
 	"github.com/karmada-io/karmada/pkg/events"
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/metrics"
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter"
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
@@ -66,7 +68,7 @@ type WorkStatusController struct {
 	InformerManager genericmanager.MultiClusterInformerManager
 	eventHandler    cache.ResourceEventHandler // eventHandler knows how to handle events from the member cluster.
 	Context         context.Context
-	worker          util.AsyncWorker // worker process resources periodic from rateLimitingQueue.
+	worker          util.AsyncPriorityWorker // worker process resources periodic from rateLimitingQueue.
 	// ConcurrentWorkStatusSyncs is the number of Work status that are allowed to sync concurrently.
 	ConcurrentWorkStatusSyncs   int
 	ObjectWatcher               objectwatcher.ObjectWatcher
@@ -137,7 +139,7 @@ func (c *WorkStatusController) buildResourceInformers(cluster *clusterv1alpha1.C
 // getEventHandler return callback function that knows how to handle events from the member cluster.
 func (c *WorkStatusController) getEventHandler() cache.ResourceEventHandler {
 	if c.eventHandler == nil {
-		c.eventHandler = fedinformer.NewHandlerOnAllEvents(c.worker.Enqueue)
+		c.eventHandler = fedinformer.NewHandlerOnEvents(c.onAdd, c.onUpdate, c.onDelete)
 	}
 	return c.eventHandler
 }
@@ -145,9 +147,10 @@ func (c *WorkStatusController) getEventHandler() cache.ResourceEventHandler {
 // RunWorkQueue initializes worker and run it, worker will process resource asynchronously.
 func (c *WorkStatusController) RunWorkQueue() {
 	workerOptions := util.Options{
-		Name:          "work-status",
-		KeyFunc:       generateKey,
-		ReconcileFunc: c.syncWorkStatus,
+		Name:             "work-status",
+		KeyFunc:          generateKey,
+		ReconcileFunc:    c.syncWorkStatus,
+		UsePriorityQueue: features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	c.worker = util.NewAsyncWorker(workerOptions)
 	c.worker.Run(c.Context, c.ConcurrentWorkStatusSyncs)
@@ -565,3 +568,28 @@ func (c *WorkStatusController) eventf(object *unstructured.Unstructured, eventTy
 	}
 	c.EventRecorder.Eventf(ref, eventType, reason, messageFmt, args...)
 }
+
+func (c *WorkStatusController) onAdd(obj any, isInInitialList bool) {
+	curObj := obj.(runtime.Object)
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
+	c.worker.EnqueueWithOpts(util.AddOpts{Priority: priority}, curObj)
+}
+
+func (c *WorkStatusController) onUpdate(old, cur any) {
+	curObj := cur.(runtime.Object)
+	if !reflect.DeepEqual(old, cur) {
+		c.worker.Enqueue(curObj)
+	}
+}
+
+func (c *WorkStatusController) onDelete(old any) {
+	if deleted, ok := old.(cache.DeletedFinalStateUnknown); ok {
+		// This object might be stale but ok for our current usage.
+		old = deleted.Obj
+		if old == nil {
+			return
+		}
+	}
+	oldObj := old.(runtime.Object)
+	c.worker.Enqueue(oldObj)
+}
diff --git a/pkg/dependenciesdistributor/dependencies_distributor.go b/pkg/dependenciesdistributor/dependencies_distributor.go
index c4240fe7e..4014fe68f 100644
--- a/pkg/dependenciesdistributor/dependencies_distributor.go
+++ b/pkg/dependenciesdistributor/dependencies_distributor.go
@@ -51,6 +51,7 @@ import (
 	configv1alpha1 "github.com/karmada-io/karmada/pkg/apis/config/v1alpha1"
 	workv1alpha2 "github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
 	"github.com/karmada-io/karmada/pkg/events"
+	"github.com/karmada-io/karmada/pkg/features"
 	"github.com/karmada-io/karmada/pkg/resourceinterpreter"
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
 	"github.com/karmada-io/karmada/pkg/util"
@@ -102,7 +103,7 @@ type DependenciesDistributor struct {
 	RateLimiterOptions  ratelimiterflag.Options
 
 	eventHandler      cache.ResourceEventHandler
-	resourceProcessor util.AsyncWorker
+	resourceProcessor util.AsyncPriorityWorker
 	genericEvent      chan event.TypedGenericEvent[*workv1alpha2.ResourceBinding]
 	// ConcurrentDependentResourceSyncs is the number of dependent resource that are allowed to sync concurrently.
 	ConcurrentDependentResourceSyncs int
@@ -119,12 +120,13 @@ func (d *DependenciesDistributor) NeedLeaderElection() bool {
 }
 
 // OnAdd handles object add event and push the object to queue.
-func (d *DependenciesDistributor) OnAdd(obj interface{}) {
+func (d *DependenciesDistributor) OnAdd(obj interface{}, isInInitialList bool) {
 	runtimeObj, ok := obj.(runtime.Object)
 	if !ok {
 		return
 	}
-	d.resourceProcessor.Enqueue(runtimeObj)
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
+	d.resourceProcessor.EnqueueWithOpts(util.AddOpts{Priority: priority}, runtimeObj)
 }
 
 // OnUpdate handles object update event and push the object to queue.
@@ -146,9 +148,9 @@ func (d *DependenciesDistributor) OnUpdate(oldObj, newObj interface{}) {
 		return
 	}
 	if !equality.Semantic.DeepEqual(unstructuredOldObj.GetLabels(), unstructuredNewObj.GetLabels()) {
-		d.OnAdd(oldObj)
+		d.OnAdd(oldObj, false)
 	}
-	d.OnAdd(newObj)
+	d.OnAdd(newObj, false)
 }
 
 // OnDelete handles object delete event and push the object to queue.
@@ -160,7 +162,7 @@ func (d *DependenciesDistributor) OnDelete(obj interface{}) {
 			return
 		}
 	}
-	d.OnAdd(obj)
+	d.OnAdd(obj, false)
 }
 
 // reconcileResourceTemplate coordinates resources that may need to be distributed, such as Configmap, Service, etc.
@@ -663,6 +665,7 @@ func (d *DependenciesDistributor) Start(ctx context.Context) error {
 		},
 		ReconcileFunc:      d.reconcileResourceTemplate,
 		RateLimiterOptions: d.RateLimiterOptions,
+		UsePriorityQueue:   features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	d.eventHandler = fedinformer.NewHandlerOnEvents(d.OnAdd, d.OnUpdate, d.OnDelete)
 	d.resourceProcessor = util.NewAsyncWorker(resourceWorkerOptions)
diff --git a/pkg/detector/detector.go b/pkg/detector/detector.go
index fe6db79eb..22a8ea9f8 100644
--- a/pkg/detector/detector.go
+++ b/pkg/detector/detector.go
@@ -83,11 +83,11 @@ type ResourceDetector struct {
 	EventRecorder       record.EventRecorder
 	// policyReconcileWorker maintains a rate limited queue which used to store PropagationPolicy's key and
 	// a reconcile function to consume the items in queue.
-	policyReconcileWorker util.AsyncWorker
+	policyReconcileWorker util.AsyncPriorityWorker
 
 	// clusterPolicyReconcileWorker maintains a rate limited queue which used to store ClusterPropagationPolicy's key and
 	// a reconcile function to consume the items in queue.
-	clusterPolicyReconcileWorker util.AsyncWorker
+	clusterPolicyReconcileWorker util.AsyncPriorityWorker
 
 	RESTMapper meta.RESTMapper
 
@@ -119,6 +119,7 @@ func (d *ResourceDetector) Start(ctx context.Context) error {
 		KeyFunc:            NamespacedKeyFunc,
 		ReconcileFunc:      d.ReconcilePropagationPolicy,
 		RateLimiterOptions: d.RateLimiterOptions,
+		UsePriorityQueue:   features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	d.policyReconcileWorker = util.NewAsyncWorker(policyWorkerOptions)
 	d.policyReconcileWorker.Run(ctx, d.ConcurrentPropagationPolicySyncs)
@@ -127,6 +128,7 @@ func (d *ResourceDetector) Start(ctx context.Context) error {
 		KeyFunc:            NamespacedKeyFunc,
 		ReconcileFunc:      d.ReconcileClusterPropagationPolicy,
 		RateLimiterOptions: d.RateLimiterOptions,
+		UsePriorityQueue:   features.FeatureGate.Enabled(features.ControllerPriorityQueue),
 	}
 	d.clusterPolicyReconcileWorker = util.NewAsyncWorker(clusterPolicyWorkerOptions)
 	d.clusterPolicyReconcileWorker.Run(ctx, d.ConcurrentClusterPropagationPolicySyncs)
@@ -311,16 +313,12 @@ func (d *ResourceDetector) EventFilter(obj interface{}) bool {
 }
 
 // OnAdd handles object add event and push the object to queue.
-func (d *ResourceDetector) OnAdd(obj interface{}, isInitialList bool) {
+func (d *ResourceDetector) OnAdd(obj interface{}, isInInitialList bool) {
 	runtimeObj, ok := obj.(runtime.Object)
 	if !ok {
 		return
 	}
-
-	var priority *int
-	if isInitialList {
-		priority = ptr.To(util.LowPriority)
-	}
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
 	d.Processor.EnqueueWithOpts(util.AddOpts{Priority: priority}, ResourceItem{Obj: runtimeObj})
 }
 
@@ -958,8 +956,9 @@ func (d *ResourceDetector) GetMatching(resourceSelectors []policyv1alpha1.Resour
 }
 
 // OnPropagationPolicyAdd handles object add event and push the object to queue.
-func (d *ResourceDetector) OnPropagationPolicyAdd(obj interface{}) {
-	d.policyReconcileWorker.Enqueue(obj)
+func (d *ResourceDetector) OnPropagationPolicyAdd(obj interface{}, isInInitialList bool) {
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
+	d.policyReconcileWorker.EnqueueWithOpts(util.AddOpts{Priority: priority}, obj)
 }
 
 // OnPropagationPolicyUpdate handles object update event and push the object to queue.
@@ -1029,8 +1028,9 @@ func (d *ResourceDetector) ReconcilePropagationPolicy(key util.QueueKey) error {
 }
 
 // OnClusterPropagationPolicyAdd handles object add event and push the object to queue.
-func (d *ResourceDetector) OnClusterPropagationPolicyAdd(obj interface{}) {
-	d.clusterPolicyReconcileWorker.Enqueue(obj)
+func (d *ResourceDetector) OnClusterPropagationPolicyAdd(obj interface{}, isInInitialList bool) {
+	priority := util.ItemPriorityIfInInitialList(isInInitialList)
+	d.clusterPolicyReconcileWorker.EnqueueWithOpts(util.AddOpts{Priority: priority}, obj)
 }
 
 // OnClusterPropagationPolicyUpdate handles object update event and push the object to queue.
diff --git a/pkg/resourceinterpreter/customized/declarative/configmanager/manager.go b/pkg/resourceinterpreter/customized/declarative/configmanager/manager.go
index bbaca0f9e..642c8c798 100644
--- a/pkg/resourceinterpreter/customized/declarative/configmanager/manager.go
+++ b/pkg/resourceinterpreter/customized/declarative/configmanager/manager.go
@@ -83,7 +83,7 @@ func NewInterpreterConfigManager(informer genericmanager.SingleClusterInformerMa
 		manager.informer = informer
 		manager.lister = informer.Lister(util.ResourceInterpreterCustomizationsGVR)
 		configHandlers := fedinformer.NewHandlerOnEvents(
-			func(_ interface{}) { _ = manager.updateConfiguration() },
+			func(_ interface{}, _ bool) { _ = manager.updateConfiguration() },
 			func(_, _ interface{}) { _ = manager.updateConfiguration() },
 			func(_ interface{}) { _ = manager.updateConfiguration() })
 		informer.ForResource(util.ResourceInterpreterCustomizationsGVR, configHandlers)
diff --git a/pkg/resourceinterpreter/customized/webhook/configmanager/manager.go b/pkg/resourceinterpreter/customized/webhook/configmanager/manager.go
index 0c67e99ee..a9caf5179 100644
--- a/pkg/resourceinterpreter/customized/webhook/configmanager/manager.go
+++ b/pkg/resourceinterpreter/customized/webhook/configmanager/manager.go
@@ -82,7 +82,7 @@ func NewExploreConfigManager(inform genericmanager.SingleClusterInformerManager)
 
 	manager.informer = inform
 	configHandlers := fedinformer.NewHandlerOnEvents(
-		func(_ interface{}) { _ = manager.updateConfiguration() },
+		func(_ interface{}, _ bool) { _ = manager.updateConfiguration() },
 		func(_, _ interface{}) { _ = manager.updateConfiguration() },
 		func(_ interface{}) { _ = manager.updateConfiguration() })
 	inform.ForResource(util.ResourceInterpreterWebhookConfigurationsGVR, configHandlers)
diff --git a/pkg/util/fedinformer/handlers.go b/pkg/util/fedinformer/handlers.go
index b1d915b6d..7d30908a4 100644
--- a/pkg/util/fedinformer/handlers.go
+++ b/pkg/util/fedinformer/handlers.go
@@ -17,42 +17,12 @@ limitations under the License.
 package fedinformer
 
 import (
-	"reflect"
-
-	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/cache"
 )
 
-// NewHandlerOnAllEvents builds a ResourceEventHandler that the function 'fn' will be called on all events(add/update/delete).
-func NewHandlerOnAllEvents(fn func(interface{})) cache.ResourceEventHandler {
-	return &cache.ResourceEventHandlerFuncs{
-		AddFunc: func(cur interface{}) {
-			curObj := cur.(runtime.Object)
-			fn(curObj)
-		},
-		UpdateFunc: func(old, cur interface{}) {
-			curObj := cur.(runtime.Object)
-			if !reflect.DeepEqual(old, cur) {
-				fn(curObj)
-			}
-		},
-		DeleteFunc: func(old interface{}) {
-			if deleted, ok := old.(cache.DeletedFinalStateUnknown); ok {
-				// This object might be stale but ok for our current usage.
-				old = deleted.Obj
-				if old == nil {
-					return
-				}
-			}
-			oldObj := old.(runtime.Object)
-			fn(oldObj)
-		},
-	}
-}
-
 // NewHandlerOnEvents builds a ResourceEventHandler.
-func NewHandlerOnEvents(addFunc func(obj interface{}), updateFunc func(oldObj, newObj interface{}), deleteFunc func(obj interface{})) cache.ResourceEventHandler {
-	return &cache.ResourceEventHandlerFuncs{
+func NewHandlerOnEvents(addFunc func(obj interface{}, isInInitialList bool), updateFunc func(oldObj, newObj interface{}), deleteFunc func(obj interface{})) cache.ResourceEventHandler {
+	return &cache.ResourceEventHandlerDetailedFuncs{
 		AddFunc:    addFunc,
 		UpdateFunc: updateFunc,
 		DeleteFunc: deleteFunc,
@@ -65,7 +35,7 @@ func NewHandlerOnEvents(addFunc func(obj interface{}), updateFunc func(oldObj, n
 // Note: An object that starts passing the filter after an update is considered an add, and
 // an object that stops passing the filter after an update is considered a deletion.
 // Like the handlers, the filter MUST NOT modify the objects it is given.
-func NewFilteringHandlerOnAllEvents(filterFunc func(obj interface{}) bool, addFunc func(obj interface{}, isInitialList bool),
+func NewFilteringHandlerOnAllEvents(filterFunc func(obj interface{}) bool, addFunc func(obj interface{}, isInInitialList bool),
 	updateFunc func(oldObj, newObj interface{}), deleteFunc func(obj interface{})) cache.ResourceEventHandler {
 	return &cache.FilteringResourceEventHandler{
 		FilterFunc: filterFunc,
diff --git a/pkg/util/helper/job.go b/pkg/util/helper/job.go
index 4ea05e060..c6bb917a8 100644
--- a/pkg/util/helper/job.go
+++ b/pkg/util/helper/job.go
@@ -36,6 +36,8 @@ func ParsingJobStatus(obj *batchv1.Job, status []workv1alpha2.AggregatedStatusIt
 	var jobFailed []string
 	var startTime, completionTime *metav1.Time
 	successfulJobs, completionJobs := 0, 0
+	// Track how many member clusters already reported SuccessCriteriaMet=true
+	successCriteriaMetClusters := 0
 	newStatus := &batchv1.JobStatus{}
 	for _, item := range status {
 		if item.Status == nil {
@@ -60,6 +62,14 @@ func ParsingJobStatus(obj *batchv1.Job, status []workv1alpha2.AggregatedStatusIt
 			jobFailed = append(jobFailed, item.ClusterName)
 		}
 
+		// Count clusters that already set SuccessCriteriaMet=true
+		for _, c := range temp.Conditions {
+			if c.Type == batchv1.JobSuccessCriteriaMet && c.Status == corev1.ConditionTrue {
+				successCriteriaMetClusters++
+				break
+			}
+		}
+
 		// StartTime
 		if startTime == nil || temp.StartTime.Before(startTime) {
 			startTime = temp.StartTime
@@ -95,6 +105,18 @@ func ParsingJobStatus(obj *batchv1.Job, status []workv1alpha2.AggregatedStatusIt
 			Reason:             "Completed",
 			Message:            "Job completed",
 		})
+
+		// Set JobSuccessCriteriaMet condition when all member clusters have met success criteria.
+		if len(status) > 0 && successCriteriaMetClusters == len(status) {
+			newStatus.Conditions = append(newStatus.Conditions, batchv1.JobCondition{
+				Type:               batchv1.JobSuccessCriteriaMet,
+				Status:             corev1.ConditionTrue,
+				LastProbeTime:      metav1.Now(),
+				LastTransitionTime: metav1.Now(),
+				Reason:             "CompletionsReached",
+				Message:            "All member clusters have met success criteria",
+			})
+		}
 	}
 
 	if startTime != nil {
diff --git a/pkg/util/worker.go b/pkg/util/worker.go
index d7a2e0da0..b7520cafa 100644
--- a/pkg/util/worker.go
+++ b/pkg/util/worker.go
@@ -24,6 +24,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue"
 
 	"github.com/karmada-io/karmada/pkg/sharedcli/ratelimiterflag"
@@ -192,10 +193,10 @@ func (w *asyncWorker) AddWithOpts(opts AddOpts, item ...any) {
 		klog.Warningf("queue is not priority queue, fallback to normal queue, queueName: %s", w.name)
 		for _, it := range item {
 			switch {
-			case opts.After > 0:
-				w.queue.AddAfter(it, opts.After)
 			case opts.RateLimited:
 				w.queue.AddRateLimited(it)
+			case opts.After > 0:
+				w.queue.AddAfter(it, opts.After)
 			default:
 				w.queue.Add(it)
 			}
@@ -242,3 +243,11 @@ func MetaNamespaceKeyFunc(obj interface{}) (QueueKey, error) {
 	}
 	return key, nil
 }
+
+// ItemPriorityIfInInitialList returns the item's priority if it belongs to the initial list; otherwise, it returns nil.
+func ItemPriorityIfInInitialList(isInInitialList bool) *int {
+	if !isInInitialList {
+		return nil
+	}
+	return ptr.To(LowPriority)
+}



Test patch:
diff --git a/pkg/controllers/multiclusterservice/endpointslice_collect_controller_test.go b/pkg/controllers/multiclusterservice/endpointslice_collect_controller_test.go
index b8a6a3a1f..6eaa3e714 100644
--- a/pkg/controllers/multiclusterservice/endpointslice_collect_controller_test.go
+++ b/pkg/controllers/multiclusterservice/endpointslice_collect_controller_test.go
@@ -74,7 +74,7 @@ func TestGetEventHandler(t *testing.T) {
 			assert.True(t, exists, "Handler should be stored in eventHandlers")
 			assert.Equal(t, handler, storedHandler, "Stored handler should match returned handler")
 			if !tc.existingHandler {
-				assert.IsType(t, &cache.ResourceEventHandlerFuncs{}, handler, "New handler should be of type *cache.ResourceEventHandlerFuncs")
+				assert.IsType(t, &cache.ResourceEventHandlerDetailedFuncs{}, handler, "New handler should be of type *cache.ResourceEventHandlerDetailedFuncs")
 			} else {
 				assert.IsType(t, &mockResourceEventHandler{}, handler, "Existing handler should be of type *mockResourceEventHandler")
 			}
@@ -92,7 +92,7 @@ func TestGenHandlerFuncs(t *testing.T) {
 			worker: mockWorker,
 		}
 		addFunc := controller.genHandlerAddFunc(clusterName)
-		addFunc(testObj)
+		addFunc(testObj, false)
 		assert.Equal(t, 1, mockWorker.addCount, "Add function should be called once")
 	})
 
@@ -448,6 +448,15 @@ func (m *mockAsyncWorker) AddAfter(_ interface{}, _ time.Duration) {}
 
 func (m *mockAsyncWorker) Enqueue(_ interface{}) {}
 
+func (m *mockAsyncWorker) AddWithOpts(_ util.AddOpts, items ...any) {
+	for _, item := range items {
+		m.Add(item)
+	}
+}
+func (m *mockAsyncWorker) EnqueueWithOpts(_ util.AddOpts, item any) {
+	m.Enqueue(item)
+}
+
 func (m *mockAsyncWorker) Run(_ context.Context, _ int) {}
 
 type mockResourceEventHandler struct{}
diff --git a/pkg/controllers/status/work_status_controller_test.go b/pkg/controllers/status/work_status_controller_test.go
index 25f354fc2..cbd5cf420 100644
--- a/pkg/controllers/status/work_status_controller_test.go
+++ b/pkg/controllers/status/work_status_controller_test.go
@@ -36,6 +36,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/uuid"
 	dynamicfake "k8s.io/client-go/dynamic/fake"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/utils/ptr"
 	controllerruntime "sigs.k8s.io/controller-runtime"
@@ -1096,3 +1097,113 @@ func TestWorkStatusController_interpretHealth(t *testing.T) {
 		})
 	}
 }
+
+type TestObject struct {
+	metav1.TypeMeta
+	metav1.ObjectMeta
+	Spec string
+}
+
+func (in *TestObject) DeepCopyObject() runtime.Object {
+	return &TestObject{
+		TypeMeta:   in.TypeMeta,
+		ObjectMeta: in.ObjectMeta,
+		Spec:       in.Spec,
+	}
+}
+
+// mockAsyncWorker implements util.AsyncPriorityWorker for testing onAdd/onUpdate/onDelete behavior.
+type mockAsyncWorker struct {
+	receivedObjs       []any  // objects passed via EnqueueWithOpts (onAdd)
+	receivedPriorities []*int // priorities captured from EnqueueWithOpts
+	enqueuedObjs       []any  // objects passed via Enqueue (onUpdate/onDelete)
+}
+
+func (m *mockAsyncWorker) EnqueueWithOpts(opts util.AddOpts, obj any) { // capture inputs for onAdd
+	m.receivedObjs = append(m.receivedObjs, obj)
+	m.receivedPriorities = append(m.receivedPriorities, opts.Priority)
+}
+func (m *mockAsyncWorker) AddWithOpts(_ util.AddOpts, _ ...any)    {}
+func (m *mockAsyncWorker) Add(_ interface{})                       {}
+func (m *mockAsyncWorker) AddAfter(_ interface{}, _ time.Duration) {}
+func (m *mockAsyncWorker) Enqueue(obj interface{}) { // capture inputs for onUpdate/onDelete
+	m.enqueuedObjs = append(m.enqueuedObjs, obj)
+}
+func (m *mockAsyncWorker) Run(_ context.Context, _ int) {}
+
+func TestWorkStatusController_onAdd(t *testing.T) {
+	cluster := newCluster("cluster", clusterv1alpha1.ClusterConditionReady, metav1.ConditionTrue)
+	c := newWorkStatusController(cluster)
+	mockWorker := &mockAsyncWorker{}
+	c.worker = mockWorker
+
+	obj := &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj"}}
+
+	t.Run("in initial list -> low priority", func(t *testing.T) {
+		c.onAdd(obj, true)
+		assert.Equal(t, 1, len(mockWorker.receivedObjs))
+		assert.Equal(t, obj, mockWorker.receivedObjs[0])
+		if assert.NotNil(t, mockWorker.receivedPriorities[0]) {
+			assert.Equal(t, util.LowPriority, *mockWorker.receivedPriorities[0])
+		}
+	})
+
+	t.Run("not in initial list -> nil priority", func(t *testing.T) {
+		c.onAdd(obj, false)
+		assert.Equal(t, 2, len(mockWorker.receivedObjs))
+		assert.Equal(t, obj, mockWorker.receivedObjs[1])
+		assert.Nil(t, mockWorker.receivedPriorities[1])
+	})
+}
+
+func TestWorkStatusController_onUpdate(t *testing.T) {
+	cluster := newCluster("cluster", clusterv1alpha1.ClusterConditionReady, metav1.ConditionTrue)
+	c := newWorkStatusController(cluster)
+	mockWorker := &mockAsyncWorker{}
+	c.worker = mockWorker
+
+	oldObj := &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj"}, Spec: "same"}
+	// Deep copy same content
+	curSame := oldObj.DeepCopyObject().(*TestObject)
+	// Different spec
+	curDiff := &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj"}, Spec: "different"}
+
+	t.Run("objects equal -> no enqueue", func(t *testing.T) {
+		c.onUpdate(oldObj, curSame)
+		assert.Equal(t, 0, len(mockWorker.enqueuedObjs))
+	})
+
+	t.Run("objects differ -> enqueue", func(t *testing.T) {
+		c.onUpdate(oldObj, curDiff)
+		assert.Equal(t, 1, len(mockWorker.enqueuedObjs))
+		assert.Equal(t, curDiff, mockWorker.enqueuedObjs[0])
+	})
+}
+
+func TestWorkStatusController_onDelete(t *testing.T) {
+	cluster := newCluster("cluster", clusterv1alpha1.ClusterConditionReady, metav1.ConditionTrue)
+	c := newWorkStatusController(cluster)
+	mockWorker := &mockAsyncWorker{}
+	c.worker = mockWorker
+
+	obj := &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "to-delete"}}
+
+	t.Run("direct object -> enqueue", func(t *testing.T) {
+		c.onDelete(obj)
+		assert.Equal(t, 1, len(mockWorker.enqueuedObjs))
+		assert.Equal(t, obj, mockWorker.enqueuedObjs[0])
+	})
+
+	t.Run("DeletedFinalStateUnknown wrapper -> enqueue", func(t *testing.T) {
+		wrapped := cache.DeletedFinalStateUnknown{Obj: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "wrapped"}}}
+		c.onDelete(wrapped)
+		assert.Equal(t, 2, len(mockWorker.enqueuedObjs))
+		assert.Equal(t, "wrapped", mockWorker.enqueuedObjs[1].(*TestObject).Name)
+	})
+
+	t.Run("DeletedFinalStateUnknown with nil Obj -> no additional enqueue", func(t *testing.T) {
+		wrappedNil := cache.DeletedFinalStateUnknown{Obj: nil}
+		c.onDelete(wrappedNil)
+		assert.Equal(t, 2, len(mockWorker.enqueuedObjs)) // unchanged
+	})
+}
diff --git a/pkg/dependenciesdistributor/dependencies_distributor_test.go b/pkg/dependenciesdistributor/dependencies_distributor_test.go
index 6cfde4af8..0af9d210d 100644
--- a/pkg/dependenciesdistributor/dependencies_distributor_test.go
+++ b/pkg/dependenciesdistributor/dependencies_distributor_test.go
@@ -73,6 +73,18 @@ func (m *MockAsyncWorker) Enqueue(obj interface{}) {
 	m.queue = append(m.queue, obj)
 }
 
+// Note: This is a dummy implementation of AddWithOpts for testing purposes.
+func (m *MockAsyncWorker) AddWithOpts(_ util.AddOpts, items ...any) {
+	for _, item := range items {
+		m.Add(item)
+	}
+}
+
+// Note: This is a dummy implementation of EnqueueWithOpts for testing purposes.
+func (m *MockAsyncWorker) EnqueueWithOpts(_ util.AddOpts, item any) {
+	m.Enqueue(item)
+}
+
 // Note: This is a dummy implementation of Run for testing purposes.
 func (m *MockAsyncWorker) Run(ctx context.Context, workerNumber int) {
 	// No actual work is done in the mock; we just simulate running
diff --git a/pkg/detector/detector_test.go b/pkg/detector/detector_test.go
index f82d50613..bb540b62b 100644
--- a/pkg/detector/detector_test.go
+++ b/pkg/detector/detector_test.go
@@ -366,7 +366,7 @@ func TestOnAdd(t *testing.T) {
 		name            string
 		obj             interface{}
 		expectedEnqueue bool
-		isInitialList   bool
+		isInInitialList bool
 	}{
 		{
 			name: "valid unstructured object",
@@ -395,7 +395,7 @@ func TestOnAdd(t *testing.T) {
 				},
 			},
 			expectedEnqueue: true,
-			isInitialList:   true,
+			isInInitialList: true,
 		},
 		{
 			name: "invalid unstructured object",
@@ -426,7 +426,7 @@ func TestOnAdd(t *testing.T) {
 			d := &ResourceDetector{
 				Processor: mockProcessor,
 			}
-			d.OnAdd(tt.obj, tt.isInitialList)
+			d.OnAdd(tt.obj, tt.isInInitialList)
 			if tt.expectedEnqueue {
 				assert.Equal(t, 1, mockProcessor.enqueueCount, "Object should be enqueued")
 				assert.IsType(t, ResourceItem{}, mockProcessor.lastEnqueued, "Enqueued item should be of type ResourceItem")
@@ -1072,7 +1072,11 @@ func (m *mockAsyncWorker) Add(_ interface{}) {
 	m.enqueueCount++
 }
 func (m *mockAsyncWorker) AddAfter(_ interface{}, _ time.Duration) {}
-func (m *mockAsyncWorker) AddWithOpts(_ util.AddOpts, _ ...any)    {}
+func (m *mockAsyncWorker) AddWithOpts(_ util.AddOpts, items ...any) {
+	for _, item := range items {
+		m.Add(item)
+	}
+}
 func (m *mockAsyncWorker) EnqueueWithOpts(_ util.AddOpts, item any) {
 	m.Enqueue(item)
 }
diff --git a/pkg/detector/preemption_test.go b/pkg/detector/preemption_test.go
index 4830dbc97..8628538a5 100644
--- a/pkg/detector/preemption_test.go
+++ b/pkg/detector/preemption_test.go
@@ -34,6 +34,7 @@ import (
 
 	policyv1alpha1 "github.com/karmada-io/karmada/pkg/apis/policy/v1alpha1"
 	"github.com/karmada-io/karmada/pkg/apis/work/v1alpha2"
+	"github.com/karmada-io/karmada/pkg/util"
 	"github.com/karmada-io/karmada/pkg/util/fedinformer/genericmanager"
 )
 
@@ -60,6 +61,18 @@ func (m *MockAsyncWorker) Enqueue(obj interface{}) {
 	m.queue = append(m.queue, obj)
 }
 
+// Note: This is a dummy implementation of AddWithOpts for testing purposes.
+func (m *MockAsyncWorker) AddWithOpts(_ util.AddOpts, items ...any) {
+	for _, item := range items {
+		m.Add(item)
+	}
+}
+
+// Note: This is a dummy implementation of EnqueueWithOpts for testing purposes.
+func (m *MockAsyncWorker) EnqueueWithOpts(_ util.AddOpts, item any) {
+	m.Enqueue(item)
+}
+
 // Note: This is a dummy implementation of Run for testing purposes.
 func (m *MockAsyncWorker) Run(ctx context.Context, workerNumber int) {
 	// No actual work is done in the mock; we just simulate running
diff --git a/pkg/resourceinterpreter/default/native/aggregatestatus_test.go b/pkg/resourceinterpreter/default/native/aggregatestatus_test.go
index e59334ae0..1b5786339 100644
--- a/pkg/resourceinterpreter/default/native/aggregatestatus_test.go
+++ b/pkg/resourceinterpreter/default/native/aggregatestatus_test.go
@@ -232,7 +232,10 @@ func TestAggregateJobStatus(t *testing.T) {
 		"failed":         0,
 		"startTime":      startTime,
 		"completionTime": completionTime,
-		"conditions":     []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}},
+		"conditions": []batchv1.JobCondition{
+			{Type: batchv1.JobComplete, Status: corev1.ConditionTrue},
+			{Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue},
+		},
 	}
 	raw, _ := helper.BuildStatusRawExtension(statusMap)
 	aggregatedStatusItems := []workv1alpha2.AggregatedStatusItem{
@@ -259,7 +262,7 @@ func TestAggregateJobStatus(t *testing.T) {
 	}
 	newJob := &batchv1.Job{
 		TypeMeta: metav1.TypeMeta{Kind: "Job", APIVersion: batchv1.SchemeGroupVersion.String()},
-		Status:   batchv1.JobStatus{Active: 0, Succeeded: 2, Failed: 0, StartTime: &startTime, CompletionTime: &completionTime, Conditions: []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed", Message: "Job completed"}}},
+		Status:   batchv1.JobStatus{Active: 0, Succeeded: 2, Failed: 0, StartTime: &startTime, CompletionTime: &completionTime, Conditions: []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue, Reason: "Completed", Message: "Job completed"}, {Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue, Reason: "CompletionsReached", Message: "All member clusters have met success criteria"}}},
 	}
 	oldObj, _ := helper.ToUnstructured(oldJob)
 	newObj, _ := helper.ToUnstructured(newJob)
@@ -297,10 +300,12 @@ func TestAggregateJobStatus(t *testing.T) {
 	}
 
 	for _, tt := range tests {
-		actualObj, _ := aggregateJobStatus(tt.curObj, tt.aggregatedStatusItems)
-		// Clean condition time before compare, due to issue: https://github.com/karmada-io/karmada/issues/1767
-		actualObj = cleanUnstructuredJobConditionTime(actualObj)
-		assert.Equal(t, tt.expectedObj, actualObj)
+		t.Run(tt.name, func(t *testing.T) {
+			actualObj, _ := aggregateJobStatus(tt.curObj, tt.aggregatedStatusItems)
+			// Clean condition time before compare, due to issue: https://github.com/karmada-io/karmada/issues/1767
+			actualObj = cleanUnstructuredJobConditionTime(actualObj)
+			assert.Equal(t, tt.expectedObj, actualObj)
+		})
 	}
 }
 
diff --git a/pkg/util/fedinformer/handlers_test.go b/pkg/util/fedinformer/handlers_test.go
index 267705765..a2dc2879b 100644
--- a/pkg/util/fedinformer/handlers_test.go
+++ b/pkg/util/fedinformer/handlers_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package fedinformer
 
 import (
-	"reflect"
 	"testing"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -59,65 +58,6 @@ func (c *CustomResourceEventHandler) OnDelete(obj interface{}) {
 	c.handler.OnDelete(obj)
 }
 
-func TestNewHandlerOnAllEvents(t *testing.T) {
-	testCases := []struct {
-		name     string
-		event    string
-		input    interface{}
-		expected runtime.Object
-	}{
-		{
-			name:     "Add event",
-			event:    "add",
-			input:    &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-add"}, Spec: "add"},
-			expected: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-add"}, Spec: "add"},
-		},
-		{
-			name:     "Update event",
-			event:    "update",
-			input:    &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-update"}, Spec: "update"},
-			expected: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-update"}, Spec: "update"},
-		},
-		{
-			name:     "Delete event",
-			event:    "delete",
-			input:    &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-delete"}, Spec: "delete"},
-			expected: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-delete"}, Spec: "delete"},
-		},
-		{
-			name:     "Delete event with DeletedFinalStateUnknown",
-			event:    "delete",
-			input:    cache.DeletedFinalStateUnknown{Obj: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-delete-unknown"}, Spec: "delete-unknown"}},
-			expected: &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "test-obj-delete-unknown"}, Spec: "delete-unknown"},
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			var calledWith runtime.Object
-			fn := func(obj interface{}) {
-				calledWith = obj.(runtime.Object)
-			}
-
-			handler := &CustomResourceEventHandler{NewHandlerOnAllEvents(fn)}
-
-			switch tc.event {
-			case "add":
-				handler.OnAdd(tc.input, false)
-			case "update":
-				oldObj := &TestObject{ObjectMeta: metav1.ObjectMeta{Name: "old-obj"}, Spec: "old"}
-				handler.OnUpdate(oldObj, tc.input)
-			case "delete":
-				handler.OnDelete(tc.input)
-			}
-
-			if !reflect.DeepEqual(calledWith, tc.expected) {
-				t.Errorf("expected %v, got %v", tc.expected, calledWith)
-			}
-		})
-	}
-}
-
 func TestNewHandlerOnEvents(t *testing.T) {
 	testCases := []struct {
 		name  string
@@ -131,7 +71,7 @@ func TestNewHandlerOnEvents(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			var addCalled, updateCalled, deleteCalled bool
-			addFunc := func(_ interface{}) { addCalled = true }
+			addFunc := func(_ interface{}, _ bool) { addCalled = true }
 			updateFunc := func(_, _ interface{}) { updateCalled = true }
 			deleteFunc := func(_ interface{}) { deleteCalled = true }
 
diff --git a/pkg/util/helper/job_test.go b/pkg/util/helper/job_test.go
index 3203311ef..28688bc10 100644
--- a/pkg/util/helper/job_test.go
+++ b/pkg/util/helper/job_test.go
@@ -88,7 +88,7 @@ func TestParsingJobStatus(t *testing.T) {
 		"startTime":      testV1time,
 		"completionTime": testV1time,
 		"failed":         0,
-		"conditions":     []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}},
+		"conditions":     []batchv1.JobCondition{{Type: batchv1.JobComplete, Status: corev1.ConditionTrue}, {Type: batchv1.JobSuccessCriteriaMet, Status: corev1.ConditionTrue}},
 	}
 	raw, _ := BuildStatusRawExtension(statusMap)
 	statusMapWithJobfailed := map[string]interface{}{
@@ -129,6 +129,14 @@ func TestParsingJobStatus(t *testing.T) {
 						Reason:             "Completed",
 						Message:            "Job completed",
 					},
+					{
+						Type:               batchv1.JobSuccessCriteriaMet,
+						Status:             corev1.ConditionTrue,
+						LastProbeTime:      testV1time,
+						LastTransitionTime: testV1time,
+						Reason:             "CompletionsReached",
+						Message:            "All member clusters have met success criteria",
+					},
 				},
 			},
 		},