[FLINK-15656][k8s] Unify the main container name of jobmanager and ta…

…skmanager pod

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecorator.java

@@ -93,7 +93,7 @@ private Container decorateMainContainer(Container container) {
  Collections.emptyMap());
 
  return new ContainerBuilder(container)
- .withName(kubernetesJobManagerParameters.getJobManagerMainContainerName())
+ .withName(Constants.MAIN_CONTAINER_NAME)
  .withImage(kubernetesJobManagerParameters.getImage())
  .withImagePullPolicy(kubernetesJobManagerParameters.getImagePullPolicy().name())
  .withResources(requirements)

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecorator.java

@@ -88,7 +88,7 @@ private Container decorateMainContainer(Container container) {
  kubernetesTaskManagerParameters.getTaskManagerExternalResources());
 
  return new ContainerBuilder(container)
- .withName(kubernetesTaskManagerParameters.getTaskManagerMainContainerName())
+ .withName(Constants.MAIN_CONTAINER_NAME)
  .withImage(kubernetesTaskManagerParameters.getImage())
  .withImagePullPolicy(kubernetesTaskManagerParameters.getImagePullPolicy().name())
  .withResources(resourceRequirements)

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesJobManagerParameters.java

@@ -45,8 +45,6 @@
  */
 public class KubernetesJobManagerParameters extends AbstractKubernetesParameters {
 
- public static final String JOB_MANAGER_MAIN_CONTAINER_NAME = "flink-job-manager";
-
  private final ClusterSpecification clusterSpecification;
 
  public KubernetesJobManagerParameters(
@@ -107,10 +105,6 @@ public Map<String, String> getRestServiceAnnotations() {
  .orElse(Collections.emptyMap());
  }
 
- public String getJobManagerMainContainerName() {
- return JOB_MANAGER_MAIN_CONTAINER_NAME;
- }
-
  public int getJobManagerMemoryMB() {
  return clusterSpecification.getMasterMemoryMB();
  }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/parameters/KubernetesTaskManagerParameters.java

@@ -38,8 +38,6 @@
  */
 public class KubernetesTaskManagerParameters extends AbstractKubernetesParameters {
 
- public static final String TASK_MANAGER_MAIN_CONTAINER_NAME = "flink-task-manager";
-
  private final String podName;
 
  private final String dynamicProperties;
@@ -99,10 +97,6 @@ public List<Map<String, String>> getTolerations() {
  .orElse(Collections.emptyList());
  }
 
- public String getTaskManagerMainContainerName() {
- return TASK_MANAGER_MAIN_CONTAINER_NAME;
- }
-
  public String getPodName() {
  return podName;
  }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/Constants.java

@@ -28,6 +28,8 @@ public class Constants {
  public static final String CONFIG_FILE_LOGBACK_NAME = "logback-console.xml";
  public static final String CONFIG_FILE_LOG4J_NAME = "log4j-console.properties";
 
+ public static final String MAIN_CONTAINER_NAME = "flink-main-container";
+
  public static final String FLINK_CONF_VOLUME = "flink-config-volume";
  public static final String CONFIG_MAP_PREFIX = "flink-config-";
 

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitJobManagerDecoratorTest.java

@@ -21,7 +21,6 @@
 import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
 import org.apache.flink.kubernetes.kubeclient.FlinkPod;
 import org.apache.flink.kubernetes.kubeclient.KubernetesJobManagerTestBase;
-import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
 import org.apache.flink.kubernetes.utils.Constants;
 
 import io.fabric8.kubernetes.api.model.Container;
@@ -103,9 +102,7 @@ public void testApiVersion() {
 
  @Test
  public void testMainContainerName() {
- assertEquals(
- KubernetesJobManagerParameters.JOB_MANAGER_MAIN_CONTAINER_NAME,
- this.resultMainContainer.getName());
+ assertEquals(Constants.MAIN_CONTAINER_NAME, this.resultMainContainer.getName());
  }
 
  @Test

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/InitTaskManagerDecoratorTest.java

@@ -120,9 +120,7 @@ public void testApiVersion() {
 
  @Test
  public void testMainContainerName() {
- assertEquals(
- kubernetesTaskManagerParameters.getTaskManagerMainContainerName(),
- this.resultMainContainer.getName());
+ assertEquals(Constants.MAIN_CONTAINER_NAME, this.resultMainContainer.getName());
  }
 
  @Test

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactoryTest.java

@@ -30,7 +30,6 @@
 import org.apache.flink.kubernetes.kubeclient.decorators.HadoopConfMountDecorator;
 import org.apache.flink.kubernetes.kubeclient.decorators.InternalServiceDecorator;
 import org.apache.flink.kubernetes.kubeclient.decorators.KerberosMountDecorator;
-import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
 import org.apache.flink.kubernetes.utils.Constants;
 import org.apache.flink.kubernetes.utils.KubernetesUtils;
 
@@ -172,9 +171,7 @@ public void testPodSpec() throws IOException {
  assertEquals(3, resultPodSpec.getVolumes().size());
 
  final Container resultedMainContainer = resultPodSpec.getContainers().get(0);
- assertEquals(
- KubernetesJobManagerParameters.JOB_MANAGER_MAIN_CONTAINER_NAME,
- resultedMainContainer.getName());
+ assertEquals(Constants.MAIN_CONTAINER_NAME, resultedMainContainer.getName());
  assertEquals(CONTAINER_IMAGE, resultedMainContainer.getImage());
  assertEquals(
  CONTAINER_IMAGE_PULL_POLICY.name(), resultedMainContainer.getImagePullPolicy());

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactoryTest.java

@@ -21,7 +21,7 @@
 import org.apache.flink.configuration.SecurityOptions;
 import org.apache.flink.kubernetes.KubernetesTestUtils;
 import org.apache.flink.kubernetes.kubeclient.KubernetesTaskManagerTestBase;
-import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesTaskManagerParameters;
+import org.apache.flink.kubernetes.utils.Constants;
 
 import io.fabric8.kubernetes.api.model.Container;
 import io.fabric8.kubernetes.api.model.Pod;
@@ -81,9 +81,7 @@ public void testContainer() {
  assertEquals(1, resultContainers.size());
 
  final Container resultMainContainer = resultContainers.get(0);
- assertEquals(
- KubernetesTaskManagerParameters.TASK_MANAGER_MAIN_CONTAINER_NAME,
- resultMainContainer.getName());
+ assertEquals(Constants.MAIN_CONTAINER_NAME, resultMainContainer.getName());
  assertEquals(CONTAINER_IMAGE, resultMainContainer.getImage());
  assertEquals(CONTAINER_IMAGE_PULL_POLICY.name(), resultMainContainer.getImagePullPolicy());