You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@yunikorn.apache.org by ma...@apache.org on 2026/10/08 06:00:37 UTC
(yunikorn-core) branch master updated: [YUNIKORN-3483] Reinstate Partition state transitions (#1174)
This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-core.git
The following commit(s) were added to refs/heads/master by this push:
new 0ef9fd72 [YUNIKORN-3483] Reinstate Partition state transitions (#1174)
0ef9fd72 is described below
commit 0ef9fd72000e2c310fc6d4bf664e2e9007f976c7
Author: mani <ma...@gmail.com>
AuthorDate: Thu Oct 8 11:29:18 2026 +0530
[YUNIKORN-3483] Reinstate Partition state transitions (#1174)
Reinstated the partition object state transitions. State would be marked as "draining" before heading to deletion process which could prevent any activity imposed on it. Activity could be either add app or node.
Closes: #1174
Signed-off-by: mani <ma...@gmail.com>
---
pkg/scheduler/objects/queue.go | 15 ++++++++-------
pkg/scheduler/partition.go | 15 +++++++++++----
pkg/scheduler/partition_manager.go | 2 ++
pkg/scheduler/partition_manager_test.go | 1 +
pkg/scheduler/partition_test.go | 20 ++++++++++++++++++++
5 files changed, 42 insertions(+), 11 deletions(-)
diff --git a/pkg/scheduler/objects/queue.go b/pkg/scheduler/objects/queue.go
index 98dfb654..242c5f18 100644
--- a/pkg/scheduler/objects/queue.go
+++ b/pkg/scheduler/objects/queue.go
@@ -785,6 +785,13 @@ func (sq *Queue) GetQueuePath() string {
return sq.QueuePath
}
+// IsStopped returns true if the queue in Stopped state.
+// No Existing applications will be scheduled
+// No new applications will be accepted.
+func (sq *Queue) IsStopped() bool {
+ return sq.stateMachine.Is(Stopped.String())
+}
+
// IsDraining returns true if the queue in Draining state.
// Existing applications will still be scheduled
// No new applications will be accepted.
@@ -797,12 +804,6 @@ func (sq *Queue) IsRunning() bool {
return sq.stateMachine.Is(Active.String())
}
-// IsStopped returns true if the queue in Stopped state.
-// The queue is skipped for scheduling in this state.
-func (sq *Queue) IsStopped() bool {
- return sq.stateMachine.Is(Stopped.String())
-}
-
// CurrentState returns the current state of the queue in string form.
func (sq *Queue) CurrentState() string {
return sq.stateMachine.Current()
@@ -1166,7 +1167,7 @@ func (sq *Queue) addChildQueue(child *Queue) error {
if sq.isLeaf {
return fmt.Errorf("cannot add a child queue to a leaf queue: %s", sq.QueuePath)
}
- if sq.IsDraining() {
+ if sq.IsDraining() || sq.IsStopped() {
return fmt.Errorf("cannot add a child queue when queue is marked for deletion: %s", sq.QueuePath)
}
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index 209a3a0e..c9ada617 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -285,14 +285,21 @@ func (pc *PartitionContext) markPartitionForRemoval() {
}
}
+// Get the state of the partition.
+// No new nodes and applications will be accepted if stopped or being removed.
+func (pc *PartitionContext) isStopped() bool {
+ return pc.stateMachine.Current() == objects.Stopped.String()
+}
+
// Get the state of the partition.
// No new nodes and applications will be accepted if stopped or being removed.
func (pc *PartitionContext) isDraining() bool {
return pc.stateMachine.Current() == objects.Draining.String()
}
-func (pc *PartitionContext) isStopped() bool {
- return pc.stateMachine.Current() == objects.Stopped.String()
+// IsRunning returns true if the partition in Active state.
+func (pc *PartitionContext) IsRunning() bool {
+ return pc.stateMachine.Is(objects.Active.String())
}
// Handle the state event for the partition.
@@ -324,7 +331,7 @@ func (pc *PartitionContext) getPlacementManager() *placement.AppPlacementManager
// NOTE: this is a lock free call. It must NOT be called holding the PartitionContext lock.
func (pc *PartitionContext) AddApplication(app *objects.Application) error {
if pc.isDraining() || pc.isStopped() {
- return fmt.Errorf("partition %s is stopped cannot add a new application %s", pc.Name, app.ApplicationID)
+ return fmt.Errorf("partition %s is stopped or draining, cannot add a new application %s", pc.Name, app.ApplicationID)
}
// Check if the app exists
@@ -727,7 +734,7 @@ func (pc *PartitionContext) AddNode(node *objects.Node) error {
zap.String("partition", pc.Name),
zap.String("nodeID", node.NodeID))
if pc.isDraining() || pc.isStopped() {
- return fmt.Errorf("partition %s is stopped cannot add a new node %s", pc.Name, node.NodeID)
+ return fmt.Errorf("partition %s is draining or stopped, cannot add a new node %s", pc.Name, node.NodeID)
}
if err := pc.addNodeToList(node); err != nil {
return err
diff --git a/pkg/scheduler/partition_manager.go b/pkg/scheduler/partition_manager.go
index 50427826..bf2b535e 100644
--- a/pkg/scheduler/partition_manager.go
+++ b/pkg/scheduler/partition_manager.go
@@ -142,6 +142,8 @@ func (manager *partitionManager) cleanQueues(queue *objects.Queue) {
func (manager *partitionManager) remove() {
log.Log(log.SchedPartition).Info("marking all queues for removal",
zap.String("partitionName", manager.pc.Name))
+ // mark partition for removal
+ manager.pc.markPartitionForRemoval()
// mark all queues for removal
manager.pc.root.MarkQueueForRemoval()
// remove applications: we do not care about return values or issues
diff --git a/pkg/scheduler/partition_manager_test.go b/pkg/scheduler/partition_manager_test.go
index 76fdccbe..fc6a2641 100644
--- a/pkg/scheduler/partition_manager_test.go
+++ b/pkg/scheduler/partition_manager_test.go
@@ -99,6 +99,7 @@ func TestRemoveAll(t *testing.T) {
// make sure all nodes and applications are removed
p.partitionManager.remove()
+ assert.Assert(t, p.isDraining())
assert.Equal(t, 0, len(p.applications))
assert.Equal(t, 0, p.nodes.GetNodeCount())
}
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index a63b61b7..d4f621b2 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -5759,3 +5759,23 @@ func TestRemoveAllocationPlaceholderReplacedWithoutReplacement(t *testing.T) {
assert.Assert(t, node1.GetAllocation("placeholder") == nil, "placeholder should be removed from node")
assert.Assert(t, resources.IsZero(partition.GetQueue(defQueue).GetAllocatedResource()), "queue resource should be zero after placeholder removed")
}
+
+func TestPartitionStates(t *testing.T) {
+ p := createPartitionContext(t)
+ err := p.handlePartitionEvent(objects.Stop)
+ if err != nil || !p.isStopped() {
+ t.Errorf("partition is not marked stopped: %v", err)
+ }
+ err = p.handlePartitionEvent(objects.Start)
+ if err != nil || !p.IsRunning() {
+ t.Errorf("partition is not marked running: %v", err)
+ }
+ err = p.handlePartitionEvent(objects.Remove)
+ if err != nil || !p.isDraining() {
+ t.Errorf("partition is not marked draining: %v", err)
+ }
+ err = p.handlePartitionEvent(objects.Start)
+ if err != nil || !p.IsRunning() {
+ t.Errorf("partition is not marked running: %v", err)
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@yunikorn.apache.org
For additional commands, e-mail: issues-help@yunikorn.apache.org