Test.java

// CHECK OUTPUT ON https://gist.github.com/796ee05425535ece1736df7b1e884cce

package com.myproj;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.ignite.Ignite;
import org.apache.ignite.Ignition;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteRunnable;

public class Test {

    public Test() {
        System.out.println("~~~ Constructor called!");
    }

    public static void main(String[] args) throws Exception {
        IgniteRunnable runnable = new IgniteRunnable() {
            @Override
            public void run() {
                try {
                    ClassLoader classLoader = getClass().getClassLoader();
                    Class<?> test = classLoader.loadClass("com.myproj.Test");
                    test.newInstance(); // <-- Works well here (prints "Constructor called!")

                    StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment(1);
                    environment.addSource(new SourceFunction<Object>() {
                        @Override
                        public void run(SourceContext<Object> ctx) throws Exception {

                        }

                        @Override
                        public void cancel() {

                        }
                    })
                    .map(new MapFunction<Object, Object>() {
                        @Override
                        public Object map(Object value) throws Exception {
                            return null;
                        }
                    });

                    environment.execute(); // Crashes here with "Cannot load user class: com.myproj.Test"
                } catch (Exception e) {
                    System.out.println("ERROR #####");
                    System.out.println(e.getMessage());
                    e.printStackTrace();
                }
            }
        };

        IgniteConfiguration config = new IgniteConfiguration()
                .setPeerClassLoadingEnabled(true)
                .setClientMode(true);

        Ignite ignite = Ignition.start(config);
        ignite.compute().broadcast(runnable); // Executes the runnable on all Ignite nodes
    }

}