K8s Task launcher to use task launcher account properties

 - Introduce Platform task launcher properties to include K8s task launcher properties
 - Construct the task launcher using this explicit property
 - Update test

Resolves spring-cloud#4186

spring-cloud-dataflow-platform-kubernetes/src/main/java/org/springframework/cloud/dataflow/server/config/kubernetes/KubernetesPlatformTaskLauncherProperties.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2017-2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.dataflow.server.config.kubernetes;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.cloud.dataflow.core.AbstractPlatformProperties;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesTaskLauncherProperties;
+
+/**
+ * @author Ilayaperumal Gopinathan
+ */
+@ConfigurationProperties("spring.cloud.dataflow.task.platform.kubernetes")
+public class KubernetesPlatformTaskLauncherProperties extends AbstractPlatformProperties<KubernetesTaskLauncherProperties> {
+}

spring-cloud-dataflow-platform-kubernetes/src/main/java/org/springframework/cloud/dataflow/server/config/kubernetes/KubernetesTaskPlatformAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -30,15 +30,16 @@
  * @author David Turanski
  */
 @Configuration
-@EnableConfigurationProperties(KubernetesPlatformProperties.class)
+@EnableConfigurationProperties({KubernetesPlatformProperties.class, KubernetesPlatformTaskLauncherProperties.class})
 @ConditionalOnTasksEnabled
 public class KubernetesTaskPlatformAutoConfiguration {
 
 	@Bean
 	public KubernetesTaskPlatformFactory kubernetesTaskPlatformFactory(
 			KubernetesPlatformProperties platformProperties,
-			@Value("${spring.cloud.dataflow.features.schedules-enabled:false}") boolean schedulesEnabled) {
-		return new KubernetesTaskPlatformFactory(platformProperties, schedulesEnabled);
+			@Value("${spring.cloud.dataflow.features.schedules-enabled:false}") boolean schedulesEnabled,
+			KubernetesPlatformTaskLauncherProperties platformTaskLauncherProperties) {
+		return new KubernetesTaskPlatformFactory(platformProperties, schedulesEnabled, platformTaskLauncherProperties);
 	}
 
 	@Bean

spring-cloud-dataflow-platform-kubernetes/src/main/java/org/springframework/cloud/dataflow/server/config/kubernetes/KubernetesTaskPlatformFactory.java

@@ -28,6 +28,7 @@
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesScheduler;
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesSchedulerProperties;
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesTaskLauncher;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesTaskLauncherProperties;
 import org.springframework.cloud.deployer.spi.scheduler.Scheduler;
 
 /**
@@ -36,25 +37,30 @@
  **/
 public class KubernetesTaskPlatformFactory extends AbstractTaskPlatformFactory<KubernetesPlatformProperties> {
 
+	private final KubernetesPlatformTaskLauncherProperties platformTaskLauncherProperties;
+
 	private final boolean schedulesEnabled;
 
 	public KubernetesTaskPlatformFactory(
 			KubernetesPlatformProperties platformProperties,
-			boolean schedulesEnabled) {
+			boolean schedulesEnabled,
+			KubernetesPlatformTaskLauncherProperties kubernetesPlatformTaskLauncherProperties) {
 		super(platformProperties, KUBERNETES_PLATFORM_TYPE);
 		this.schedulesEnabled = schedulesEnabled;
+		this.platformTaskLauncherProperties = kubernetesPlatformTaskLauncherProperties;
 	}
 
 	@Override
 	public Launcher createLauncher(String account) {
 		KubernetesDeployerProperties kubernetesProperties = this.platformProperties.accountProperties(account);
+		KubernetesTaskLauncherProperties taskLauncherProperties = this.platformTaskLauncherProperties.accountProperties(account);
 		ContainerFactory containerFactory = new DefaultContainerFactory(
 				this.platformProperties.accountProperties(account));
 		KubernetesClient kubernetesClient =
 				KubernetesClientFactory.getKubernetesClient(this.platformProperties.accountProperties(account));
 
 		KubernetesTaskLauncher kubernetesTaskLauncher = new KubernetesTaskLauncher(
-				kubernetesProperties, kubernetesClient, containerFactory);
+				kubernetesProperties, taskLauncherProperties, kubernetesClient, containerFactory);
 
 		KubernetesSchedulerProperties kubernetesSchedulerProperties = new KubernetesSchedulerProperties();
 		BeanUtils.copyProperties(kubernetesProperties, kubernetesSchedulerProperties);

spring-cloud-dataflow-platform-kubernetes/src/test/java/org/springframework/cloud/dataflow/server/config/kubernetes/KubernetesTaskPlatformFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019 the original author or authors.
+ * Copyright 2019-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,23 +25,28 @@
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesDeployerProperties;
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesSchedulerProperties;
 import org.springframework.cloud.deployer.spi.kubernetes.KubernetesTaskLauncher;
+import org.springframework.cloud.deployer.spi.kubernetes.KubernetesTaskLauncherProperties;
+import org.springframework.cloud.deployer.spi.kubernetes.RestartPolicy;
 import org.springframework.test.util.ReflectionTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author David Turanski
+ * @author Ilayaperumal Gopinathan
  **/
 public class KubernetesTaskPlatformFactoryTests {
 
 	@Test
 	public void kubernetesTaskPlatformNoScheduler() {
 		KubernetesPlatformProperties platformProperties = new KubernetesPlatformProperties();
 		KubernetesDeployerProperties deployerProperties = new KubernetesDeployerProperties();
+		KubernetesTaskLauncherProperties taskLauncherProperties = new KubernetesTaskLauncherProperties();
 		platformProperties.setAccounts(Collections.singletonMap("k8s", deployerProperties));
-
+		KubernetesPlatformTaskLauncherProperties platformTaskLauncherProperties = new KubernetesPlatformTaskLauncherProperties();
+		platformTaskLauncherProperties.setAccounts(Collections.singletonMap("k8s", taskLauncherProperties));
 		KubernetesTaskPlatformFactory kubernetesTaskPlatformFactory = new KubernetesTaskPlatformFactory(
-				platformProperties, false);
+				platformProperties, false, platformTaskLauncherProperties);
 
 		TaskPlatform taskPlatform = kubernetesTaskPlatformFactory.createTaskPlatform();
 		assertThat(taskPlatform.getName()).isEqualTo("Kubernetes");
@@ -60,10 +65,14 @@ public void kubernetesTaskPlatformWithScheduler() {
 		KubernetesPlatformProperties platformProperties = new KubernetesPlatformProperties();
 		KubernetesDeployerProperties deployerProperties = new KubernetesDeployerProperties();
 		deployerProperties.getLimits().setMemory("5555Mi");
+		KubernetesTaskLauncherProperties taskLauncherProperties = new KubernetesTaskLauncherProperties();
+		taskLauncherProperties.setBackoffLimit(5);
+		taskLauncherProperties.setRestartPolicy(RestartPolicy.Never);
 		platformProperties.setAccounts(Collections.singletonMap("k8s", deployerProperties));
-
+		KubernetesPlatformTaskLauncherProperties platformTaskLauncherProperties = new KubernetesPlatformTaskLauncherProperties();
+		platformTaskLauncherProperties.setAccounts(Collections.singletonMap("k8s", taskLauncherProperties));
 		KubernetesTaskPlatformFactory kubernetesTaskPlatformFactory = new KubernetesTaskPlatformFactory(
-				platformProperties, true);
+				platformProperties, true, platformTaskLauncherProperties);
 
 		TaskPlatform taskPlatform = kubernetesTaskPlatformFactory.createTaskPlatform();
 		assertThat(taskPlatform.getName()).isEqualTo("Kubernetes");
@@ -80,6 +89,11 @@ public void kubernetesTaskPlatformWithScheduler() {
 
 			+ "\\[.+\\], api version = \\[.+\\]$");
 
+		KubernetesTaskLauncherProperties taskLauncherProps = (KubernetesTaskLauncherProperties) ReflectionTestUtils.getField(taskLauncher.getTaskLauncher(), "taskLauncherProperties");
+
+		assertThat(taskLauncherProps.getBackoffLimit()).isEqualTo(5);
+		assertThat(taskLauncherProps.getRestartPolicy()).isEqualTo(RestartPolicy.Never);
+
 	}
 
 }