dump same keytab from shipfile

Author: simenliuxing

launcher/src/main/java/com/dtstack/flink/sql/launcher/executor/YarnJobClusterExecutor.java

@@ -26,6 +26,7 @@
 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.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Stream;
 
 
 /**
@@ -83,6 +85,8 @@ public void exec() throws Exception {
             }
         }
 
+        dumpSameKeytab(flinkConfiguration, shipFiles);
+
         clusterDescriptor.addShipFiles(shipFiles);
 
         ClusterSpecification clusterSpecification = YarnClusterClientFactory.INSTANCE.getClusterSpecification(flinkConfiguration);
@@ -94,6 +98,14 @@ public void exec() throws Exception {
         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());