TestSources serialization failure for Tuple2<String, String>

[neilstevenson]

On Jet 4.4-SNAPSHOT, readFrom(TestSources.items(List.of(Tuple2.tuple2("hello", "world")))) gives

java.lang.IllegalArgumentException: "fillBufferFn" must be serializable
	at neil.demo.MyThingTest.test(MyThingTest.java:43)
Caused by: java.io.NotSerializableException: com.hazelcast.jet.datamodel.Tuple2
	at neil.demo.MyThingTest.test(MyThingTest.java:43)

Full reproducer

package neil.demo;

import java.util.List;

import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;

import com.hazelcast.jet.Jet;
import com.hazelcast.jet.JetInstance;
import com.hazelcast.jet.datamodel.Tuple2;
import com.hazelcast.jet.pipeline.Pipeline;
import com.hazelcast.jet.pipeline.Sinks;
import com.hazelcast.jet.pipeline.test.TestSources;

public class MyThingTest {

    private static JetInstance jetInstance;

    @BeforeAll
    public static void beforeAll() throws Exception {
        jetInstance = Jet.newJetInstance();
    }

    @AfterAll
    public static void afterAll() {
        jetInstance.shutdown();
    }
    
    @Test
    public void test() {
        Object item1 = "hello";
        Object item2 = Tuple2.tuple2("hello", "world");
        
        List<Object> input =
                List.of(item1);
        // This line breaks it
        input = List.of(item2);
        
        Pipeline pipeline = Pipeline.create();

        pipeline
        .readFrom(TestSources.items(input))
        .writeTo(Sinks.logger());

        jetInstance.newJob(pipeline).join();
    }

}

Could be related, hazelcast/hazelcast#17788, if IMDG is trying to deserialize the Java list, IMDG may not know about Tuple2 which is a Jet class.

[viliam-durina]

This is because the Tuple2 isn't java-serializable. We serialize the processor suppliers using the java serialization. If a java-serialized object (the processor supplier) refers to an object that's not java-serializable, it will fail, even if the object is hz-serializable.

Use Map.Entry (see Util.entry()) in your case, but it's not a generic answer. The generic answer currently is that we don't support it.

We could add both serialization methods (java and hz), but we're reluctant to do this because it would hide issues where java serialization is used and hz-serialization could be used.

Rest of the team can have an opinion on this.

[neilstevenson]

My actual code uses a Tuple4, I just used a Tuple2 to simplify

[neilstevenson]

This means, for example, if you had a <String, String> input stream for first name and last name, and you wanted to enrich with salutation

        .mapUsingIMap("salutation",
                Tuple2::f1,
                (tuple2, salutation) ->
                    Tuple3.tuple3(tuple2.f0(), tuple2.f1(), salutation))

you couldn't have a test for it