2 node cluster reading file from ftp

[Mohan mohan]

Hi,

Context :

# Started cluster with 2 nodes

     node1 (Master & slave)

     node2 (Slave)

# Source path to CsvInputFormat is like "ftp://test:[hidden email]:5678/source.csv" and DataSet parallelism is 2

# Path to FileOutputFormat is like "ftp://test:[hidden email]:5678/target.csv" and DataSink parallelism is 1

# ExecutionEnvironment.createRemoteEnvironment("x.x.x.x", 6123, "xxx-DEVELOP-SNAPSHOT-fat.jar");

# Flink version 1.4.0

While executing above environment transformation is not successful, below is stacktrace,

org.apache.flink.client.program.ProgramInvocationException: The program execution failed: Job execution failed.

at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:492)

at org.apache.flink.client.program.StandaloneClusterClient.submitJob(StandaloneClusterClient.java:105)

at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:456)

at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:444)

at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:419)

at org.apache.flink.client.RemoteExecutor.executePlanWithJars(RemoteExecutor.java:208)

at org.apache.flink.client.RemoteExecutor.executePlan(RemoteExecutor.java:185)

at org.apache.flink.api.java.RemoteEnvironment.execute(RemoteEnvironment.java:172)

at org.apache.flink.api.java.ExecutionEnvironment.execute(ExecutionEnvironment.java:815)

at com.adaequare.etl2.batch.transform.ETLTransformer.transform(ETLTransformer.java:128)

at etl2.batch.EmployeeSamples.basic(EmployeeSamples.java:91)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:498)

at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)

at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)

at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)

at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)

at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)

at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)

at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)

at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)

at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)

at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)

at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)

at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)

at org.junit.runners.ParentRunner.run(ParentRunner.java:363)

at org.junit.runner.JUnitCore.run(JUnitCore.java:137)

at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)

at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)

at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)

at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.

at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6.apply$mcV$sp(JobManager.scala:897)

at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6.apply(JobManager.scala:840)

at org.apache.flink.runtime.jobmanager.JobManager$$anonfun$handleMessage$1$$anonfun$applyOrElse$6.apply(JobManager.scala:840)

at scala.concurrent.impl.Future$PromiseCompletingRunnable.liftedTree1$1(Future.scala:24)

at scala.concurrent.impl.Future$PromiseCompletingRunnable.run(Future.scala:24)

at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:39)

at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:415)

at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)

at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)

at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)

at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)

Caused by: org.apache.flink.runtime.io.network.partition.PartitionNotFoundException: Partition dcf1cf429b3b4fe14b98b34bdcde3644@f17a5b9a7a9afd093213ca0d05f9ebdc not found.

at org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel.failPartitionRequest(RemoteInputChannel.java:270)

at org.apache.flink.runtime.io.network.partition.consumer.RemoteInputChannel.retriggerSubpartitionRequest(RemoteInputChannel.java:172)

at org.apache.flink.runtime.io.network.partition.consumer.SingleInputGate.retriggerPartitionRequest(SingleInputGate.java:392)

at org.apache.flink.runtime.taskmanager.Task.onPartitionStateUpdate(Task.java:1300)

at org.apache.flink.runtime.taskmanager.Task.lambda$triggerPartitionProducerStateCheck$0(Task.java:1139)

at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760)

at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736)

at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442)

... 6 more

Please suggest what is going wrong.

Thank you.