[FLINK-36264] fix can't setup rest-service-exposed-type

Author: 陈宇智

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilder.java

@@ -243,7 +243,8 @@ protected FlinkConfigBuilder applyPodTemplate() throws IOException {
 
     protected FlinkConfigBuilder applyIngressDomain() {
         // Web UI
-        if (spec.getIngress() != null) {
+        if (spec.getIngress() != null
+                && effectiveConfig.getOptional(REST_SERVICE_EXPOSED_TYPE).isEmpty()) {
             effectiveConfig.set(
                     REST_SERVICE_EXPOSED_TYPE,
                     KubernetesConfigOptions.ServiceExposedType.ClusterIP);

flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigBuilderTest.java

@@ -67,6 +67,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import static org.apache.flink.configuration.DeploymentOptions.SHUTDOWN_ON_APPLICATION_FINISH;
 import static org.apache.flink.kubernetes.operator.api.utils.BaseTestUtils.IMAGE;
@@ -381,14 +382,27 @@ private PodTemplateSpec getTmPod(Configuration configuration) throws IOException
                 PodTemplateSpec.class);
     }
 
-    @Test
-    public void testApplyIngressDomain() {
+    @ParameterizedTest
+    @MethodSource("serviceExposedTypes")
+    public void testApplyIngressDomain(
+            KubernetesConfigOptions.ServiceExposedType serviceExposedType) {
+
         final Configuration configuration =
-                new FlinkConfigBuilder(flinkDeployment, new Configuration())
+                new FlinkConfigBuilder(
+                                flinkDeployment,
+                                serviceExposedType == null
+                                        ? new Configuration()
+                                        : new Configuration()
+                                                .set(
+                                                        KubernetesConfigOptions
+                                                                .REST_SERVICE_EXPOSED_TYPE,
+                                                        serviceExposedType))
                         .applyIngressDomain()
                         .build();
         assertEquals(
-                KubernetesConfigOptions.ServiceExposedType.ClusterIP,
+                serviceExposedType == null
+                        ? KubernetesConfigOptions.ServiceExposedType.ClusterIP
+                        : serviceExposedType,
                 configuration.get(KubernetesConfigOptions.REST_SERVICE_EXPOSED_TYPE));
     }
 
@@ -961,4 +975,13 @@ private PodTemplateSpec createTestPodWithContainers() {
                 TestUtils.getTestPodTemplate("hostname", List.of(mainContainer, sideCarContainer));
         return pod;
     }
+
+    private static Stream<KubernetesConfigOptions.ServiceExposedType> serviceExposedTypes() {
+        return Stream.of(
+                null,
+                KubernetesConfigOptions.ServiceExposedType.ClusterIP,
+                KubernetesConfigOptions.ServiceExposedType.LoadBalancer,
+                KubernetesConfigOptions.ServiceExposedType.Headless_ClusterIP,
+                KubernetesConfigOptions.ServiceExposedType.NodePort);
+    }
 }