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 c9211f30 [YUNIKORN-3321] Fix removeAsksInternal return value ignored
(#1114)
c9211f30 is described below
commit c9211f300acaebd3d740794bcf39bcd576a4c019
Author: kaijaytu <[email protected]>
AuthorDate: Mon Aug 3 16:20:01 2026 +0530
[YUNIKORN-3321] Fix removeAsksInternal return value ignored (#1114)
The removeAsksInternal call returns the number of reservations released.
The partition tracks reservations and these releases must be passed back
to update the counter correctly.
- partitionContext.removeApplication: capture return value of
RemoveAllocationAsk and call decReservationCount
- partitionContext.removeAllocation: same fix
- timeoutPlaceholderProcessing: add reservationReleasedCallback
(follows existing terminatedCallback pattern) so the async
placeholder timeout can notify the partition to decrement
Closes: #1114
Signed-off-by: mani <[email protected]>
---
pkg/scheduler/objects/application.go | 26 ++++++++++++++++-----
pkg/scheduler/objects/application_test.go | 32 ++++++++++++++++++++++++++
pkg/scheduler/partition.go | 7 ++++--
pkg/scheduler/partition_test.go | 38 +++++++++++++++++++++++++++++++
4 files changed, 95 insertions(+), 8 deletions(-)
diff --git a/pkg/scheduler/objects/application.go
b/pkg/scheduler/objects/application.go
index d22f094e..d5b408cf 100644
--- a/pkg/scheduler/objects/application.go
+++ b/pkg/scheduler/objects/application.go
@@ -121,11 +121,12 @@ type Application struct {
runnableByUserLimit bool // whether the
application is runnable/schedulable based on user/group quota. Default is true.
backoffDeadline time.Time // no scheduling from
this application until this deadline
- rmEventHandler handler.EventHandler
- rmID string
- terminatedCallback func(appID string)
- appEvents *schedEvt.ApplicationEvents
- sendStateChangeEvents bool // whether to send state-change events or
not (simplifies testing)
+ rmEventHandler handler.EventHandler
+ rmID string
+ terminatedCallback func(appID string)
+ reservationReleasedCallback func(released int)
+ appEvents *schedEvt.ApplicationEvents
+ sendStateChangeEvents bool // whether to send state-change events
or not (simplifies testing)
locking.RWMutex
}
@@ -483,7 +484,8 @@ func (sa *Application) timeoutPlaceholderProcessing() {
zap.Int("pending", len(pendingRelease)),
zap.Int("preempted", preempted),
zap.String("gang scheduling style",
sa.gangSchedulingStyle))
- sa.removeAsksInternal("", si.EventRecord_REQUEST_TIMEOUT)
+ released := sa.removeAsksInternal("",
si.EventRecord_REQUEST_TIMEOUT)
+ sa.executeReservationReleasedCallback(released)
// trigger the release of the allocated placeholders:
accounting updates when the release is done
sa.notifyRMAllocationReleased(toRelease,
si.TerminationType_TIMEOUT, "releasing allocated placeholders on placeholder
timeout")
// trigger the release of the pending placeholders: accounting
has been done
@@ -2162,6 +2164,18 @@ func (sa *Application) executeTerminatedCallback() {
}
}
+func (sa *Application) SetReservationReleasedCallback(callback func(released
int)) {
+ sa.Lock()
+ defer sa.Unlock()
+ sa.reservationReleasedCallback = callback
+}
+
+func (sa *Application) executeReservationReleasedCallback(released int) {
+ if released > 0 && sa.reservationReleasedCallback != nil {
+ go sa.reservationReleasedCallback(released)
+ }
+}
+
// notifyRMAllocationReleased send an allocation release event to the RM to if
the event handler is configured
// and at least one allocation has been released.
// No locking must be called while holding the lock
diff --git a/pkg/scheduler/objects/application_test.go
b/pkg/scheduler/objects/application_test.go
index 2d21f250..fe65d485 100644
--- a/pkg/scheduler/objects/application_test.go
+++ b/pkg/scheduler/objects/application_test.go
@@ -4300,3 +4300,35 @@ func TestApplicationBackoff(t *testing.T) {
assert.Assert(t, result == nil)
assert.Assert(t, app.GetBackoffDeadline().After(beforeTryAlloc))
}
+
+func TestReservationReleasedCallback(t *testing.T) {
+ app := newApplication(appID1, "default", "root.unknown")
+ queue, err := createRootQueue(nil)
+ assert.NilError(t, err, "queue create failed")
+ app.queue = queue
+
+ // callback not set - should not panic
+ app.executeReservationReleasedCallback(5)
+
+ // callback with released = 0 - should not be called
+ called := make(chan int, 1)
+ app.SetReservationReleasedCallback(func(released int) {
+ called <- released
+ })
+ app.executeReservationReleasedCallback(0)
+ select {
+ case <-called:
+ t.Fatal("callback should not be called when released = 0")
+ case <-time.After(10 * time.Millisecond):
+ // expected
+ }
+
+ // callback with released > 0 - should be called with correct value
+ app.executeReservationReleasedCallback(3)
+ select {
+ case val := <-called:
+ assert.Equal(t, 3, val, "callback called with wrong value")
+ case <-time.After(time.Second):
+ t.Fatal("callback was not called")
+ }
+}
diff --git a/pkg/scheduler/partition.go b/pkg/scheduler/partition.go
index 3a8d2ddc..2dd27962 100644
--- a/pkg/scheduler/partition.go
+++ b/pkg/scheduler/partition.go
@@ -402,6 +402,7 @@ func (pc *PartitionContext) AddApplication(app
*objects.Application) error {
// all is OK update the app and add it to the partition
app.SetQueue(queue)
app.SetTerminatedCallback(pc.moveTerminatedApp)
+ app.SetReservationReleasedCallback(pc.decReservationCount)
queue.AddApplication(app)
pc.applications[appID] = app
pc.appQueueMapping.AddAppQueueMapping(appID, queue)
@@ -418,7 +419,8 @@ func (pc *PartitionContext) removeApplication(appID string)
[]*objects.Allocatio
return nil
}
// Remove all asks and thus all reservations and pending resources
(queue included)
- _ = app.RemoveAllocationAsk("")
+ released := app.RemoveAllocationAsk("")
+ pc.decReservationCount(released)
// Remove app from queue
if queue := app.GetQueue(); queue != nil {
queue.RemoveApplication(app)
@@ -1580,7 +1582,8 @@ func (pc *PartitionContext) removeAllocation(release
*si.AllocationRelease) ([]*
if release.TerminationType != si.TerminationType_TIMEOUT {
// handle ask releases as well
- _ = app.RemoveAllocationAsk(allocationKey)
+ released := app.RemoveAllocationAsk(allocationKey)
+ pc.decReservationCount(released)
}
return released, confirmed
diff --git a/pkg/scheduler/partition_test.go b/pkg/scheduler/partition_test.go
index 3368ae0c..e3187301 100644
--- a/pkg/scheduler/partition_test.go
+++ b/pkg/scheduler/partition_test.go
@@ -5562,3 +5562,41 @@ func
TestRemoveAllocationSchedulingFailedOnRMNodeNotFound(t *testing.T) {
assert.Assert(t,
resources.IsZero(partition.GetQueue(defQueue).GetAllocatedResource()), "queue
resource should be zero after rollback")
assert.Assert(t,
resources.StrictlyGreaterThanZero(app.GetPendingResource()), "ask should be
pending again after rollback")
}
+
+func TestRemoveAppWithReservations(t *testing.T) {
+ setupUGM()
+ partition := createQueuesNodes(t)
+ assert.Assert(t, partition != nil, "partition create failed")
+ defer partition.userGroupCache.Stop()
+
+ app := newApplication(appID1, "default", "root.parent.sub-leaf")
+ res, err := resources.NewResourceFromConf(map[string]string{"vcore":
"10"})
+ assert.NilError(t, err, "failed to create resource")
+
+ err = partition.AddApplication(app)
+ assert.NilError(t, err, "failed to add app to partition")
+
+ ask1 := newAllocationAsk(allocKey, appID1, res)
+ ask1.SetRequiredNode(nodeID1)
+ err = app.AddAllocationAsk(ask1)
+ assert.NilError(t, err, "failed to add ask")
+ ask2 := newAllocationAsk(allocKey2, appID1, res)
+ ask2.SetRequiredNode(nodeID1)
+ err = app.AddAllocationAsk(ask2)
+ assert.NilError(t, err, "failed to add ask")
+
+ // ask1 occupies node1
+ result := partition.tryAllocate()
+ assert.Assert(t, result != nil && result.Request != nil, "no alloc")
+ assert.Equal(t, objects.Allocated, result.ResultType)
+ assert.Equal(t, 0, partition.getReservationCount())
+
+ // ask2 gets reserved
+ result = partition.tryAllocate()
+ assert.Assert(t, result == nil)
+ assert.Equal(t, 1, partition.getReservationCount())
+
+ // remove the application - reservation count must be decremented
+ partition.removeApplication(appID1)
+ assert.Equal(t, 0, partition.getReservationCount())
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
