Could not load the task's invokable class. "Third time lucky"

[Saleh]

Hi everyone once more. I am trying to run a Java flink streaming job (basic word count with Kafka) locally. But I am stuck with the following error. Any ideas on the cause of the error?

Job execution switched to status RUNNING.
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(1/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(1/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(2/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(2/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(3/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(3/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(4/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(4/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(5/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(5/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(6/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(6/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(7/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(7/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(8/8) switched to SCHEDULED
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(8/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(1/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(1/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(2/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(2/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(3/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(3/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(4/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(4/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(5/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(5/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(6/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(6/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(7/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(7/8) switched to DEPLOYING
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(8/8) switched to SCHEDULED
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(8/8) switched to DEPLOYING
10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(5/8) switched to FAILED
java.lang.Exception: Could not load the task's invokable class.
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:696)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:450)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: class org.apache.flink.streaming.runtime.tasks.SourceStreamTask
        at java.lang.Class.asSubclass(Class.java:3404)
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:693)
        ... 2 more

10/29/2015 11:15:54 Custom Source -> Flat Map -> Map(2/8) switched to FAILED
java.lang.Exception: Could not load the task's invokable class.
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:696)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:450)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: class org.apache.flink.streaming.runtime.tasks.SourceStreamTask
        at java.lang.Class.asSubclass(Class.java:3404)
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:693)
        ... 2 more

10/29/2015 11:15:54 Job execution switched to status FAILING.
10/29/2015 11:15:54 Grouped Aggregation -> (Map -> Stream Sink, Stream Sink)(1/8) switched to FAILED
java.lang.Exception: Could not load the task's invokable class.
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:696)
        at org.apache.flink.runtime.taskmanager.Task.run(Task.java:450)
        at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: class org.apache.flink.streaming.runtime.tasks.OneInputStreamTask
        at java.lang.Class.asSubclass(Class.java:3404)
        at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:693)
        ... 2 more