Hi there,
I faced with issue in adding file to distributed cache in Flink.
My setup:
- Java 1.8
- Flink 1.8
- OS: Windows, Linux
Test scenario:
1. Create simple stream environment
2. Add to distributed cache local file
3. Add simple source function and sink
4. Execute job from Flink CLI (Windows/Linux)
In order to restore job from savepoint or from checkpoint we need to run our
job from Flink CLI. And pipelines that have distributed cache fails their
execution.
Moreover it is different in Linux and Windows systems: in Windows we get
"java.nio.file.InvalidPathException: Illegal char <:> at index 4" and on Linux
we have our Flink freezing (it just stuck and do not do anything, no any error
message or stacktrace).
My piece of code for windows environment:
public class CachePipeline {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment see =
StreamExecutionEnvironment.getExecutionEnvironment();
see.registerCachedFile("file:///D:/test.csv", "MyFile");
see.addSource(new SourceFunction<Integer>() {
@Override
public void run(SourceContext<Integer> ctx) throws Exception {
while(true){
synchronized(ctx.getCheckpointLock()){
ctx.collect(5);
}
Thread.sleep(1000);
}
}
@Override
public void cancel() {}
}).print();
see.execute();
}
}
command for running job that I used for:
flink run -c test.CachePipeline D:\path\to\jar\cache-test.jar
In case with Linux OS I changed file location to:
see.registerCachedFile("file:///home/test.csv", "MyFile");
Windows stacktrace:
flink run -c com.CachePipeline D:\repository\cache-test.jar
log4j:WARN No appenders could be found for logger
(org.apache.flink.client.cli.CliFrontend).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more
info.
Starting execution of program
------------------------------------------------------------
The program finished with the following exception:
org.apache.flink.client.program.ProgramInvocationException: Could not retrieve
the execution result. (JobID: 38631d859b64cd86201bbe09a32c62f3)
at
org.apache.flink.client.program.rest.RestClusterClient.submitJob(RestClusterClient.java:261)
at
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:483)
at
org.apache.flink.streaming.api.environment.StreamContextEnvironment.execute(StreamContextEnvironment.java:66)
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1509)
at com.granduke.teleprocessing.CachePipeline.main(CachePipeline.java:29)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
at java.lang.reflect.Method.invoke(Unknown Source)
at
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529)
at
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421)
at
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:423)
at
org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:813)
at
org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:287)
at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:213)
at
org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1050)
at
org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1126)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Unknown Source)
at
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
at
org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1126)
Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed to
submit JobGraph.
at
org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$8(RestClusterClient.java:388)
at java.util.concurrent.CompletableFuture.uniExceptionally(Unknown
Source)
at
java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(Unknown Source)
at java.util.concurrent.CompletableFuture.postComplete(Unknown Source)
at java.util.concurrent.CompletableFuture$AsyncSupply.run(Unknown
Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)
Caused by: java.nio.file.InvalidPathException: Illegal char <:> at index 4:
file:/D:/test.csv
at sun.nio.fs.WindowsPathParser.normalize(Unknown Source)
at sun.nio.fs.WindowsPathParser.parse(Unknown Source)
at sun.nio.fs.WindowsPathParser.parse(Unknown Source)
at sun.nio.fs.WindowsPath.parse(Unknown Source)
at sun.nio.fs.WindowsFileSystem.getPath(Unknown Source)
at java.nio.file.Paths.get(Unknown Source)
at
org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$3(RestClusterClient.java:353)
at java.util.concurrent.CompletableFuture.uniApply(Unknown Source)
at java.util.concurrent.CompletableFuture$UniApply.tryFire(Unknown
Source)
... 5 more
1. Also I run on different OS different veriosn of Flink, and have next
results:
· Flink 1.8.0 (Windows/Linux) - not working
· Flink 1.7.2 (Windows/Linux) - not working
· Flink 1.5.6 (Windows/Linux) - not working
· Flink 1.4 (Windows/linux) - works as expected
Did anybody face with this?
