Merging node seems to not forward records into transformer node

[maciejniepokoj]

I found out, that merging node passes record through only if next node is an IProcessor. In hypothetical topology:

SOURCE 1 SOURCE 2
| |
\ /
MERGE
|
TRANSFORMER

All records will be ignored, nothing will reach the ITransformer. I believe it results directly from how AbstractProcessor<K, V> is implemented? My workaround is to add a mapper, that returns records without changing them, between merge and transformer.

I wrote and run a test using test driver:

test.txt
Unfortunately it passes without a problem.

I'm also attaching topology on which I observed this behavior:

topology.txt

And logs, from which I understand a record was forwarded from merge node (KSTREAM-MERGE-0000000014) to next node, but next node never received it:

logs.txt

[LGouellec]

Hey @maciejniepokoj ,

I did this topology with a real kafka cluster and it works ! I have messages downstream the Merge and Transformer nodes.

var source1Stream = builder.Stream<string, string>("source1");
            var source2Stream = builder.Stream<string, string>("source2");
        
            source1Stream
                .Merge(source2Stream)
                .Transform(TransformerBuilder
                    .New<string, string, string, string>()
                    .Transformer((record) => Record<string, string>.Create(record.Key, record.Value.ToUpper()))
                    .Build())
                .To("output");

Are you sure you don't miss anything ?