How to debug Flink itself using Intellij IDEA

Question

I want to study stream principles through the flink source code, I need to debug it and see how it works I've compiled the Flink source code and imported the project to Intellij IEDA, but I cannot run it as a standalone cluster. I am working on Windows 10 with JDK 11 and maven 3.6.3 I set the main class as

org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint

and run it from IDEA, but I got some warnings and an excetion

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.hadoop.security.authentication.util.KerberosUtil (file:/C:/Users/%e5%88%98%e8%bf%9c%e5%8d%9a/.m2/repository/org/apache/hadoop/hadoop-auth/2.8.5/hadoop-auth-2.8.5.jar) to method sun.security.krb5.Config.getInstance()
WARNING: Please consider reporting this to the maintainers of org.apache.hadoop.security.authentication.util.KerberosUtil
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release

Exception in thread "main" java.lang.NullPointerException
    at java.base/java.util.Objects.requireNonNull(Objects.java:221)
    at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.shutDownAsync(ClusterEntrypoint.java:586)
    at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.startCluster(ClusterEntrypoint.java:242)
    at org.apache.flink.runtime.entrypoint.ClusterEntrypoint.runClusterEntrypoint(ClusterEntrypoint.java:729)
    at org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint.main(StandaloneSessionClusterEntrypoint.java:59)

How can I start JobManager and debug it?

I try to start a JobManager and then a TaskManager through Flink sourc code using IDEA. And I wish I could debug them.

Answer 1

The NPE you've mentioned is thrown on shutdown hook and is hiding the original exception. The original exception is NoClassDefFoundError for log4j classes. That's because these dependencies have the provided scope. See https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/flinkdev/ide_setup/#examples-fail-with-a-noclassdeffounderror-for-flink-classes. In order to fix it you need to change log4j dependencies' scope in flink-parent pom.xml:

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-api</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
        </dependency>

        <dependency>
            <!-- API bridge between log4j 1 and 2 -->
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-1.2-api</artifactId>
        </dependency>

To run jobmanager:

main class:

org.apache.flink.runtime.entrypoint.StandaloneSessionClusterEntrypoint

arguments:

-c flink-dist\target\flink-{version}-bin\flink-{version}\conf

VM options (optional, enables logs):

-Dlog4j.configurationFile={source_code_path}/flink-dist/target/flink-{version}-bin/flink-{version}/conf/log4j-console.properties

To run taskmanager:

main class:

org.apache.flink.runtime.taskexecutor.TaskManagerRunner

arguments:

-c flink-dist\target\flink-{version}-bin\flink-{version}\conf

VM options:

-Dlog4j.configurationFile={source_code_path}/flink-dist/target/flink-{version}-bin/flink-{version}/conf/log4j-console.properties
-Dtaskmanager.memory.network.min=134217730b
-Dtaskmanager.cpu.cores=10.0
-Dtaskmanager.memory.task.off-heap.size=0b
-Dtaskmanager.memory.jvm-metaspace.size=268435456b
-Dtaskmanager.memory.jvm-overhead.min=201326592b
-Dtaskmanager.memory.framework.off-heap.size=134217728b
-Dtaskmanager.memory.network.max=134217730b
-Dtaskmanager.memory.framework.heap.size=134217728b
-Dtaskmanager.memory.managed.size=536870920b
-Dtaskmanager.memory.task.heap.size=402653174b
-Dtaskmanager.numberOfTaskSlots=10
-Dtaskmanager.memory.jvm-overhead.max=201326592b

If this doesn't work for some reason, see flink-dist/src/main/flink-bin/bin/jobmanager.sh and flink-dist/src/main/flink-bin/bin/taskmanager.sh to find the actual params that are passed on a startup.

Answer 2

This problem is solved by configing VM opitons and Program arguments. With Windows10 the VM options are:

-Dlog4j.configuration=file:{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\conf\my-log4j-console.properties 
-classpath {source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-cep-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-connector-files-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-csv-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-dist-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-json-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-scala_2.12-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-shaded-zookeeper-3.5.9.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-table-api-java-uber-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-table-planner-loader-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\flink-table-runtime-1.16.0.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\log4j-core-2.17.1.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\log4j-slf4j-impl-2.17.1.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\log4j-1.2-api-2.17.1.jar;{source_code_path}\flink-1.16.0\flink-dist\target\flink-1.16.0-bin\flink-1.16.0\lib\log4j-api-2.17.1.jar;

And the Program arguments is

-c flink-dist\target\flink-1.16.0-bin\flink-1.16.0\conf

The target path is a little bit different from those on Liunx

Answer 3

Just write a simple workflow (or use one of the many examples provided on the Flink website), and execute it using a LocalStreamEnvironment. That will automatically start a JobManager & TaskManager, as part of the Flink MiniCluster.