zhengcanbin commented on a change in pull request #11456: [FLINK-16602][k8s]
Rework the internal/external Service
URL: https://github.com/apache/flink/pull/11456#discussion_r404616847
##########
File path:
flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/InternalServiceDecorator.java
##########
@@ -19,63 +19,67 @@
package org.apache.flink.kubernetes.kubeclient.decorators;
import org.apache.flink.configuration.JobManagerOptions;
-import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import
org.apache.flink.kubernetes.kubeclient.parameters.KubernetesJobManagerParameters;
import org.apache.flink.kubernetes.utils.Constants;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
import io.fabric8.kubernetes.api.model.HasMetadata;
-import io.fabric8.kubernetes.api.model.ServicePort;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
import java.io.IOException;
+import java.util.Collections;
import java.util.List;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
/**
* Creates an internal Service which forwards the requests from the
TaskManager(s) to the
* active JobManager.
* Note that only the non-HA scenario relies on this Service for internal
communication, since
* in the HA mode, the TaskManager(s) directly connects to the JobManager via
IP address.
*/
-public class InternalServiceDecorator extends AbstractServiceDecorator {
+public class InternalServiceDecorator extends AbstractKubernetesStepDecorator {
+
+ private final KubernetesJobManagerParameters
kubernetesJobManagerParameters;
public InternalServiceDecorator(KubernetesJobManagerParameters
kubernetesJobManagerParameters) {
- super(kubernetesJobManagerParameters);
+ this.kubernetesJobManagerParameters =
checkNotNull(kubernetesJobManagerParameters);
}
@Override
public List<HasMetadata> buildAccompanyingKubernetesResources() throws
IOException {
- final String serviceName = getServiceName();
+ if (!kubernetesJobManagerParameters.isInternalServiceEnabled())
{
+ return Collections.emptyList();
+ }
+
+ final String serviceName =
KubernetesUtils.getInternalServiceName(kubernetesJobManagerParameters.getClusterId());
+
+ final Service headlessService = new ServiceBuilder()
+ .withApiVersion(Constants.API_VERSION)
+ .withNewMetadata()
+ .withName(serviceName)
+
.withLabels(kubernetesJobManagerParameters.getCommonLabels())
+ .endMetadata()
+ .withNewSpec()
+ .withClusterIP("None")
Review comment:
> @zhengcanbin does this setting refer to `Headless service`?
Yes.
> Can we move `"None"` into `Constants`?
Fixed by
[1188bfd](https://github.com/apache/flink/pull/11456/commits/1188bfdbd6ecde4c6277b93cea471636852ea9b9)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
