Apache Flink - java.lang.NoClassDefFoundError: org/apache/flink/api/connector/sink2/Sink

Question

I wrote a Flink pipeline that writes the data stream to file in parquet format. I use sinkTo method to write the output to the file. I get the below exception when the application starts.

java.lang.RuntimeException: Could not look up the main(String[]) method from the class com.fk.logs.StreamingJob: org/apache/flink/api/connector/sink2/Sink
    at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:315)
    at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:161)
    at org.apache.flink.client.program.PackagedProgram.<init>(PackagedProgram.java:65)
    at org.apache.flink.client.program.PackagedProgram$Builder.build(PackagedProgram.java:691)
    at org.apache.flink.client.cli.CliFrontend.buildProgram(CliFrontend.java:851)
    at org.apache.flink.client.cli.CliFrontend.getPackagedProgram(CliFrontend.java:271)
    at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:245)
    at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
    at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
    at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
    at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
Caused by: java.lang.NoClassDefFoundError: org/apache/flink/api/connector/sink2/Sink
    at java.base/java.lang.Class.getDeclaredMethods0(Native Method)
    at java.base/java.lang.Class.privateGetDeclaredMethods(Class.java:3166)
    at java.base/java.lang.Class.getMethodsRecursive(Class.java:3307)
    at java.base/java.lang.Class.getMethod0(Class.java:3293)
    at java.base/java.lang.Class.getMethod(Class.java:2106)
    at org.apache.flink.client.program.PackagedProgram.hasMainMethod(PackagedProgram.java:307)
    ... 10 more
Caused by: java.lang.ClassNotFoundException: org.apache.flink.api.connector.sink2.Sink
    at java.base/java.net.URLClassLoader.findClass(URLClassLoader.java:476)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:589)
    at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
    at org.apache.flink.util.ChildFirstClassLoader.loadClassWithoutExceptionHandling(ChildFirstClassLoader.java:65)
    at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
    at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522)

Code for reference

DataStream<String> stream = env.fromSource(getSource(parser), WatermarkStrategy.noWatermarks(),
                "event source").uid("log file event source").filter(f -> !f.isEmpty());
        readLogFile(parser, stream)
                .flatMap((value, out) -> value.lines()
                        .forEach(out::collect), TypeInformation.of(String.class))
                .map(StreamingJob::parseLine).filter(Optional::isPresent)
                .map(Optional::get, mapType)
                .map(lm -> {
                    if (lm.containsKey("msg")) {
                        parseLine(lm.get("msg").toString()).ifPresent(mf -> lm.putAll(mf));
                    }
                    return lm;
                }, mapType)
                .keyBy(m -> m.computeIfAbsent("appid", k -> "unknownapp"))
                .map(m -> newRecord(m))
                .sinkTo(getSink());

Flink 1.15 version was used.

Answer 1

The problem seems to be in the 1.15 version I downloaded and ran my application with. I have clean up with installation and reinstall to run it properly.