Как заставить простой DAG работать в Hazelcast Jet?

Работая над своим DAG в hazelcast Jet, я столкнулся со странной проблемой. Чтобы проверить ошибку, я полностью упростил свой подход и: кажется, что края не работают в соответствии с учебником.

Приведенный ниже код почти настолько же прост, насколько это возможно. Две вершины (один источник, один сток), одно ребро.

Источник читает с карты, приемник должен помещать в карту.

data.addEntryListener правильно говорит мне, что карта заполнена 100 списками (каждый с 25 объектами по 400 байт) другим приложением... и ничего. Карта заполняется, но даг с ней вообще не взаимодействует.

Есть идеи, где искать проблему?

package be.andersch.clusterbench;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.hazelcast.config.Config;
import com.hazelcast.config.SerializerConfig;
import com.hazelcast.core.EntryEvent;
import com.hazelcast.jet.*;
import com.hazelcast.jet.config.JetConfig;
import com.hazelcast.jet.stream.IStreamMap;
import com.hazelcast.map.listener.EntryAddedListener;
import be.andersch.anotherpackage.myObject;

import java.util.List;
import java.util.concurrent.ExecutionException;

import static com.hazelcast.jet.Edge.between;
import static com.hazelcast.jet.Processors.*;

/**
 * Created by abernard on 24.03.2017.
 */
public class Analyzer {
    private static final ObjectMapper mapper = new ObjectMapper();
    private static JetInstance jet;
    private static final IStreamMap<Long, List<String>> data;
    private static final IStreamMap<Long, List<String>> testmap;

    static {
        JetConfig config = new JetConfig();
        Config hazelConfig = config.getHazelcastConfig();
        hazelConfig.getGroupConfig().setName( "name" ).setPassword( "password" );
        hazelConfig.getNetworkConfig().getInterfaces().setEnabled( true ).addInterface( "my_IP_range_here" );
        hazelConfig.getSerializationConfig().getSerializerConfigs().add(
                new SerializerConfig().
                        setTypeClass(myObject.class).
                        setImplementation(new OsamKryoSerializer()));
        jet = Jet.newJetInstance(config);
        data = jet.getMap("data");
        testmap = jet.getMap("testmap");
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        DAG dag = new DAG();
        Vertex source = dag.newVertex("source", readMap("data"));
        Vertex test = dag.newVertex("test", writeMap("testmap"));

        dag.edge(between(source, test));

        jet.newJob(dag).execute()get();

        data.addEntryListener((EntryAddedListener<Long, List<String>>) (EntryEvent<Long, List<String>> entryEvent) -> {
            System.out.println("Got data: " + entryEvent.getKey() + " at " + System.currentTimeMillis() + ", Size: " + jet.getHazelcastInstance().getMap("data").size());
        }, true);

        testmap.addEntryListener((EntryAddedListener<Long, List<String>>) (EntryEvent<Long, List<String>> entryEvent) -> {
            System.out.println("Got test: " + entryEvent.getKey() + " at " + System.currentTimeMillis());
        }, true);

        Runtime.getRuntime().addShutdownHook(new Thread(() -> Jet.shutdownAll()));
    }
}