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-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 84b0e403 [YUNIKORN-3367] Fix Kubernetes shim shutdown (#1074)
84b0e403 is described below
commit 84b0e403bdd739c07aaea602d06d36f5d26a8a98
Author: andyhuangdev <[email protected]>
AuthorDate: Tue Aug 25 11:50:14 2026 +0530
[YUNIKORN-3367] Fix Kubernetes shim shutdown (#1074)
Fix KubernetesShim.Stop() to close its shared stop channel exactly once.
Previously, Stop() sent one value to an unbuffered channel shared by two
scheduling loops. Only one loop stopped, while the other could continue
scheduling after the shim was stopped. Also, if Run() failed before the
scheduling loops started, no receiver was present and Stop() skipped dispatcher
and placeholder-manager cleanup.
This change uses sync.Once to make shutdown idempotent, closes the channel
to stop both scheduling loops, and always releases dispatcher and
placeholder-manager resources during failed startup cleanup.
Closes: #1074
Signed-off-by: Manikandan R <[email protected]>
---
pkg/shim/scheduler.go | 19 ++++++++++++------
pkg/shim/scheduler_mock_test.go | 1 -
pkg/shim/scheduler_test.go | 44 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 57 insertions(+), 7 deletions(-)
diff --git a/pkg/shim/scheduler.go b/pkg/shim/scheduler.go
index 7f5a07f1..9b8b5c9b 100644
--- a/pkg/shim/scheduler.go
+++ b/pkg/shim/scheduler.go
@@ -20,6 +20,7 @@ package shim
import (
ctx "context"
+ "sync"
"time"
"go.uber.org/zap"
@@ -49,6 +50,7 @@ type KubernetesShim struct {
phManager *cache.PlaceholderManager
callback api.ResourceManagerCallback
stopChan chan struct{}
+ stopOnce sync.Once
lock *locking.RWMutex
outstandingAppsFound bool
}
@@ -224,14 +226,19 @@ func (ss *KubernetesShim) Run() error {
}
func (ss *KubernetesShim) Stop() {
- log.Log(log.ShimScheduler).Info("stopping scheduler")
- select {
- case ss.stopChan <- struct{}{}:
- // stop the dispatcher
- dispatcher.Stop()
+ stopped := false
+ ss.stopOnce.Do(func() {
+ stopped = true
+ log.Log(log.ShimScheduler).Info("stopping scheduler")
+ close(ss.stopChan)
+ // stop the client library code that communicates with
Kubernetes
+ ss.apiFactory.Stop()
// stop the placeholder manager
ss.phManager.Stop()
- default:
+ // stop the dispatcher
+ dispatcher.Stop()
+ })
+ if !stopped {
log.Log(log.ShimScheduler).Info("scheduler is already stopped")
}
}
diff --git a/pkg/shim/scheduler_mock_test.go b/pkg/shim/scheduler_mock_test.go
index f91a3c21..96f4024f 100644
--- a/pkg/shim/scheduler_mock_test.go
+++ b/pkg/shim/scheduler_mock_test.go
@@ -204,7 +204,6 @@ func (fc *MockScheduler) waitAndVerifySchedulerAllocations(
func (fc *MockScheduler) stop() {
close(fc.stopChan)
fc.scheduler.Stop()
- fc.apiProvider.Stop()
fc.started.Store(false)
}
diff --git a/pkg/shim/scheduler_test.go b/pkg/shim/scheduler_test.go
index 145cabc1..c3e482bf 100644
--- a/pkg/shim/scheduler_test.go
+++ b/pkg/shim/scheduler_test.go
@@ -175,9 +175,53 @@ func TestSchedulerRegistrationFailed(t *testing.T) {
ctx := cache.NewContext(mockedAPIProvider)
shim := newShimSchedulerInternal(ctx, mockedAPIProvider, callback)
assert.Error(t, shim.Run(), "some error")
+ assertStopChannelClosed(t, shim)
+
+ // Stop must remain safe when Run() already stopped the shim after its
failure.
+ shim.Stop()
+}
+
+func TestSchedulerStopClosesStopChannel(t *testing.T) {
+ mockedAPIProvider := client.NewMockedAPIProvider(false)
+ shim := newShimSchedulerInternal(cache.NewContext(mockedAPIProvider),
mockedAPIProvider, nil)
+
+ shim.Stop()
+ assertStopChannelClosed(t, shim)
+
+ // Stop must be idempotent because callers can invoke it after a failed
Run().
+ shim.Stop()
+}
+
+func TestSchedulerStopStopsAPIFactory(t *testing.T) {
+ mockedAPIProvider := client.NewMockedAPIProvider(false)
+ apiProvider := &trackingAPIProvider{APIProvider: mockedAPIProvider}
+ shim := newShimSchedulerInternal(cache.NewContext(apiProvider),
apiProvider, nil)
+
+ shim.Stop()
+ assert.Check(t, apiProvider.stopped.Load(), "API provider should be
stopped with the scheduler")
+
shim.Stop()
}
+func assertStopChannelClosed(t *testing.T, shim *KubernetesShim) {
+ t.Helper()
+ select {
+ case <-shim.stopChan:
+ default:
+ t.Fatal("scheduler stop channel should be closed")
+ }
+}
+
+type trackingAPIProvider struct {
+ client.APIProvider
+ stopped atomic.Bool
+}
+
+func (p *trackingAPIProvider) Stop() {
+ p.stopped.Store(true)
+ p.APIProvider.Stop()
+}
+
func TestTaskFailures(t *testing.T) {
// init and register scheduler
cluster := MockScheduler{}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
