Получение исключения ClassNotFound в Flink SourceFunction

Спрашивающий уже нашел способ заставить это работать. Я извлек из вопроса соответствующую часть. Отметим, что причина, по которой это произошло, остается невыясненной.

Я не использовал синтаксис кавычек, так как это много текста, но автор вопроса поделился приведенным ниже:

Итак, наконец, я заставил его работать. Я создал объект BlockingQueue внутри SourceFunction (Consumer) и вызвал класс Producer из класса SourceFunction (Consumer) вместо создания BlockingQueue и вызова класса Producer в основном методе программы. и теперь это работает!

Вот мой полный рабочий код во Flink:

public class Main {

public static void main(String[] args) throws Exception {

    final int port, buffer;
    //final String ip;
    try {
        final ParameterTool params = ParameterTool.fromArgs(args);
        port = params.getInt("p");
        buffer = params.getInt("b");
    } catch (Exception e) {
        System.err.println("No port number and/or buffer size specified.");
        return;
    }

    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


    DataStream<Event.MyEvent> stream = env.addSource(new Consumer(port, buffer));
    //DataStream<Event.MyEvent> stream = env.fromCollection(queue);


    Pattern<Event.MyEvent, ?> crashedPattern = Pattern.<Event.MyEvent>begin("start")
            .where(new FilterFunction<Event.MyEvent>() {
                @Override
                public boolean filter(Event.MyEvent myEvent) throws Exception {
                    return (myEvent.getItems().getValue() >= 120);
                }
            })
            .<Event.MyEvent>followedBy("next").where(new FilterFunction<Event.MyEvent>() {
                @Override
                public boolean filter(Event.MyEvent myEvent) throws Exception {
                    return (myEvent.getItems().getValue() <= 10);
                }
            })
            .within(Time.seconds(3));

    PatternStream<Event.MyEvent> crashed = CEP.pattern(stream.keyBy(new KeySelector<Event.MyEvent, String>() {
        @Override
        public String getKey(Event.MyEvent myEvent) throws Exception {
            return myEvent.getEventType();
        }
    }), crashedPattern);

    DataStream<String> alarm = crashed.select(new PatternSelectFunction<Event.MyEvent, String>() {
        @Override
        public String select(Map<String, Event.MyEvent> pattern) throws Exception {
            Event.MyEvent start = pattern.get("start");
            Event.MyEvent next = pattern.get("next");
            return start.getEventType() + " | Speed from " + start.getItems().getValue() + " to " + next.getItems().getValue() + " in 3 seconds\n";
        }
    });

    DataStream<String> rate = alarm.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(1)))
            .apply(new AllWindowFunction<String, String, TimeWindow>() {
                @Override
                public void apply(TimeWindow timeWindow, Iterable<String> iterable, Collector<String> collector) throws Exception {
                    int sum = 0;
                    for (String s: iterable) {
                        sum ++;
                    }
                    collector.collect ("CEP Output Rate: " + sum + "\n");
                }
            });

    rate.writeToSocket(InetAddress.getLocalHost().getHostName(), 7070, new SimpleStringSchema());

    env.execute("Flink Taxi Crash Streaming");
}

private static class Producer implements Runnable {

    Boolean running = true;
    Socket socket = null, bufferSocket = null;
    PrintStream ps = null;
    BlockingQueue<Event.MyEvent> queue;
    final int port;

    Producer(BlockingQueue<Event.MyEvent> queue, int port){
        this.port = port;
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            socket = new Socket("127.0.0.1", port);
            bufferSocket = new Socket(InetAddress.getLocalHost(), 6060);
            ps = new PrintStream(bufferSocket.getOutputStream());
            while (running) {
                queue.put(Event.MyEvent.parseDelimitedFrom(socket.getInputStream()));
                ps.println("Items in Queue: " + queue.size());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

private static class Consumer implements SourceFunction<Event.MyEvent> {

    Boolean running = true;
    final int port;
    BlockingQueue<Event.MyEvent> queue;

    Consumer(int port, int buffer){
        queue = new ArrayBlockingQueue<>(buffer);
        this.port = port;
    }

    @Override
    public void run(SourceContext<Event.MyEvent> sourceContext) {
        try {
            new Thread(new Producer(queue, port)).start();
            while (running) {
                sourceContext.collect(queue.take());
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}