You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by zh...@apache.org on 2020/01/22 03:55:11 UTC
[flink] 04/04: [FLINK-15632][kubernetes] Update the cluster id and
high-availability jobmanager port when HA enabled
This is an automated email from the ASF dual-hosted git repository.
zhuzh pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e1436f8172d61d610d892a3f4e24ab137d3e03b6
Author: wangyang0918 <da...@alibaba-inc.com>
AuthorDate: Tue Jan 21 11:45:23 2020 +0800
[FLINK-15632][kubernetes] Update the cluster id and high-availability jobmanager port when HA enabled
When HA enabled, the config option HighAvailabilityOptions.HA_CLUSTER_ID will be set to the user specified cluster id. And if the config option HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE is not set or with value 0, it will be updated to the value of JobManagerOptions.PORT.
---
.../kubernetes/KubernetesClusterDescriptor.java | 28 ++++++++++------------
.../flink/kubernetes/utils/KubernetesUtils.java | 23 ++++++++++++++++++
.../flink/kubernetes/KubernetesUtilsTest.java | 21 ++++++++++++++++
3 files changed, 57 insertions(+), 15 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
index 9b5c73e..acef01c 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/KubernetesClusterDescriptor.java
@@ -158,26 +158,24 @@ public class KubernetesClusterDescriptor implements ClusterDescriptor<String> {
flinkConfig.setString(KubernetesConfigOptionsInternal.ENTRY_POINT_CLASS, entryPoint);
// Rpc, blob, rest, taskManagerRpc ports need to be exposed, so update them to fixed values.
- if (KubernetesUtils.parsePort(flinkConfig, BlobServerOptions.PORT) == 0) {
- flinkConfig.setString(BlobServerOptions.PORT, String.valueOf(Constants.BLOB_SERVER_PORT));
- LOG.warn(
- "Kubernetes service requires a fixed port. Configuration {} will be set to {}",
- BlobServerOptions.PORT.key(),
- Constants.BLOB_SERVER_PORT);
- }
-
- if (KubernetesUtils.parsePort(flinkConfig, TaskManagerOptions.RPC_PORT) == 0) {
- flinkConfig.setString(TaskManagerOptions.RPC_PORT, String.valueOf(Constants.TASK_MANAGER_RPC_PORT));
- LOG.warn(
- "Kubernetes service requires a fixed port. Configuration {} will be set to {}",
- TaskManagerOptions.RPC_PORT.key(),
- Constants.TASK_MANAGER_RPC_PORT);
- }
+ KubernetesUtils.checkAndUpdatePortConfigOption(flinkConfig, BlobServerOptions.PORT, Constants.BLOB_SERVER_PORT);
+ KubernetesUtils.checkAndUpdatePortConfigOption(
+ flinkConfig,
+ TaskManagerOptions.RPC_PORT,
+ Constants.TASK_MANAGER_RPC_PORT);
// Set jobmanager address to namespaced service name
final String nameSpace = flinkConfig.getString(KubernetesConfigOptions.NAMESPACE);
flinkConfig.setString(JobManagerOptions.ADDRESS, clusterId + "." + nameSpace);
+ if (HighAvailabilityMode.isHighAvailabilityModeActivated(flinkConfig)) {
+ flinkConfig.setString(HighAvailabilityOptions.HA_CLUSTER_ID, clusterId);
+ KubernetesUtils.checkAndUpdatePortConfigOption(
+ flinkConfig,
+ HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE,
+ flinkConfig.get(JobManagerOptions.PORT));
+ }
+
try {
final KubernetesService internalSvc = client.createInternalService(clusterId).get();
// Update the service id in Flink config, it will be used for gc.
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
index 0a8d8da..0f108b7 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/utils/KubernetesUtils.java
@@ -36,6 +36,8 @@ import io.fabric8.kubernetes.api.model.ResourceRequirementsBuilder;
import io.fabric8.kubernetes.api.model.Volume;
import io.fabric8.kubernetes.api.model.VolumeMount;
import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
@@ -62,6 +64,8 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
*/
public class KubernetesUtils {
+ private static final Logger LOG = LoggerFactory.getLogger(KubernetesUtils.class);
+
/**
* Read file content to string.
*
@@ -86,6 +90,25 @@ public class KubernetesUtils {
}
/**
+ * Check whether the port config option is a fixed port. If not, the fallback port will be set to configuration.
+ * @param flinkConfig flink configuration
+ * @param port config option need to be checked
+ * @param fallbackPort the fallback port that will be set to the configuration
+ */
+ public static void checkAndUpdatePortConfigOption(
+ Configuration flinkConfig,
+ ConfigOption<String> port,
+ int fallbackPort) {
+ if (KubernetesUtils.parsePort(flinkConfig, port) == 0) {
+ flinkConfig.setString(port, String.valueOf(fallbackPort));
+ LOG.info(
+ "Kubernetes deployment requires a fixed port. Configuration {} will be set to {}",
+ port.key(),
+ fallbackPort);
+ }
+ }
+
+ /**
* Parse a valid port for the config option. A fixed port is expected, and do not support a range of ports.
*
* @param flinkConfig flink config
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesUtilsTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesUtilsTest.java
index 160a429..6bd106f 100644
--- a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesUtilsTest.java
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/KubernetesUtilsTest.java
@@ -24,6 +24,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
+import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.utils.KubernetesUtils;
@@ -238,6 +239,26 @@ public class KubernetesUtilsTest extends TestLogger {
}
}
+ @Test
+ public void testCheckWithDynamicPort() {
+ testCheckAndUpdatePortConfigOption("0", "6123", "6123");
+ }
+
+ @Test
+ public void testCheckWithFixedPort() {
+ testCheckAndUpdatePortConfigOption("6123", "16123", "6123");
+ }
+
+ private void testCheckAndUpdatePortConfigOption(String port, String fallbackPort, String expectedPort) {
+ final Configuration cfg = new Configuration();
+ cfg.setString(HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE, port);
+ KubernetesUtils.checkAndUpdatePortConfigOption(
+ cfg,
+ HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE,
+ Integer.valueOf(fallbackPort));
+ assertEquals(expectedPort, cfg.get(HighAvailabilityOptions.HA_JOB_MANAGER_PORT_RANGE));
+ }
+
private String getJobManagerExpectedCommand(String jvmAllOpts, String logging, String mainClassArgs) {
return java + " " + classpath + " " + jmJvmMem +
(jvmAllOpts.isEmpty() ? "" : " " + jvmAllOpts) +