Add scheduler func for clearing batch scheduling on completed

pkg/batchscheduler/interface/interface.go

@@ -25,4 +25,5 @@ type BatchScheduler interface {
 
 	ShouldSchedule(app *v1beta2.SparkApplication) bool
 	DoBatchSchedulingOnSubmission(app *v1beta2.SparkApplication) error
+	CleanupOnCompletion(app *v1beta2.SparkApplication) error
 }

pkg/batchscheduler/volcano/volcano_scheduler.go

@@ -160,6 +160,12 @@ func (v *VolcanoBatchScheduler) syncPodGroup(app *v1beta2.SparkApplication, size
 	return nil
 }
 
+func (v *VolcanoBatchScheduler) CleanupOnCompletion(app *v1beta2.SparkApplication) error {
+	podGroupName := v.getAppPodGroupName(app)
+	//Remove pod group for Spark Application
+	return v.volcanoClient.SchedulingV1beta1().PodGroups(app.Namespace).Delete(podGroupName, &metav1.DeleteOptions{})
+}
+
 func New(config *rest.Config) (schedulerinterface.BatchScheduler, error) {
 	vkClient, err := volcanoclient.NewForConfig(config)
 	if err != nil {

pkg/controller/sparkapplication/controller.go

@@ -597,6 +597,14 @@ func (c *Controller) syncSparkApplication(key string) error {
 			glog.Errorf("failed to update SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
 			return err
 		}
+
+		if state := appCopy.Status.AppState.State; state == v1beta2.CompletedState ||
+			state == v1beta2.FailedState {
+			if err := c.cleanUpOnTermination(app, appCopy); err != nil {
+				glog.Errorf("failed to clean up resources for SparkApplication %s/%s: %v", app.Namespace, app.Name, err)
+				return err
+			}
+		}
 	}
 
 	return nil
@@ -1001,3 +1009,16 @@ func (c *Controller) hasApplicationExpired(app *v1beta2.SparkApplication) bool {
 
 	return false
 }
+
+// Clean up when the spark application is terminated.
+func (c *Controller) cleanUpOnTermination(oldApp, newApp *v1beta2.SparkApplication) error {
+	if needScheduling, scheduler := c.shouldDoBatchScheduling(newApp); needScheduling {
+		// batch schduler is cleaned up only when state is changed to completion state
+		if newApp.Status.AppState.State != oldApp.Status.AppState.State {
+			if err := scheduler.CleanupOnCompletion(newApp); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}