Commit 9df82827 authored by wenmo's avatar wenmo

[fix-214][gateway,common] Failed to submit fink jar when configuration has empty string

parent 5277697c
......@@ -25,10 +25,32 @@ public class Asserts {
return isNull(str) || "".equals(str);
}
public static boolean isAllNullString(String... str) {
boolean isNull = true;
for (String item : str) {
if (isNotNullString(item)) {
isNull = false;
}
}
return isNull;
}
public static boolean isNotNullString(String str) {
return !isNullString(str);
}
public static boolean isAllNotNullString(String... str) {
boolean isNotNull = true;
for (String item : str) {
if (isNullString(item)) {
isNotNull = false;
}
}
return isNotNull;
}
public static boolean isEquals(String str1, String str2) {
if (isNull(str1) && isNull(str2)) {
return true;
......
......@@ -7,6 +7,7 @@ import com.dlink.gateway.exception.GatewayException;
import com.dlink.gateway.result.GatewayResult;
import com.dlink.gateway.result.KubernetesResult;
import com.dlink.utils.LogUtil;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
import org.apache.flink.client.program.ClusterClient;
......@@ -47,7 +48,11 @@ public class KubernetesApplicationGateway extends KubernetesGateway {
AppConfig appConfig = config.getAppConfig();
configuration.set(PipelineOptions.JARS, Collections.singletonList(appConfig.getUserJarPath()));
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(appConfig.getUserJarParas(), appConfig.getUserJarMainAppClass());
String[] userJarParas = appConfig.getUserJarParas();
if (Asserts.isNull(userJarParas)) {
userJarParas = new String[0];
}
ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(userJarParas, appConfig.getUserJarMainAppClass());
KubernetesClusterDescriptor kubernetesClusterDescriptor = new KubernetesClusterDescriptor(configuration, client);
try {
ClusterClientProvider<String> clusterClientProvider = kubernetesClusterDescriptor.deployApplicationCluster(clusterSpecification, applicationConfiguration);
......
......@@ -9,6 +9,7 @@ import com.dlink.gateway.model.JobInfo;
import com.dlink.gateway.result.SavePointResult;
import com.dlink.gateway.result.TestResult;
import com.dlink.utils.LogUtil;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
......@@ -78,10 +79,12 @@ public abstract class KubernetesGateway extends AbstractGateway {
private void addConfigParas(Map<String, String> configMap) {
if (Asserts.isNotNull(configMap)) {
for (Map.Entry<String, String> entry : configMap.entrySet()) {
if (Asserts.isAllNotNullString(entry.getKey(), entry.getValue())) {
this.configuration.setString(entry.getKey(), entry.getValue());
}
}
}
}
public SavePointResult savepointCluster() {
return savepointCluster(null);
......
......@@ -8,6 +8,7 @@ import com.dlink.gateway.exception.GatewayException;
import com.dlink.gateway.result.GatewayResult;
import com.dlink.gateway.result.YarnResult;
import com.dlink.utils.LogUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.deployment.application.ApplicationConfiguration;
......@@ -59,7 +60,11 @@ public class YarnApplicationGateway extends YarnGateway {
AppConfig appConfig = config.getAppConfig();
configuration.set(PipelineOptions.JARS, Collections.singletonList(appConfig.getUserJarPath()));
ClusterSpecification clusterSpecification = new ClusterSpecification.ClusterSpecificationBuilder().createClusterSpecification();
ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(appConfig.getUserJarParas(), appConfig.getUserJarMainAppClass());
String[] userJarParas = appConfig.getUserJarParas();
if (Asserts.isNull(userJarParas)) {
userJarParas = new String[0];
}
ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(userJarParas, appConfig.getUserJarMainAppClass());
YarnClusterDescriptor yarnClusterDescriptor = new YarnClusterDescriptor(
configuration, yarnConfiguration, yarnClient, YarnClientYarnClusterInformationRetriever.create(yarnClient), true);
try {
......
......@@ -9,6 +9,7 @@ import com.dlink.gateway.model.JobInfo;
import com.dlink.gateway.result.SavePointResult;
import com.dlink.gateway.result.TestResult;
import com.dlink.utils.LogUtil;
import org.apache.flink.api.common.JobID;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
......@@ -110,10 +111,12 @@ public abstract class YarnGateway extends AbstractGateway {
private void addConfigParas(Map<String, String> configMap) {
if (Asserts.isNotNull(configMap)) {
for (Map.Entry<String, String> entry : configMap.entrySet()) {
if (Asserts.isAllNotNullString(entry.getKey(), entry.getValue())) {
this.configuration.setString(entry.getKey(), entry.getValue());
}
}
}
}
public SavePointResult savepointCluster() {
return savepointCluster(null);
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment