Add support for SchedulerName for Flink pods (#300)

- Provides the ability to add Scheduler name from config.

Author: anandswaminathan

pkg/controller/config/config.go

@@ -25,6 +25,7 @@ type Config struct {
 	MaxBackoffDuration    config.Duration `json:"maxBackoffDuration" pflag:"\"30s\",Determines the max backoff for exponential retries."`
 	MaxErrDuration        config.Duration `json:"maxErrDuration" pflag:"\"5m\",Determines the max time to wait on errors."`
 	FlinkJobVertexTimeout config.Duration `json:"flinkJobVertexTimeout" pflag:"\"3m\",Determines the max time to wait on job vertex state turns into RUNNING."`
+	SchedulerName         string          `json:"schedulerName"`
 }
 
 func GetConfig() *Config {

pkg/controller/config/config_flags.go

@@ -371,6 +371,7 @@ func jobmanagerTemplate(app *v1beta1.FlinkApplication) *v1.Deployment {
 					NodeSelector:     app.Spec.JobManagerConfig.NodeSelector,
 					Tolerations:      app.Spec.JobManagerConfig.Tolerations,
 					Affinity:         app.Spec.JobManagerConfig.Affinity,
+					SchedulerName:    config.GetConfig().SchedulerName,
 				},
 			},
 		},

pkg/controller/flink/job_manager_controller.go

@@ -216,6 +216,7 @@ func taskmanagerTemplate(app *v1beta1.FlinkApplication) *v1.Deployment {
 					NodeSelector:     app.Spec.TaskManagerConfig.NodeSelector,
 					Tolerations:      app.Spec.TaskManagerConfig.Tolerations,
 					Affinity:         app.Spec.TaskManagerConfig.Affinity,
+					SchedulerName:    config.GetConfig().SchedulerName,
 				},
 			},
 		},