dump same keytab from shipfile
@@ -26,6 +26,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.flink.client.deployment.ClusterSpecification;
import org.apache.flink.client.program.ClusterClientProvider;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.SecurityOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.yarn.YarnClusterDescriptor;
@@ -42,6 +43,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
+import java.util.stream.Stream;
/**
@@ -83,6 +85,8 @@ public class YarnJobClusterExecutor {
}
}
+ dumpSameKeytab(flinkConfiguration, shipFiles);
+
clusterDescriptor.addShipFiles(shipFiles);
ClusterSpecification clusterSpecification = YarnClusterClientFactory.INSTANCE.getClusterSpecification(flinkConfiguration);
@@ -94,6 +98,14 @@ public class YarnJobClusterExecutor {
LOG.info(String.format("deploy per_job with appId: %s, jobId: %s", applicationId, flinkJobId));
}
+ private void dumpSameKeytab(Configuration flinkConfiguration, List<File> shipFiles) {
+ String flinkKeytab = Stream
+ .of(flinkConfiguration.getString(SecurityOptions.KERBEROS_LOGIN_KEYTAB).split(File.separator))
+ .reduce((a, b) -> b)
+ .get();
+ shipFiles.removeIf(f -> f.getName().equals(flinkKeytab));
+ }
+
private void appendApplicationConfig(Configuration flinkConfig, JobParamsInfo jobParamsInfo) {
if (!StringUtils.isEmpty(jobParamsInfo.getName())) {
flinkConfig.setString(YarnConfigOptions.APPLICATION_NAME, jobParamsInfo.getName());
