No results are written into sink when projected field is nullable

[ncherkas]

Consider the following example, Hazelcast Member:

        Config config = new Config();
        config.getSerializationConfig().addPortableFactory(1, new PortableFactoryImpl());
        HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance(config);

        IMap<String, User> mapTest1 = hazelcastInstance.getMap("test1");
        mapTest1.put("k0", new User("k0", "n0", LocalDate.now()));
        mapTest1.put("k1", new User("k1", "n1", LocalDate.now()));
        mapTest1.put("k2", new User("k2", null, LocalDate.now()));
        mapTest1.put("k3", new User("k3", null, LocalDate.now()));
        mapTest1.put("k4", new User("k4", "n4", LocalDate.now()));
        mapTest1.put("k5", new User("k5", "n5", LocalDate.now()));

        System.out.println("Users count: " + mapTest1.size());

and Jet Member:

        JetInstance jet = Jet.newJetInstance();
        Pipeline p1 = Pipeline.create();

        p1.drawFrom(Sources.<String, ImdgMember.User, String>remoteMap("test1", new ClientConfig(),
// Workaround is to filter out NULLs
//                Predicates.and(Predicates.in("id", "k1", "k2", "k3", "k4"), Predicates.not(Predicates.equal("name", null))),
                Predicates.in("id", "k1", "k2", "k3", "k4"),
                Projections.singleAttribute("name"))).drainTo(Sinks.list("test1"));

        jet.newJob(p1).join();

        IList<String> sink = jet.getList("test1");
        System.out.println("Fetched (count=" + sink.size() + "): " + sink.subList(0, sink.size()));

No items are written to the sink List when some of the matching entries have null as the value of field name.

[viliam-durina]

Problem is that the projectionFn returns null, which might cause the traverser to stop before iterating all items. It should be fixed.

However, jet doesn't allow null items to be emitted, so we'll make it consistent with Processors.mapP, which documents that items mapped to null are filtered out. After this change, the above example should add "n1" and "n4" to the sink.

Also these should be fixed (code and documentation):

• kafka source
• mapJournal source
• Traverser.map()