Repository: karmada-io/karmada

Issue:
StaticWeightList/TakeByWeight may reduce replicas in some clusters when total replicas increase
### Description

When distributing replicas across clusters using [TakeByWeight](https://github.com/karmada-io/karmada/blob/b0b0bb482e6a963d10850b8b72723079c83588af/pkg/util/helper/binding.go#L112), the rounding and remainder handling logic can cause temporary inconsistencies in the total number of active replicas.

Specifically, after rounding down fractional allocations (`int32` cast), the remaining replicas are redistributed sequentially by incrementing each cluster’s replica count. However, this can lead to scenarios where one cluster’s replicas are reduced compared to the previous iteration, causing the overall system to run with fewer replicas than intended until all clusters catch up.

This is particularly problematic for us because we deploy **LLM servers across multiple regions**, and **a new server can take up to \~10 minutes to become ready**. During that time, the system may run noticeably below the requested replica count, affecting availability.

### Weights

The following examples use our production weights:

* `southcentralus = 170`
* `westus = 110`
* `eastus2 = 60`
  (sum = 340)

### Example

* **Total replicas = 5**

  * Ideal (weight/sum × total):

    * `southcentralus` = 5 × 170/340 = **2.5** → 2
    * `westus`        = 5 × 110/340 = **1.6176…** → 1
    * `eastus2`       = 5 × 60/340  = **0.8823…** → 0
  * After remainder pass (current logic: +1 in order): **3, 2, 0**

* **Total replicas = 6**

  * Ideal:

    * `southcentralus` = 6 × 170/340 = **3.0** → 3
    * `westus`        = 6 × 110/340 = **1.9411…** → 1
    * `eastus2`       = 6 × 60/340  = **1.0588…** → 1
  * After remainder pass: **4, 1, 1**
  * **Issue:** Compared to the 5-replica case, `westus` drops from **2 → 1**. Given \~10 minutes startup for new LLM servers, we effectively run at only **4 active replicas** for a noticeable period.

### Expected Behavior

* When the global total increases, **no cluster’s replica count should decrease**

Discussion:
@RainbowMango: > When the global total increases, no cluster’s replica count should decrease

I totally agree. During scale-up, the replica on the existing cluster should never be reduced. 

I will try to reproduce it on my side.

@akakakakakaa by the way, do you have an idea how to fix it?
@akakakakakaa: I am not very familiar with how this kind of issue is usually solved. To explore possible approaches, I asked GPT for some ideas. One of the suggestions was to switch from the current rounding logic to a divisor method (e.g., D’Hondt or Webster). After reviewing the idea, it seems like a reasonable solution.

---
## Divisor Method

A divisor method assigns replicas incrementally, always giving the “next replica” to the cluster with the highest priority score.

* **D’Hondt (Jefferson):** favors larger weights slightly.
* **Webster (Sainte-Laguë):** more neutral / proportional.
* Both guarantee that when total replicas increase, **no cluster’s allocation decreases**.
* Because this approach is implemented with a **priority queue**, replicas in each cluster grow **linearly with total replicas**, and allocations remain consistent:

  * if two clusters have the same weight, the one with fewer replicas will always be prioritized until they balance.
---
```go
package main

import (
	"container/heap"
	"fmt"
)

// ==== Data structures ====

// Cluster allocation target
type TargetCluster struct {
	Name     string
	Replicas int32
}

// Item used in the priority queue
type Item struct {
	name   string
	weight float64
	alloc  int
	method string
}

// Score calculation (priority)
func (it Item) score() float64 {
	switch it.method {
	case "webster": // Webster (Sainte-Laguë): divisors 1,3,5,...
		return it.weight / float64(2*it.alloc+1)
	default: // D'Hondt (Jefferson): divisors 1,2,3,...
		return it.weight / float64(it.alloc+1)
	}
}

// PriorityQueue implements heap.Interface
type PriorityQueue []Item

func (pq PriorityQueue) Len() int { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool {
	// max-heap by score
	si, sj := pq[i].score(), pq[j].score()
	if si != sj {
		return si > sj
	}
	// tie-break by weight, then name
	if pq[i].weight != pq[j].weight {
		return pq[i].weight > pq[j].weight
	}
	return pq[i].name < pq[j].name
}
func (pq PriorityQueue) Swap(i, j int) { pq[i], pq[j] = pq[j], pq[i] }

func (pq *PriorityQueue) Push(x interface{}) {
	*pq = append(*pq, x.(Item))
}
func (pq *PriorityQueue) Pop() interface{} {
	old := *pq
	n := len(old)
	item := old[n-1]
	*pq = old[0 : n-1]
	return item
}

// ==== Divisor allocation ====

// TakeByWeightDivisor allocates replicas using D’Hondt or Webster method.
func TakeByWeightDivisor(weights map[string]int, total int, method string) []TargetCluster {
	// Build priority queue
	h := &PriorityQueue{}
	for name, w := range weights {
		*h = append(*h, Item{name: name, weight: float64(w), alloc: 0, method: method})
	}
	heap.Init(h)

	// Assign replicas one by one
	for i := 0; i < total; i++ {
		it := heap.Pop(h).(Item)
		it.alloc++
		heap.Push(h, it)
	}

	// Collect result
	result := make([]TargetCluster, 0, len(*h))
	for h.Len() > 0 {
		it := heap.Pop(h).(Item)
		result = append(result, TargetCluster{
			Name:     it.name,
			Replicas: int32(it.alloc),
		})
	}
	return result
}
```
@RainbowMango: Interesting, I will put this in my queue and explore it further.

@akakakakakaa are you using Karmada now? Or in the phase of evaluation? I'm interested in the use case of running inferencing workloads with Karmada.@akakakakakaa: A couple of months ago we ran into GPU supply issues, which pushed us to move from a single-cluster setup to a multi-region one. Our main requirements were:

1. It should run on Kubernetes for operational ease.
2. Traffic should be easily distributed across regions.
3. Because GPU quotas differ by region, we needed to deploy with different weights per region.

After evaluating different options, we found Karmada to be the best fit and adopted it. We’ve been running it in production for about two months now without major issues. Combined with Istio multicluster, it works very well, even if pods are deployed unevenly across regions, traffic naturally shifts to match capacity.@RainbowMango: That's awesome! I'm glad to hear that. 

Since you are already running Karmada in production, we recommend joining [the Karmada Adopter Group](https://github.com/karmada-io/community/tree/main/adopter-group). This way, you'll receive important updates(like critical bugfixes) and stay connected with the community.

Joining is easy ---  just leave a comment on the https://github.com/karmada-io/karmada/issues/4540, then the community member will help you get onboard. @RainbowMango: I reproduced this on my side: 

**1: Create a Deployment with `5` replicas as follows:**
```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: nginx
  labels:
    app: nginx
spec:
  replicas: 5
  selector:
    matchLabels:
      app: nginx
  template:
    metadata:
      labels:
        app: nginx
    spec:
      containers:
      - image: nginx
        name: nginx
```

**2: Create a PropagationPolicy with cluster weight `170:110:60` (member1:member2:member3):**
```yaml
apiVersion: policy.karmada.io/v1alpha1
kind: PropagationPolicy
metadata:
  name: nginx-propagation
spec:
  resourceSelectors:
    - apiVersion: apps/v1
      kind: Deployment
      name: nginx
  placement:
    clusterAffinity:
      clusterNames:
        - member1
        - member2
        - member3
    replicaScheduling:
      replicaDivisionPreference: Weighted
      replicaSchedulingType: Divided
      weightPreference:
        staticWeightList:
          - targetCluster:
              clusterNames:
                - member1
            weight: 170
          - targetCluster:
              clusterNames:
                - member2
            weight: 110
          - targetCluster:
              clusterNames:
                - member3
            weight: 60
    spreadConstraints:
      - spreadByField: cluster
        maxGroups: 3
        minGroups: 3
```

**3: Check schedule result: `member1: 3`, `member2:2`**
```
I0909 09:01:51.748566       1 generic_scheduler.go:108] Assigned Replicas: [{member1 3} {member2 2}]
```

**4: Scale deployment replicas from `5` to `6`**
```
-bash-5.0# kubectl scale deployment nginx --replicas 6
deployment.apps/nginx scaled
```

**5: Check the schedule result: `member1: 4`, `member2:1`, `member3: 1`**
```
I0909 09:07:06.500411       1 generic_scheduler.go:108] Assigned Replicas: [{member1 4} {member2 1} {member3 1}]
```

This test shows the replicas in `member2` reduced from `2` to `1` during the scale operation. This is the unexpected behavior. @RainbowMango: @akakakakakaa Now I can confirm that the root cause is exactly as you mentioned, the [TakeByWeight](https://github.com/karmada-io/karmada/blob/b0b0bb482e6a963d10850b8b72723079c83588af/pkg/util/helper/binding.go#L112) uses an unstable algorithm. 
I will continue to investigate how to fix it. (will look at the solution you provided above)@RainbowMango: @akakakakakaa The `D’Hondt` algorithm you provided above looks pretty good, but I need to explore it further, as [another scenario](https://github.com/karmada-io/karmada/issues/6424) requires each cluster to have at least one replica assigned. 

By the way, what do you think of [the adopter thing](https://github.com/karmada-io/karmada/issues/6735#issuecomment-3264943038) I mentioned above?@FYLSunghwan: Hi @RainbowMango,
I’m Sunghwan Kim, an ML Engineer working together with @akakakakakaa.

Our team is positive about joining the Adopter Group, and as you suggested, we’ve already created an [issue comment](https://github.com/karmada-io/karmada/issues/4540#issuecomment-3279110408) to apply.
We look forward to work with the Karmada community to address challenges in serving LLMs across multi-cluster environments :)@RainbowMango: Awesome! Thank you @FYLSunghwan and @akakakakakaa. We'd love to cooperate with every user. I believe this issue would be a good start.

I just created https://github.com/karmada-io/community/issues/160 to get you onboard.@RainbowMango: A quick update on this issue: I've been working on it over the past few days. I have almost finished the new algorithm, but given that this feature is widely used, I just sent #6757 to build the test base, so that we can clearly identify the effect of adopting the new algorithm.@RainbowMango: Another quick update:

I am trying to adopt [the Webster method](https://en.wikipedia.org/wiki/Sainte-Lagu%C3%AB_method) to refactor the [TakeByWeight](https://github.com/karmada-io/karmada/blob/b0b0bb482e6a963d10850b8b72723079c83588af/pkg/util/helper/binding.go#L112), it breaks quite a lot of unit tests. I've fixed most of them. But some of the tests were introduced by [another feature](https://github.com/karmada-io/karmada/issues/4220), which needs more time to evaluate the effect.

Here is the code snapshot: https://github.com/RainbowMango/karmada/commit/f87985d892b74737770b20107428a7f0cd68b5e2
And failing tests are:
```
-bash-5.0# go test ./pkg/scheduler/core/... -run=Test_EvenDistributionOfReplicas
I0918 17:19:53.939512 3249861 defaults.go:54] default Karmada Image Version: v0.0.0-master
--- FAIL: Test_EvenDistributionOfReplicas (0.00s)
    --- FAIL: Test_EvenDistributionOfReplicas/replica_7,_static_weighted_2:1:1:1,_change_replicas_from_7_to_8,_after_change (0.00s)
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:2:1:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:1:2:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:1:1:2]
    --- FAIL: Test_EvenDistributionOfReplicas/replica_9,_static_weighted_2:1:1:1,_change_replicas_from_9_to_8,_before_change (0.00s)
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 2} {member4 2}], wants 3:2:2:2 --> [4:2:2:1 4:1:2:2 4:2:1:2]
    --- FAIL: Test_EvenDistributionOfReplicas/replica_9,_static_weighted_2:1:1:1,_change_replicas_from_9_to_8,_after_change (0.00s)
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:1:1:2 4:1:2:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:1:1:2 4:2:1:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 3} {member2 2} {member3 1} {member4 2}], wants 3:2:1:2 --> [4:2:1:1 4:1:2:1]
    --- FAIL: Test_EvenDistributionOfReplicas/replica_5,_static_weighted_1:1:1,_add_a_new_cluster_and_change_static_weight_to_1:1:1:1,_after_change (0.00s)
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 1} {member2 1} {member3 1} {member4 2}], wants 1:1:1:2 --> [1:2:1:1 2:1:1:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 1} {member2 1} {member3 1} {member4 2}], wants 1:1:1:2 --> [1:1:2:1 2:1:1:1]
        generic_scheduler_test.go:527: AssignReplicas() got = [{member1 1} {member2 1} {member3 1} {member4 2}], wants 1:1:1:2 --> [1:1:2:1 1:2:1:1]
FAIL
FAIL	github.com/karmada-io/karmada/pkg/scheduler/core	0.023s
```
@RainbowMango: @akakakakakaa I've sent another PR #6772 to get the priority queue onboard. And after that, the [TakeByWeight](https://github.com/karmada-io/karmada/blob/b0b0bb482e6a963d10850b8b72723079c83588af/pkg/util/helper/binding.go#L112) will be refactored by follow-up PRs iteratively.  To ensure the changes won't break anything else, just fix the bug. All the tasks are tracked by #6739.  (By the way, if you want to join the iteration, just leave a comment and claim which task you'd like to take.)@RainbowMango: @akakakakakaa @FYLSunghwan 
This issue has been resolved by #6793. And the patch is included in [the Karmada v1.15.1 release](https://github.com/karmada-io/karmada/releases/tag/v1.15.1), which was available on Sep 30, 2025.  Please feel free to let us know if you need any help, like upgrading your Karmada deployment. 

Solution patch:
diff --git a/pkg/karmadactl/explain/explain.go b/pkg/karmadactl/explain/explain.go
index b5f6ae7a7..7b30e2649 100644
--- a/pkg/karmadactl/explain/explain.go
+++ b/pkg/karmadactl/explain/explain.go
@@ -31,15 +31,14 @@ import (
 )
 
 var (
-	explainLong = templates.LongDesc(`
-		Describe fields and structure of various resources in Karmada control plane or a member cluster.
-
-		This command describes the fields associated with each supported API resource.
-		Fields are identified via a simple JSONPath identifier:
-
-			<type>.<fieldName>[.<fieldName>]
-
-		Information about each field is retrieved from the server in OpenAPI format.`)
+	explainLong = templates.LongDesc("Describe fields and structure of various resources in Karmada control plane or a member cluster.\n" +
+		"\n" +
+		"This command describes the fields associated with each supported API resource.\n" +
+		"Fields are identified via a simple JSONPath identifier:\n" +
+		"\n" +
+		"    `<type>.<fieldName>[.<fieldName>]`\n" +
+		"\n" +
+		"Information about each field is retrieved from the server in OpenAPI format.")
 
 	explainExamples = templates.Examples(`
 		# Get the documentation of the resource and its fields in Karmada control plane
diff --git a/pkg/util/helper/webstermethod.go b/pkg/util/helper/webstermethod.go
new file mode 100644
index 000000000..976f55ccd
--- /dev/null
+++ b/pkg/util/helper/webstermethod.go
@@ -0,0 +1,161 @@
+/*
+Copyright 2025 The Karmada 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 helper
+
+import (
+	"container/heap"
+	"sort"
+)
+
+// Party represents a party in the Webster apportionment method.
+type Party struct {
+	// Name is the name of the party.
+	Name string
+	// Votes is the number of votes the party received.
+	Votes int64
+	// Seats is the number of seats assigned to the party.
+	Seats int32
+}
+
+// WebsterPriorityQueue implements heap.Interface using the Webster (Sainte-Laguë) method.
+// The party with the highest Webster priority (Votes/(2*Seats+1)) is at the top.
+type WebsterPriorityQueue struct {
+	// Parties contains all the parties participating in the Webster apportionment method.
+	// Each party has a name, vote count, and current seat allocation.
+	Parties []Party
+
+	// TieBreaker is an optional function used to break ties when two parties have
+	// the same Webster priority. If nil, ties are broken by vote count, then seat count,
+	// then party name to ensure a deterministic result.
+	TieBreaker func(a Party, b Party) bool
+}
+
+// Check if our WebsterPriorityQueue implements necessary interfaces
+var _ heap.Interface = &WebsterPriorityQueue{}
+
+// Len returns the number of Parties in the queue.
+func (pq *WebsterPriorityQueue) Len() int {
+	return len(pq.Parties)
+}
+
+// Less compares two Parties by their Webster priority.
+// The party with the higher priority comes first.
+func (pq *WebsterPriorityQueue) Less(i, j int) bool {
+	// In the Webster method, compare the priority of two Parties:
+	// the one with the higher value of Votes/(2*Seats+1) gets the next seat.
+	iPriority := float64(pq.Parties[i].Votes) / (float64(2*pq.Parties[i].Seats + 1))
+	jPriority := float64(pq.Parties[j].Votes) / (float64(2*pq.Parties[j].Seats + 1))
+	if iPriority == jPriority {
+		if pq.TieBreaker != nil {
+			return pq.TieBreaker(pq.Parties[i], pq.Parties[j])
+		}
+
+		// The party with fewer seats wins the tie.
+		// This is because the Webster method inherently favors smaller parties (or, in Karmada's context, smaller clusters).
+		// In Karmada, this method is often used to assign replicas based on static or dynamic weights.
+		// The rationale is that assigning to smaller clusters earlier helps establish High Availability (HA) and enables
+		// Horizontal Pod Autoscaling (HPA) more effectively. By preferring parties (clusters) with fewer seats (replicas),
+		// we promote a more balanced and resilient distribution, which is desirable for HA scenarios.
+		if pq.Parties[i].Seats != pq.Parties[j].Seats {
+			return pq.Parties[i].Seats < pq.Parties[j].Seats
+		}
+
+		// If both parties have the same priority and the same number of seats,
+		// that means they also have the same number of votes. So we don't have to compare the votes now.
+		// The party with the lexicographically smaller name wins the tie.
+		// For example, party 'a' wins over party 'b' because 'a' < 'b' lexicographically.
+		return pq.Parties[i].Name < pq.Parties[j].Name
+	}
+
+	return iPriority > jPriority
+}
+
+// Swap swaps two Parties in the queue.
+func (pq *WebsterPriorityQueue) Swap(i, j int) {
+	pq.Parties[i], pq.Parties[j] = pq.Parties[j], pq.Parties[i]
+}
+
+// Push adds a new party to the queue.
+func (pq *WebsterPriorityQueue) Push(x interface{}) {
+	pq.Parties = append(pq.Parties, x.(Party))
+}
+
+// Pop removes and returns the party with the highest priority.
+func (pq *WebsterPriorityQueue) Pop() interface{} {
+	old := pq.Parties
+	n := len(old)
+	item := old[n-1]
+	pq.Parties = old[0 : n-1]
+	return item
+}
+
+// AllocateWebsterSeats allocates new seats using the Webster method.
+// newSeats: number of new seats to allocate across all Parties.
+// partyVotes: map of party name to number of votes.
+// initialAssignments: map of party name to initial seat assignments.
+// tieBreaker: function to break ties between two Parties.
+// Returns: slice of Party with updated Seats, sorted by Name in ascending order.
+func AllocateWebsterSeats(newSeats int32, partyVotes map[string]int64, initialAssignments map[string]int32, tieBreaker func(a, b Party) bool) []Party {
+	partiesMap := make(map[string]Party, len(partyVotes)+len(initialAssignments))
+
+	// set initial party assignments, the votes default to 0
+	// If a party is present in initialAssignments but not in partyVotes,
+	// it will be initialized with 0 votes and its initial seat count.
+	// Such a party will still be included in the allocation process, but
+	// the priority of such a party is 0, which is the lowest priority.
+	for n, s := range initialAssignments {
+		partiesMap[n] = Party{Name: n, Votes: 0, Seats: s}
+	}
+
+	for n, v := range partyVotes {
+		if p, ok := partiesMap[n]; ok {
+			p.Votes = v
+			partiesMap[n] = p
+		} else {
+			partiesMap[n] = Party{Name: n, Votes: v, Seats: 0}
+		}
+	}
+
+	pq := WebsterPriorityQueue{
+		Parties:    make([]Party, 0, len(partiesMap)),
+		TieBreaker: tieBreaker,
+	}
+
+	for _, party := range partiesMap {
+		pq.Parties = append(pq.Parties, party)
+	}
+
+	if len(pq.Parties) == 0 {
+		return nil
+	}
+
+	heap.Init(&pq)
+
+	for remaining := newSeats; remaining > 0; remaining-- {
+		nextParty := heap.Pop(&pq).(Party)
+		nextParty.Seats++
+		heap.Push(&pq, nextParty)
+	}
+
+	// sort the parties by name in ascending order, to ensure the result is deterministic.
+	// This will break the heap structure, but it is ok because we will not use the heap structure again.
+	sort.Slice(pq.Parties, func(i, j int) bool {
+		return pq.Parties[i].Name < pq.Parties[j].Name
+	})
+
+	return pq.Parties
+}



Test patch:
diff --git a/pkg/util/helper/webstermethod_test.go b/pkg/util/helper/webstermethod_test.go
new file mode 100644
index 000000000..a588a16c2
--- /dev/null
+++ b/pkg/util/helper/webstermethod_test.go
@@ -0,0 +1,313 @@
+/*
+Copyright 2025 The Karmada 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 helper
+
+import (
+	"fmt"
+	"reflect"
+	"testing"
+)
+
+func ExampleAllocateWebsterSeats() {
+	partiesVotes := map[string]int64{
+		"Alpha": 1200,
+		"Beta":  900,
+		"Gamma": 400,
+	}
+	totalSeats := int32(4)
+	result := AllocateWebsterSeats(totalSeats, partiesVotes, nil, nil)
+	for _, p := range result {
+		fmt.Printf("%s: %d seats\n", p.Name, p.Seats)
+	}
+	// Output:
+	// Alpha: 2 seats
+	// Beta: 1 seats
+	// Gamma: 1 seats
+}
+
+func TestAllocateWebsterSeats(t *testing.T) {
+	// This is used for test the classic example from https://en.wikipedia.org/wiki/Sainte-Lagu%C3%AB_method,
+	// where 230,000 voters allocate 8 seats among 4 Parties.
+	// The expected seat distribution after each round matches the step-by-step allocation shown in the Wikipedia.
+	partiesVotes := map[string]int64{
+		"PartyA": 100000,
+		"PartyB": 80000,
+		"PartyC": 30000,
+		"PartyD": 20000,
+	}
+
+	tests := []struct {
+		name               string
+		newSeats           int32
+		partyVotes         map[string]int64
+		initialAssignments map[string]int32
+		tieBreaker         func(a, b Party) bool
+		expected           []Party
+	}{
+		{
+			name:               "classic example, round 1",
+			newSeats:           1,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 1},
+				{Name: "PartyB", Votes: 80000, Seats: 0},
+				{Name: "PartyC", Votes: 30000, Seats: 0},
+				{Name: "PartyD", Votes: 20000, Seats: 0},
+			},
+		},
+		{
+			name:               "classic example, round 2",
+			newSeats:           2,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 1},
+				{Name: "PartyB", Votes: 80000, Seats: 1},
+				{Name: "PartyC", Votes: 30000, Seats: 0},
+				{Name: "PartyD", Votes: 20000, Seats: 0},
+			},
+		},
+		{
+			name:               "classic example, round 3",
+			newSeats:           3,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 2},
+				{Name: "PartyB", Votes: 80000, Seats: 1},
+				{Name: "PartyC", Votes: 30000, Seats: 0},
+				{Name: "PartyD", Votes: 20000, Seats: 0},
+			},
+		},
+		{
+			name:               "classic example, round 4",
+			newSeats:           4,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 2},
+				{Name: "PartyB", Votes: 80000, Seats: 1},
+				{Name: "PartyC", Votes: 30000, Seats: 1},
+				{Name: "PartyD", Votes: 20000, Seats: 0},
+			},
+		},
+		{
+			name:               "classic example, round 5",
+			newSeats:           5,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 2},
+				{Name: "PartyB", Votes: 80000, Seats: 2},
+				{Name: "PartyC", Votes: 30000, Seats: 1},
+				{Name: "PartyD", Votes: 20000, Seats: 0},
+			},
+		},
+		{
+			name:               "classic example, round 6",
+			newSeats:           6,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 2},
+				{Name: "PartyB", Votes: 80000, Seats: 2},
+				{Name: "PartyC", Votes: 30000, Seats: 1},
+				{Name: "PartyD", Votes: 20000, Seats: 1},
+			},
+		},
+		{
+			name:               "classic example, round 7",
+			newSeats:           7,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 3},
+				{Name: "PartyB", Votes: 80000, Seats: 2},
+				{Name: "PartyC", Votes: 30000, Seats: 1},
+				{Name: "PartyD", Votes: 20000, Seats: 1},
+			},
+		},
+		{
+			name:               "classic example, round 8",
+			newSeats:           8,
+			partyVotes:         partiesVotes,
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100000, Seats: 3},
+				{Name: "PartyB", Votes: 80000, Seats: 3},
+				{Name: "PartyC", Votes: 30000, Seats: 1},
+				{Name: "PartyD", Votes: 20000, Seats: 1},
+			},
+		},
+		{
+			name:               "empty party votes, expect nil result",
+			newSeats:           1,
+			partyVotes:         map[string]int64{},
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected:           nil,
+		},
+		{
+			name:               "party votes not empty, newSeats is 0, expect all seats 0",
+			newSeats:           0,
+			partyVotes:         map[string]int64{"PartyA": 100, "PartyB": 200},
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100, Seats: 0},
+				{Name: "PartyB", Votes: 200, Seats: 0},
+			},
+		},
+		{
+			name:               "both party votes and newSeats are 0, expect nil result",
+			newSeats:           0,
+			partyVotes:         map[string]int64{},
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected:           nil,
+		},
+		{
+			name:     "tie-breaker is nil, expect break tie by seats",
+			newSeats: 1,
+			partyVotes: map[string]int64{
+				"PartyA": 3,
+				"PartyB": 1,
+			},
+			initialAssignments: map[string]int32{
+				"PartyA": 1,
+				"PartyB": 0,
+			},
+			tieBreaker: nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 3, Seats: 1},
+				{Name: "PartyB", Votes: 1, Seats: 1},
+			},
+		},
+		{
+			name:     "tie-breaker is nil, expect break tie by name",
+			newSeats: 1,
+			partyVotes: map[string]int64{
+				"PartyA": 1,
+				"PartyB": 1,
+			},
+			initialAssignments: nil,
+			tieBreaker:         nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 1, Seats: 1},
+				{Name: "PartyB", Votes: 1, Seats: 0},
+			},
+		},
+		{
+			name:     "custom tie-breaker returns true, expect custom tie-breaker controls seat",
+			newSeats: 1,
+			partyVotes: map[string]int64{
+				"PartyA": 1,
+				"PartyB": 1,
+			},
+			initialAssignments: nil,
+			tieBreaker: func(a, b Party) bool {
+				return a.Name < b.Name
+			},
+			expected: []Party{
+				{Name: "PartyA", Votes: 1, Seats: 1},
+				{Name: "PartyB", Votes: 1, Seats: 0},
+			},
+		},
+		{
+			name:     "custom tie-breaker returns false, expect custom tie-breaker controls seat",
+			newSeats: 1,
+			partyVotes: map[string]int64{
+				"PartyA": 1,
+				"PartyB": 1,
+			},
+			initialAssignments: nil,
+			tieBreaker: func(a, b Party) bool {
+				return a.Name > b.Name
+			},
+			expected: []Party{
+				{Name: "PartyA", Votes: 1, Seats: 0},
+				{Name: "PartyB", Votes: 1, Seats: 1},
+			},
+		},
+		{
+			name:     "non-initial allocation, new party joins, only new party gets new seats",
+			newSeats: 2,
+			partyVotes: map[string]int64{
+				"NewParty1": 1,
+				"NewParty2": 1,
+			},
+			initialAssignments: map[string]int32{
+				"OldParty": 0,
+			},
+			tieBreaker: nil,
+			expected: []Party{
+				{Name: "NewParty1", Votes: 1, Seats: 1},
+				{Name: "NewParty2", Votes: 1, Seats: 1},
+				{Name: "OldParty", Votes: 0, Seats: 0},
+			},
+		},
+		{
+			name:     "non-initial allocation, both new and old parties compete for new seats",
+			newSeats: 2,
+			partyVotes: map[string]int64{
+				"OldParty": 1,
+				"NewParty": 1,
+			},
+			initialAssignments: map[string]int32{
+				"OldParty": 1,
+			},
+			tieBreaker: nil,
+			expected: []Party{
+				{Name: "NewParty", Votes: 1, Seats: 2},
+				{Name: "OldParty", Votes: 1, Seats: 1},
+			},
+		},
+		{
+			name:     "non-initial allocation, initialAssignments party not in partyVotes, expect no new seats for that party",
+			newSeats: 2,
+			partyVotes: map[string]int64{
+				"PartyA": 100,
+			},
+			initialAssignments: map[string]int32{
+				"PartyB": 3,
+			},
+			tieBreaker: nil,
+			expected: []Party{
+				{Name: "PartyA", Votes: 100, Seats: 2},
+				{Name: "PartyB", Votes: 0, Seats: 3},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			got := AllocateWebsterSeats(test.newSeats, test.partyVotes, test.initialAssignments, test.tieBreaker)
+			if !reflect.DeepEqual(got, test.expected) {
+				t.Errorf("expected %v, got %v", test.expected, got)
+			}
+		})
+	}
+}
diff --git a/test/e2e/suites/base/failover_test.go b/test/e2e/suites/base/failover_test.go
index 8b2975c8e..201eb3c1d 100644
--- a/test/e2e/suites/base/failover_test.go
+++ b/test/e2e/suites/base/failover_test.go
@@ -40,99 +40,89 @@ import (
 	testhelper "github.com/karmada-io/karmada/test/helper"
 )
 
-// failover testing is used to test the rescheduling situation when some initially scheduled clusters fail
-var _ = framework.SerialDescribe("failover testing", func() {
-	ginkgo.Context("Deployment propagation testing", func() {
-		var policyNamespace, policyName string
-		var deploymentNamespace, deploymentName string
-		var deployment *appsv1.Deployment
-		var maxGroups, minGroups, numOfFailedClusters int
-		var policy *policyv1alpha1.PropagationPolicy
+// cluster failover testing is used to test the rescheduling situation when some initially scheduled clusters fail
+var _ = framework.SerialDescribe("cluster failover testing", func() {
+	var policyNamespace, policyName string
+	var deploymentNamespace, deploymentName string
+	var deployment *appsv1.Deployment
+	var maxGroups, minGroups int
+	var policy *policyv1alpha1.PropagationPolicy
+
+	ginkgo.BeforeEach(func() {
+		policyNamespace = testNamespace
+		policyName = deploymentNamePrefix + rand.String(RandomStrLength)
+		deploymentNamespace = testNamespace
+		deploymentName = policyName
+		deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
+		maxGroups = 1
+		minGroups = 1
 
-		ginkgo.BeforeEach(func() {
-			policyNamespace = testNamespace
-			policyName = deploymentNamePrefix + rand.String(RandomStrLength)
-			deploymentNamespace = testNamespace
-			deploymentName = policyName
-			deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
-			maxGroups = 1
-			minGroups = 1
-			numOfFailedClusters = 1
-
-			// set MaxGroups=MinGroups=1, label is location=CHN.
-			policy = testhelper.NewPropagationPolicy(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
+		policy = testhelper.NewPropagationPolicy(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
+			{
+				APIVersion: deployment.APIVersion,
+				Kind:       deployment.Kind,
+				Name:       deployment.Name,
+			},
+		}, policyv1alpha1.Placement{
+			ClusterAffinity: &policyv1alpha1.ClusterAffinity{
+				ClusterNames: framework.ClusterNames()[:2],
+			},
+			ClusterTolerations: []corev1.Toleration{
 				{
-					APIVersion: deployment.APIVersion,
-					Kind:       deployment.Kind,
-					Name:       deployment.Name,
+					Key:               framework.TaintClusterNotReady,
+					Operator:          corev1.TolerationOpExists,
+					Effect:            corev1.TaintEffectNoExecute,
+					TolerationSeconds: ptr.To[int64](2),
 				},
-			}, policyv1alpha1.Placement{
-				ClusterAffinity: &policyv1alpha1.ClusterAffinity{
-					LabelSelector: &metav1.LabelSelector{
-						// only test push mode clusters
-						// because pull mode clusters cannot be disabled by changing APIEndpoint
-						MatchLabels: pushModeClusterLabels,
-					},
-				},
-				ClusterTolerations: []corev1.Toleration{
-					{
-						Key:               framework.TaintClusterNotReady,
-						Operator:          corev1.TolerationOpExists,
-						Effect:            corev1.TaintEffectNoExecute,
-						TolerationSeconds: ptr.To[int64](2),
-					},
-				},
-				SpreadConstraints: []policyv1alpha1.SpreadConstraint{
-					{
-						SpreadByField: policyv1alpha1.SpreadByFieldCluster,
-						MaxGroups:     maxGroups,
-						MinGroups:     minGroups,
-					},
+			},
+			SpreadConstraints: []policyv1alpha1.SpreadConstraint{
+				{
+					SpreadByField: policyv1alpha1.SpreadByFieldCluster,
+					MaxGroups:     maxGroups,
+					MinGroups:     minGroups,
 				},
-			})
+			},
 		})
+	})
 
-		ginkgo.BeforeEach(func() {
-			framework.CreatePropagationPolicy(karmadaClient, policy)
-			framework.CreateDeployment(kubeClient, deployment)
-			ginkgo.DeferCleanup(func() {
-				framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
-				framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
-			})
+	ginkgo.JustBeforeEach(func() {
+		framework.CreatePropagationPolicy(karmadaClient, policy)
+		framework.CreateDeployment(kubeClient, deployment)
+		ginkgo.DeferCleanup(func() {
+			framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
+			framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
 		})
+	})
 
+	ginkgo.Context("Cluster failover testing with nil failover behavior", func() {
 		ginkgo.It("deployment failover testing", func() {
 			var disabledClusters []string
 			targetClusterNames := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
 
-			ginkgo.By(fmt.Sprintf("add taint %v to the random one cluster", framework.NotReadyTaintTemplate), func() {
-				temp := numOfFailedClusters
+			ginkgo.By(fmt.Sprintf("add taint %v to the target clusters", framework.NotReadyTaintTemplate), func() {
 				for _, targetClusterName := range targetClusterNames {
-					if temp > 0 {
-						err := framework.AddClusterTaint(controlPlaneClient, targetClusterName, *framework.NotReadyTaintTemplate)
-						gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+					err := framework.AddClusterTaint(controlPlaneClient, targetClusterName, *framework.NotReadyTaintTemplate)
+					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
 
-						disabledClusters = append(disabledClusters, targetClusterName)
-						temp--
-					}
+					disabledClusters = append(disabledClusters, targetClusterName)
 				}
 			})
 
 			ginkgo.By("check whether deployment of failed cluster is rescheduled to other available cluster", func() {
 				gomega.Eventually(func() int {
-					targetClusterNames = framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
-					for _, targetClusterName := range targetClusterNames {
-						// the target cluster should be overwritten to another available cluster
-						if !testhelper.IsExclude(targetClusterName, disabledClusters) {
+					currentClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+					for _, currentCluster := range currentClusters {
+						// the current cluster should be overwritten to another available cluster
+						if !testhelper.IsExclude(currentCluster, disabledClusters) {
 							return 0
 						}
 					}
 
-					return len(targetClusterNames)
+					return len(currentClusters)
 				}, pollTimeout, pollInterval).Should(gomega.Equal(minGroups))
 			})
 
-			ginkgo.By("recover cluster", func() {
+			ginkgo.By(fmt.Sprintf("remove taint %v from the target clusters", framework.NotReadyTaintTemplate), func() {
 				for _, disabledCluster := range disabledClusters {
 					err := framework.RemoveClusterTaint(controlPlaneClient, disabledCluster, *framework.NotReadyTaintTemplate)
 					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
@@ -145,50 +135,23 @@ var _ = framework.SerialDescribe("failover testing", func() {
 		})
 	})
 
-	ginkgo.Context("Taint cluster testing", func() {
-		var policyNamespace, policyName string
-		var deploymentNamespace, deploymentName string
-		var deployment *appsv1.Deployment
+	ginkgo.Context("Cluster failover testing with purgeMode gracefully", func() {
 		var taint corev1.Taint
-		var maxGroups, minGroups, numOfFailedClusters int
-		var policy *policyv1alpha1.PropagationPolicy
-		maxGroups = 1
-		minGroups = 1
-		numOfFailedClusters = 1
 
 		ginkgo.BeforeEach(func() {
-			policyNamespace = testNamespace
-			policyName = deploymentNamePrefix + rand.String(RandomStrLength)
-			deploymentNamespace = testNamespace
-			deploymentName = policyName
-			deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
-
-			policy = testhelper.NewPropagationPolicy(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
+			policy.Spec.Placement.ClusterTolerations = []corev1.Toleration{
 				{
-					APIVersion: deployment.APIVersion,
-					Kind:       deployment.Kind,
-					Name:       deployment.Name,
+					Key:               "fail-test",
+					Effect:            corev1.TaintEffectNoExecute,
+					Operator:          corev1.TolerationOpExists,
+					TolerationSeconds: ptr.To[int64](3),
 				},
-			}, policyv1alpha1.Placement{
-				ClusterAffinity: &policyv1alpha1.ClusterAffinity{
-					ClusterNames: framework.ClusterNames(),
-				},
-				ClusterTolerations: []corev1.Toleration{
-					{
-						Key:               "fail-test",
-						Effect:            corev1.TaintEffectNoExecute,
-						Operator:          corev1.TolerationOpExists,
-						TolerationSeconds: ptr.To[int64](3),
-					},
-				},
-				SpreadConstraints: []policyv1alpha1.SpreadConstraint{
-					{
-						SpreadByField: policyv1alpha1.SpreadByFieldCluster,
-						MaxGroups:     maxGroups,
-						MinGroups:     minGroups,
-					},
+			}
+			policy.Spec.Failover = &policyv1alpha1.FailoverBehavior{
+				Cluster: &policyv1alpha1.ClusterFailoverBehavior{
+					PurgeMode: policyv1alpha1.PurgeModeGracefully,
 				},
-			})
+			}
 
 			taint = corev1.Taint{
 				Key:    "fail-test",
@@ -196,129 +159,220 @@ var _ = framework.SerialDescribe("failover testing", func() {
 			}
 		})
 
-		ginkgo.BeforeEach(func() {
-			framework.CreatePropagationPolicy(karmadaClient, policy)
-			framework.CreateDeployment(kubeClient, deployment)
-			ginkgo.DeferCleanup(func() {
-				framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
-				framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
-			})
-		})
-
 		ginkgo.It("taint Cluster with NoExecute taint", func() {
 			var disabledClusters []string
 			targetClusterNames := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
-			ginkgo.By(fmt.Sprintf("add taint %v to the random one cluster", taint), func() {
-				temp := numOfFailedClusters
+			ginkgo.By(fmt.Sprintf("add taint %v to the target clusters", taint), func() {
 				for _, targetClusterName := range targetClusterNames {
-					if temp > 0 {
-						err := framework.AddClusterTaint(controlPlaneClient, targetClusterName, taint)
-						gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+					err := framework.AddClusterTaint(controlPlaneClient, targetClusterName, taint)
+					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
 
-						disabledClusters = append(disabledClusters, targetClusterName)
-						temp--
-					}
+					disabledClusters = append(disabledClusters, targetClusterName)
 				}
 			})
 
 			ginkgo.By("check whether deployment of taint cluster is rescheduled to other available cluster", func() {
 				gomega.Eventually(func() int {
-					targetClusterNames = framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
-					for _, targetClusterName := range targetClusterNames {
-						// the target cluster should be overwritten to another available cluster
-						if !testhelper.IsExclude(targetClusterName, disabledClusters) {
+					currentClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+					for _, currentCluster := range currentClusters {
+						// the current cluster should be overwritten to another available cluster
+						if !testhelper.IsExclude(currentCluster, disabledClusters) {
 							return 0
 						}
 					}
 
-					return len(targetClusterNames)
+					return len(currentClusters)
 				}, pollTimeout, pollInterval).Should(gomega.Equal(minGroups))
 			})
 
-			ginkgo.By("recover cluster", func() {
+			ginkgo.By(fmt.Sprintf("remove taint %v from the target clusters", taint), func() {
 				for _, disabledCluster := range disabledClusters {
 					err := framework.RemoveClusterTaint(controlPlaneClient, disabledCluster, taint)
 					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
 				}
 			})
 
-			ginkgo.By("check whether the deployment disappears in the recovered clusters", func() {
+			ginkgo.By("check whether the deployment disappears in the target clusters", func() {
 				framework.WaitDeploymentDisappearOnClusters(disabledClusters, deploymentNamespace, deploymentName)
 			})
 		})
 	})
 
-	ginkgo.Context("Application failover testing with purgeMode gracefully", func() {
-		var policyNamespace, policyName string
-		var deploymentNamespace, deploymentName string
-		var deployment *appsv1.Deployment
-		var policy *policyv1alpha1.PropagationPolicy
-		var overridePolicy *policyv1alpha1.OverridePolicy
-		var maxGroups, minGroups int
-		var gracePeriodSeconds, tolerationSeconds int32
+	ginkgo.Context("Cluster failover testing with purgeMode directly", func() {
+		var taint corev1.Taint
+
 		ginkgo.BeforeEach(func() {
-			policyNamespace = testNamespace
-			policyName = deploymentNamePrefix + rand.String(RandomStrLength)
-			deploymentNamespace = testNamespace
-			deploymentName = policyName
-			deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
-			maxGroups = 1
-			minGroups = 1
-			gracePeriodSeconds = 30
-			tolerationSeconds = 30
+			// In order to prevent the injected information from being removed after the new application becomes healthy,
+			// we set the deployment image in both clusters to a non-existent image.
+			// After a failover occurs once, since there are only two candidate clusters, migration cannot be triggered again.
+			deployment.Spec.Template.Spec.Containers[0].Image = "fake/nginx:1.19.0"
 
-			policy = &policyv1alpha1.PropagationPolicy{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: policyNamespace,
-					Name:      policyName,
+			policy.Spec.Placement.ClusterTolerations = []corev1.Toleration{
+				{
+					Key:               "fail-test",
+					Effect:            corev1.TaintEffectNoExecute,
+					Operator:          corev1.TolerationOpExists,
+					TolerationSeconds: ptr.To[int64](3),
 				},
-				Spec: policyv1alpha1.PropagationSpec{
-					ResourceSelectors: []policyv1alpha1.ResourceSelector{
-						{
-							APIVersion: deployment.APIVersion,
-							Kind:       deployment.Kind,
-							Name:       deployment.Name,
-						},
-					},
-					Placement: policyv1alpha1.Placement{
-						ClusterAffinity: &policyv1alpha1.ClusterAffinity{
-							ClusterNames: framework.ClusterNames(),
-						},
-						SpreadConstraints: []policyv1alpha1.SpreadConstraint{
+			}
+			policy.Spec.Failover = &policyv1alpha1.FailoverBehavior{
+				Cluster: &policyv1alpha1.ClusterFailoverBehavior{
+					PurgeMode: policyv1alpha1.PurgeModeDirectly,
+					StatePreservation: &policyv1alpha1.StatePreservation{
+						Rules: []policyv1alpha1.StatePreservationRule{
 							{
-								SpreadByField: policyv1alpha1.SpreadByFieldCluster,
-								MaxGroups:     maxGroups,
-								MinGroups:     minGroups,
+								AliasLabelName: "test-alias",
+								JSONPath:       "{.replicas}",
 							},
 						},
 					},
-					PropagateDeps: true,
-					Failover: &policyv1alpha1.FailoverBehavior{
-						Application: &policyv1alpha1.ApplicationFailoverBehavior{
-							DecisionConditions: policyv1alpha1.DecisionConditions{
-								TolerationSeconds: ptr.To[int32](tolerationSeconds),
-							},
-							PurgeMode:          policyv1alpha1.PurgeModeGracefully,
-							GracePeriodSeconds: ptr.To[int32](gracePeriodSeconds),
+				},
+			}
+
+			taint = corev1.Taint{
+				Key:    "fail-test",
+				Effect: corev1.TaintEffectNoExecute,
+			}
+		})
+
+		ginkgo.It("taint Cluster with NoExecute taint", func() {
+			targetClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+			ginkgo.By(fmt.Sprintf("add taint %v to the target clusters", taint), func() {
+				for _, targetCluster := range targetClusters {
+					err := framework.AddClusterTaint(controlPlaneClient, targetCluster, taint)
+					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				}
+			})
+
+			var newClusters []string
+			ginkgo.By("check whether deployment of taint cluster is rescheduled to other available cluster", func() {
+				gomega.Eventually(func() int {
+					newClusters = framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+					for _, newCluster := range newClusters {
+						// the target cluster should be overwritten to another available cluster
+						if !testhelper.IsExclude(newCluster, targetClusters) {
+							return 0
+						}
+					}
+
+					return len(newClusters)
+				}, pollTimeout, pollInterval).Should(gomega.Equal(minGroups))
+			})
+
+			ginkgo.By("check whether the new deployment has the correct inject info", func() {
+				gomega.Eventually(func() bool {
+					for _, cluster := range newClusters {
+						clusterClient := framework.GetClusterClient(cluster)
+						gomega.Expect(clusterClient).ShouldNot(gomega.BeNil())
+
+						deployment, err := clusterClient.AppsV1().Deployments(deploymentNamespace).Get(context.TODO(), deploymentName, metav1.GetOptions{})
+						if err != nil {
+							return false
+						}
+						_, exist := deployment.Labels["test-alias"]
+						if !exist {
+							return false
+						}
+					}
+					return true
+				}, pollTimeout, pollInterval).Should(gomega.BeTrue())
+			})
+
+			ginkgo.By("check whether the failed deployment disappears in the targetClusters", func() {
+				for _, cluster := range targetClusters {
+					clusterClient := framework.GetClusterClient(cluster)
+					gomega.Expect(clusterClient).ShouldNot(gomega.BeNil())
+
+					_, err := clusterClient.AppsV1().Deployments(deploymentNamespace).Get(context.TODO(), deploymentName, metav1.GetOptions{})
+					gomega.Expect(apierrors.IsNotFound(err)).Should(gomega.BeTrue())
+				}
+			})
+
+			ginkgo.By(fmt.Sprintf("remove taint %v from the target clusters", taint), func() {
+				for _, targetCluster := range targetClusters {
+					err := framework.RemoveClusterTaint(controlPlaneClient, targetCluster, taint)
+					gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+				}
+			})
+		})
+	})
+})
+
+var _ = ginkgo.Describe("application failover testing", func() {
+	var policyNamespace, policyName string
+	var deploymentNamespace, deploymentName string
+	var deployment *appsv1.Deployment
+	var policy *policyv1alpha1.PropagationPolicy
+	var maxGroups, minGroups int
+	var gracePeriodSeconds, tolerationSeconds int32
+
+	ginkgo.BeforeEach(func() {
+		policyNamespace = testNamespace
+		policyName = deploymentNamePrefix + rand.String(RandomStrLength)
+		deploymentNamespace = testNamespace
+		deploymentName = policyName
+		deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
+		maxGroups = 1
+		minGroups = 1
+
+		policy = &policyv1alpha1.PropagationPolicy{
+			ObjectMeta: metav1.ObjectMeta{
+				Namespace: policyNamespace,
+				Name:      policyName,
+			},
+			Spec: policyv1alpha1.PropagationSpec{
+				ResourceSelectors: []policyv1alpha1.ResourceSelector{
+					{
+						APIVersion: deployment.APIVersion,
+						Kind:       deployment.Kind,
+						Name:       deployment.Name,
+					},
+				},
+				Placement: policyv1alpha1.Placement{
+					ClusterAffinity: &policyv1alpha1.ClusterAffinity{
+						ClusterNames: framework.ClusterNames()[:2],
+					},
+					SpreadConstraints: []policyv1alpha1.SpreadConstraint{
+						{
+							SpreadByField: policyv1alpha1.SpreadByFieldCluster,
+							MaxGroups:     maxGroups,
+							MinGroups:     minGroups,
 						},
 					},
 				},
-			}
+				PropagateDeps: true,
+			},
+		}
+	})
+
+	ginkgo.JustBeforeEach(func() {
+		framework.CreatePropagationPolicy(karmadaClient, policy)
+		framework.CreateDeployment(kubeClient, deployment)
+		ginkgo.DeferCleanup(func() {
+			framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
+			framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
 		})
+	})
 
+	ginkgo.Context("Application failover testing with purgeMode gracefully", func() {
 		ginkgo.BeforeEach(func() {
-			framework.CreatePropagationPolicy(karmadaClient, policy)
-			framework.CreateDeployment(kubeClient, deployment)
-			ginkgo.DeferCleanup(func() {
-				framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
-				framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
-			})
+			gracePeriodSeconds = 30
+			tolerationSeconds = 30
+			policy.Spec.Failover = &policyv1alpha1.FailoverBehavior{
+				Application: &policyv1alpha1.ApplicationFailoverBehavior{
+					DecisionConditions: policyv1alpha1.DecisionConditions{
+						TolerationSeconds: ptr.To[int32](tolerationSeconds),
+					},
+					PurgeMode:          policyv1alpha1.PurgeModeGracefully,
+					GracePeriodSeconds: ptr.To[int32](gracePeriodSeconds),
+				},
+			}
 		})
 
 		ginkgo.It("application failover with purgeMode gracefully when the application come back to healthy on the new cluster", func() {
 			disabledClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
-			ginkgo.By("create an error op", func() {
-				overridePolicy = testhelper.NewOverridePolicyByOverrideRules(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
+			ginkgo.By("create an overridePolicy to make the application unhealthy", func() {
+				overridePolicy := testhelper.NewOverridePolicyByOverrideRules(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
 					{
 						APIVersion: deployment.APIVersion,
 						Kind:       deployment.Kind,
@@ -342,6 +396,7 @@ var _ = framework.SerialDescribe("failover testing", func() {
 				})
 				framework.CreateOverridePolicy(karmadaClient, overridePolicy)
 			})
+			defer framework.RemoveOverridePolicy(karmadaClient, policyNamespace, policyName)
 
 			ginkgo.By("check if deployment present on member clusters has correct image value", func() {
 				framework.WaitDeploymentPresentOnClustersFitWith(disabledClusters, deployment.Namespace, deployment.Name,
@@ -372,15 +427,11 @@ var _ = framework.SerialDescribe("failover testing", func() {
 					return len(targetClusterNames)
 				}, pollTimeout, pollInterval).Should(gomega.Equal(minGroups))
 			})
-
-			ginkgo.By("delete the error op", func() {
-				framework.RemoveOverridePolicy(karmadaClient, policyNamespace, policyName)
-			})
 		})
 
 		ginkgo.It("application failover with purgeMode gracefully when the GracePeriodSeconds is reach out", func() {
 			gracePeriodSeconds = 10
-			ginkgo.By("update pp", func() {
+			ginkgo.By("update the gracePeriodSeconds of the pp", func() {
 				// modify gracePeriodSeconds to create a time difference with tolerationSecond to avoid cluster interference
 				patch := []map[string]interface{}{
 					{
@@ -394,8 +445,8 @@ var _ = framework.SerialDescribe("failover testing", func() {
 
 			disabledClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
 			var beginTime time.Time
-			ginkgo.By("create an error op", func() {
-				overridePolicy = testhelper.NewOverridePolicyByOverrideRules(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
+			ginkgo.By("create an overridePolicy to make the application unhealthy", func() {
+				overridePolicy := testhelper.NewOverridePolicyByOverrideRules(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
 					{
 						APIVersion: deployment.APIVersion,
 						Kind:       deployment.Kind,
@@ -444,95 +495,22 @@ var _ = framework.SerialDescribe("failover testing", func() {
 	})
 
 	ginkgo.Context("Application failover testing with purgeMode never", func() {
-		var policyNamespace, policyName string
-		var deploymentNamespace, deploymentName string
-		var deployment *appsv1.Deployment
-		var policy *policyv1alpha1.PropagationPolicy
-		var overridePolicy *policyv1alpha1.OverridePolicy
-		var maxGroups, minGroups int
 		ginkgo.BeforeEach(func() {
-			policyNamespace = testNamespace
-			policyName = deploymentNamePrefix + rand.String(RandomStrLength)
-			deploymentNamespace = testNamespace
-			deploymentName = policyName
-			deployment = testhelper.NewDeployment(deploymentNamespace, deploymentName)
-			maxGroups = 1
-			minGroups = 1
-
-			policy = &policyv1alpha1.PropagationPolicy{
-				ObjectMeta: metav1.ObjectMeta{
-					Namespace: policyNamespace,
-					Name:      policyName,
-				},
-				Spec: policyv1alpha1.PropagationSpec{
-					ResourceSelectors: []policyv1alpha1.ResourceSelector{
-						{
-							APIVersion: deployment.APIVersion,
-							Kind:       deployment.Kind,
-							Name:       deployment.Name,
-						},
-					},
-					Placement: policyv1alpha1.Placement{
-						ClusterAffinity: &policyv1alpha1.ClusterAffinity{
-							ClusterNames: framework.ClusterNames(),
-						},
-						SpreadConstraints: []policyv1alpha1.SpreadConstraint{
-							{
-								SpreadByField: policyv1alpha1.SpreadByFieldCluster,
-								MaxGroups:     maxGroups,
-								MinGroups:     minGroups,
-							},
-						},
-					},
-					PropagateDeps: true,
-					Failover: &policyv1alpha1.FailoverBehavior{
-						Application: &policyv1alpha1.ApplicationFailoverBehavior{
-							DecisionConditions: policyv1alpha1.DecisionConditions{
-								TolerationSeconds: ptr.To[int32](30),
-							},
-							PurgeMode: policyv1alpha1.Never,
-						},
+			tolerationSeconds = 30
+			policy.Spec.Failover = &policyv1alpha1.FailoverBehavior{
+				Application: &policyv1alpha1.ApplicationFailoverBehavior{
+					DecisionConditions: policyv1alpha1.DecisionConditions{
+						TolerationSeconds: ptr.To[int32](tolerationSeconds),
 					},
+					PurgeMode: policyv1alpha1.Never,
 				},
 			}
-		})
 
-		ginkgo.BeforeEach(func() {
-			framework.CreatePropagationPolicy(karmadaClient, policy)
-			framework.CreateDeployment(kubeClient, deployment)
-			ginkgo.DeferCleanup(func() {
-				framework.RemoveDeployment(kubeClient, deployment.Namespace, deployment.Name)
-				framework.RemovePropagationPolicy(karmadaClient, policy.Namespace, policy.Name)
-			})
+			deployment.Spec.Template.Spec.Containers[0].Image = "fake/nginx:1.19.0"
 		})
 
 		ginkgo.It("application failover with purgeMode never", func() {
 			disabledClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
-			ginkgo.By("create an error op", func() {
-				overridePolicy = testhelper.NewOverridePolicyByOverrideRules(policyNamespace, policyName, []policyv1alpha1.ResourceSelector{
-					{
-						APIVersion: deployment.APIVersion,
-						Kind:       deployment.Kind,
-						Name:       deployment.Name,
-					},
-				}, []policyv1alpha1.RuleWithCluster{
-					{
-						TargetCluster: &policyv1alpha1.ClusterAffinity{
-							ClusterNames: disabledClusters,
-						},
-						Overriders: policyv1alpha1.Overriders{
-							ImageOverrider: []policyv1alpha1.ImageOverrider{
-								{
-									Component: "Registry",
-									Operator:  policyv1alpha1.OverriderOpReplace,
-									Value:     "fake",
-								},
-							},
-						},
-					},
-				})
-				framework.CreateOverridePolicy(karmadaClient, overridePolicy)
-			})
 
 			ginkgo.By("check if deployment present on member clusters has correct image value", func() {
 				framework.WaitDeploymentPresentOnClustersFitWith(disabledClusters, deployment.Namespace, deployment.Name,
@@ -563,9 +541,80 @@ var _ = framework.SerialDescribe("failover testing", func() {
 			ginkgo.By("check whether the failed deployment is present on the disabledClusters", func() {
 				framework.WaitDeploymentPresentOnClustersFitWith(disabledClusters, deploymentNamespace, deploymentName, func(*appsv1.Deployment) bool { return true })
 			})
+		})
+	})
 
-			ginkgo.By("delete the error op", func() {
-				framework.RemoveOverridePolicy(karmadaClient, policyNamespace, policyName)
+	ginkgo.Context("Application failover testing with purgeMode directly", func() {
+		ginkgo.BeforeEach(func() {
+			tolerationSeconds = 30
+			policy.Spec.Failover = &policyv1alpha1.FailoverBehavior{
+				Application: &policyv1alpha1.ApplicationFailoverBehavior{
+					DecisionConditions: policyv1alpha1.DecisionConditions{
+						TolerationSeconds: ptr.To[int32](tolerationSeconds),
+					},
+					PurgeMode: policyv1alpha1.PurgeModeDirectly,
+					StatePreservation: &policyv1alpha1.StatePreservation{
+						Rules: []policyv1alpha1.StatePreservationRule{
+							{
+								AliasLabelName: "test-alias",
+								JSONPath:       "{.replicas}",
+							},
+						},
+					},
+				},
+			}
+
+			// In order to simulate a deployment resource failure scenario, we set a non-existent image here.
+			// At the same time, to prevent the injected information from being removed after the new application becomes healthy,
+			// we set the deployment image in both clusters to a non-existent image.
+			// After a failover occurs once, since there are only two candidate clusters, migration cannot be triggered again.
+			deployment.Spec.Template.Spec.Containers[0].Image = "fake/nginx:1.19.0"
+		})
+
+		ginkgo.It("Delete the old ones first, then create the new ones", func() {
+			disabledClusters := framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+
+			var targetClusterNames []string
+			ginkgo.By("check whether the failed deployment is rescheduled to other available cluster", func() {
+				gomega.Eventually(func() int {
+					targetClusterNames = framework.ExtractTargetClustersFromRB(controlPlaneClient, deployment.Kind, deployment.Namespace, deployment.Name)
+					for _, targetClusterName := range targetClusterNames {
+						// the target cluster should be overwritten to another available cluster
+						if !testhelper.IsExclude(targetClusterName, disabledClusters) {
+							return 0
+						}
+					}
+					return len(targetClusterNames)
+				}, pollTimeout, pollInterval).Should(gomega.Equal(minGroups))
+			})
+
+			ginkgo.By("check whether the new deployment has the correct inject info", func() {
+				gomega.Eventually(func() bool {
+					for _, cluster := range targetClusterNames {
+						clusterClient := framework.GetClusterClient(cluster)
+						gomega.Expect(clusterClient).ShouldNot(gomega.BeNil())
+
+						deployment, err := clusterClient.AppsV1().Deployments(deploymentNamespace).Get(context.TODO(), deploymentName, metav1.GetOptions{})
+						if err != nil {
+							return false
+						}
+						_, exist := deployment.Labels["test-alias"]
+						if !exist {
+							return false
+						}
+					}
+					return true
+				}, pollTimeout, pollInterval).Should(gomega.BeTrue())
+			})
+
+			ginkgo.By("check whether the failed deployment disappears in the disabledClusters", func() {
+				for _, cluster := range disabledClusters {
+					clusterClient := framework.GetClusterClient(cluster)
+					gomega.Expect(clusterClient).ShouldNot(gomega.BeNil())
+
+					_, err := clusterClient.AppsV1().Deployments(deploymentNamespace).Get(context.TODO(), deploymentName, metav1.GetOptions{})
+					gomega.Expect(apierrors.IsNotFound(err)).Should(gomega.BeTrue())
+				}
 			})
 		})
 	})