[FLINK-28829][k8s] Support prepreparing K8S resources before JM creation

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClient.java

@@ -112,6 +112,10 @@ public Fabric8FlinkKubeClient(
     public void createJobManagerComponent(KubernetesJobManagerSpecification kubernetesJMSpec) {
         final Deployment deployment = kubernetesJMSpec.getDeployment();
         final List<HasMetadata> accompanyingResources = kubernetesJMSpec.getAccompanyingResources();
+        final List<HasMetadata> prePreparedResources = kubernetesJMSpec.getPrePreparedResources();
+
+        // before create Deployment
+        this.internalClient.resourceList(prePreparedResources).createOrReplace();
 
         // create Deployment
         LOG.debug(
@@ -121,6 +125,8 @@ public void createJobManagerComponent(KubernetesJobManagerSpecification kubernet
         final Deployment createdDeployment =
                 this.internalClient.apps().deployments().create(deployment);
 
+        // Add all prepared AccompanyingResources to refresh owner reference
+        accompanyingResources.addAll(prePreparedResources);
         // Note that we should use the uid of the created Deployment for the OwnerReference.
         setOwnerReference(createdDeployment, accompanyingResources);
 

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/KubernetesJobManagerSpecification.java

@@ -30,10 +30,15 @@ public class KubernetesJobManagerSpecification {
 
     private List<HasMetadata> accompanyingResources;
 
+    private List<HasMetadata> prePreparedResources;
+
     public KubernetesJobManagerSpecification(
-            Deployment deployment, List<HasMetadata> accompanyingResources) {
+            Deployment deployment,
+            List<HasMetadata> accompanyingResources,
+            List<HasMetadata> prePreparedResources) {
         this.deployment = deployment;
         this.accompanyingResources = accompanyingResources;
+        this.prePreparedResources = prePreparedResources;
     }
 
     public Deployment getDeployment() {
@@ -43,4 +48,8 @@ public Deployment getDeployment() {
     public List<HasMetadata> getAccompanyingResources() {
         return accompanyingResources;
     }
+
+    public List<HasMetadata> getPrePreparedResources() {
+        return prePreparedResources;
+    }
 }

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

@@ -87,4 +87,9 @@ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
     public List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException {
         return Collections.emptyList();
     }
+
+    @Override
+    public List<HasMetadata> buildPrePreparedResources() {
+        return Collections.emptyList();
+    }
 }

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

@@ -43,4 +43,7 @@ public interface KubernetesStepDecorator {
      * feature. This could only be applicable on the client-side submission process.
      */
     List<HasMetadata> buildAccompanyingKubernetesResources() throws IOException;
+
+    /** Build the Kubernetes resources before Flink Job Manager deployment creation. */
+    List<HasMetadata> buildPrePreparedResources();
 }

flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesJobManagerFactory.java

@@ -60,6 +60,7 @@ public static KubernetesJobManagerSpecification buildKubernetesJobManagerSpecifi
             throws IOException {
         FlinkPod flinkPod = Preconditions.checkNotNull(podTemplate).copy();
         List<HasMetadata> accompanyingResources = new ArrayList<>();
+        List<HasMetadata> prePreparedResources = new ArrayList<>();
 
         final KubernetesStepDecorator[] stepDecorators =
                 new KubernetesStepDecorator[] {
@@ -76,14 +77,16 @@ public static KubernetesJobManagerSpecification buildKubernetesJobManagerSpecifi
                 };
 
         for (KubernetesStepDecorator stepDecorator : stepDecorators) {
+            prePreparedResources.addAll(stepDecorator.buildPrePreparedResources());
             flinkPod = stepDecorator.decorateFlinkPod(flinkPod);
             accompanyingResources.addAll(stepDecorator.buildAccompanyingKubernetesResources());
         }
 
         final Deployment deployment =
                 createJobManagerDeployment(flinkPod, kubernetesJobManagerParameters);
 
-        return new KubernetesJobManagerSpecification(deployment, accompanyingResources);
+        return new KubernetesJobManagerSpecification(
+                deployment, accompanyingResources, prePreparedResources);
     }
 
     private static Deployment createJobManagerDeployment(

flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/Fabric8FlinkKubeClientTest.java

@@ -630,4 +630,37 @@ private KubernetesConfigMap buildTestingConfigMap() {
                         .withData(data)
                         .build());
     }
+
+    @Test
+    void testMockPrePreparedResources() throws Exception {
+        // regenerate a test JM spec
+        Deployment testDeployment = kubernetesJobManagerSpecification.getDeployment();
+        List<HasMetadata> testAccompanyingResources =
+                kubernetesJobManagerSpecification.getAccompanyingResources();
+        List<HasMetadata> mockPrePreparedResources = new ArrayList<>();
+        for (int i = 0; i < 3; i++) {
+            Pod mockPod =
+                    new PodBuilder()
+                            .editOrNewMetadata()
+                            .withName("mock-pod-preprepared-resource-" + i)
+                            .endMetadata()
+                            .editOrNewSpec()
+                            .endSpec()
+                            .build();
+            mockPrePreparedResources.add(mockPod);
+        }
+        KubernetesJobManagerSpecification testKubernetesJobManagerSpecification =
+                new KubernetesJobManagerSpecification(
+                        testDeployment, testAccompanyingResources, mockPrePreparedResources);
+        flinkKubeClient.createJobManagerComponent(testKubernetesJobManagerSpecification);
+
+        // check the preprepared resources had been created.
+        final List<Pod> resultPods = kubeClient.pods().inNamespace(NAMESPACE).list().getItems();
+        assertThat(resultPods).hasSize(3);
+
+        final List<Deployment> resultedDeployments =
+                kubeClient.apps().deployments().inNamespace(NAMESPACE).list().getItems();
+        // check resource owner reference had been refreshed.
+        testOwnerReferenceSetting(resultedDeployments.get(0), resultPods);
+    }
 }

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

@@ -468,4 +468,14 @@ void testSetJobManagerDeploymentReplicas() throws Exception {
         assertThat(kubernetesJobManagerSpecification.getDeployment().getSpec().getReplicas())
                 .isEqualTo(JOBMANAGER_REPLICAS);
     }
+
+    @Test
+    void testPrePreparedResourcesWontAffectOriginalProcess() throws IOException {
+        kubernetesJobManagerSpecification =
+                KubernetesJobManagerFactory.buildKubernetesJobManagerSpecification(
+                        flinkPod, kubernetesJobManagerParameters);
+        final List<HasMetadata> prePreparedResources =
+                this.kubernetesJobManagerSpecification.getPrePreparedResources();
+        assertThat(prePreparedResources).hasSize(0);
+    }
 }