'Flink Job submission throws java.nio.file.NoSuchFileException while the file actually exists

I tried to submit a flink job that is already packaged in a JAR. Basically it consumes a kafka topic protected by SASL authentication, thus it requires a .jks file which I already include them in JAR and read in the code as:

try(InputStream resourceStream = loader.getResourceAsStream(configFile)){
        properties.load(resourceStream);
        properties.setProperty("ssl.truststore.location",
            loader.getResource(properties.getProperty("ssl.truststore.location")).toURI().getPath());
      }

catch(Exception e){
        System.out.println("Failed to load config");
      }

I tried to submit the job on two different (different VM specs) standalone server for the sake of testing. One server runs succesfully, but another throw a java.nio.file.NoSuchFileException, saying that my .jks file is not found. Can someone please point out the possible issue on it?

Here, the flink is deployed on a standalone cluster mode with the following version:

  • Flink version: 1.14.0
  • Java version: 11.0.13
javaapache-flink


Solution 1:[1]

I realize my question was really silly. This part actually returns null and trigger exception.

loader.getResource(properties.getProperty("ssl.truststore.location")).toURI().getPath()

The problem was that I submit the job through web UI thus I couldn't see the printed message. Thus, the filename resolves to the original one stored under the configFile, which is a relative path. Why one machine works and another one doesn't? Cause I previously somehow has the .jks on my homedir for another testing :).

For others to not jump into this mistake, here is the summary of what will .getResource() resolve if run from IDE (gradle run task) and jar, respectively.

//      file:home/gradle-demo/build/resources/main/kafka-client.truststore.jks
//      jar:file:home/gradle-demo/build/libs/gradle-demo-1.0-SNAPSHOT.jar!/kafka-client.truststore.jks
      System.out.println(loader.getResource("kafka-client.trustore.jks").toString());

//      home/gradle-demo/build/resources/main/kafka-client.truststore.jks
//      file:home/gradle-demo/build/libs/gradle-demo-1.0-SNAPSHOT.jar!/kafka-client.truststore.jks
      System.out.println(loader.getResource("kafka-client.trustore.jks").getPath());

//      home/gradle-demo/build/resources/main/kafka-client.truststore.jks
//      null
      System.out.println(loader.getResource("kafka-client.trustore.jks").toURI().getPath());

//      file:home/gradle-demo/build/resources/main/kafka-client.truststore.jks
//      jar:file:home/gradle-demo/build/libs/gradle-demo-1.0-SNAPSHOT.jar!/kafka-client.truststore.jks
      System.out.println(loader.getResource("kafka-client.trustore.jks").toURI());

Solution 2:[2]

kafka-client:2.4.1 org.apache.kafka.common.security.ssl.SslEngineBuilder#285

try (InputStream in = Files.newInputStream(Paths.get(path))) {
    KeyStore ks = KeyStore.getInstance(type);
    // If a password is not set access to the truststore is still available, but integrity checking is disabled.
    char[] passwordChars = password != null ? password.value().toCharArray() : null;
    ks.load(in, passwordChars);
    return ks;
  } catch (GeneralSecurityException | IOException e) {
    throw new KafkaException("Failed to load SSL keystore " + path + " of type " + type, e);
}

It looks like we should put jks file in file system(nfs or hdfs) where task manager can access by absolute path.

Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source
Solution 1
Solution 2 Leo

Related Questions