Code block completing before method inside MapValuesAsync finishes

[RossSteele1]

Description

Streamiz version: 1.4.2
ConfluentKafka: 1.9.3

I have a Stream where I want to take in a string and call a method on MapValuesAsync to determine the correct version type then either upscale it or return the correct event version. I then want to map the type of the string value to become a eventEnvelope

Further in the code block I have a join on this stream which I want to happen after mapvaluesAsync has completed.

Is it possible to force the code to wait until MapValuesAsync has completed before progressing the rest of the code block?

How to reproduce

Method:
`
private Task GetStreamTopology()
{
var builder = new StreamBuilder();
var stream = builder
.Stream<string, string>(inputTopic)
.MapValuesAsync(
async (record, _) =>
{
var test = await _eventProcessor.ProcessRawMessage(JsonConvert.SerializeObject(record.Value));
return test;
}, null,
new RequestSerDes<string, string>(new StringSerDes(), new StringSerDes()),
new ResponseSerDes<string, EventEnvelope(new StringSerDes(),
new JsonSerDes<EventEnvelope>()));

    var table1 = builder
        .GlobalTable(topic),
            new StringSerDes(),
            new JsonSerDes<EventEnvelope<Event>>(),
            InMemory.As<string, EventEnvelope<Event>>(store)
                .WithValueSerdes<JsonSerDes<EventEnvelope<Event>>>());

    var table2 = builder
        .GlobalTable(topic),
            new StringSerDes(),
            new JsonSerDes<EventEnvelope<Event>>(),
            InMemory.As<string, EventEnvelope<Event>>(store)
                .WithValueSerdes<JsonSerDes<EventEnvelope<Event>>>());

    stream.Join(table1,
            (_, value) => value.id,
            (value1, value2) =>
            {
                var aggregateEvent = new AggregateEvent
                {
                    aggregatedEvent = value1,
                    service = value2
                };

                return aggregateEvent;
            })
        .Join(table2,
            (_, value) => value.id
            (aggregateEvent, settings) =>
            {
                aggregateEvent.version = settings.Payload?.version;
                return aggregateEvent;
            })
        .Filter((_, value) => value.services != null && value.services.Contains("id", StringComparer.OrdinalIgnoreCase) && value.version == 3)
        .MapValues((_, value) => value.Order)
        .To<StringSerDes, JsonSerDes<EventEnvelope<Event>>>(outputTopic);

    return Task.FromResult(builder.Build()).WaitAsync(new CancellationToken());
}

TEST: [IsUnit, Theory, AutoMoqData]
public async Task CreateStream_ProducesOrderToOutputTopic(
[Frozen] StreamSettings streamSettings,
StreamCreator streamCreator)
{
var stream = await streamCreator.CreateStream();

    using var driver = new TopologyTestDriver(stream.Topology, stream.Config);
    var inputTopic = driver.CreateInputTopic<string, string>($"test-inputTopic".ToLowerInvariant());
    var table1Topic = driver.CreateInputTopic<string, string>("test-table1");
    var table2Topic = driver.CreateInputTopic<string, string>("test-table2");
    var outputTopic = driver.CreateOuputTopic<string, string>($"test-outputTopic".ToLowerInvariant(), TimeSpan.FromSeconds(5));

    CreateTable1Event(table1Topic, "id", "id");
    CreateTable2Event(table2Topic, "id", id);
    CreateEvent(inputTopic, "id", "id");

    var output = outputTopic.ReadKeyValue();

    output.Should().NotBeNull();
    output.Message.Key.Should().Be("id");
    JsonConvert.DeserializeObject<EventEnvelope<Event>>(output.Message.Value)?.Payload.Id.Should().Be("id");
}`

CreateEvent() method in the test kicks off the event that streamCreator is listening to.
The issue is that builder.build after stream.join happens before MapValuesAsync, so the output topic is overwritten and output is null in test.
If I remove the line for mapValuesAsync I get a passing test.

Checklist

Please provide the following information:

• [x ] A complete (i.e. we can run it), minimal program demonstrating the problem. No need to supply a project file.
• [ x] A code snippet with your topology builder (ex: builder.Stream<string, string>("topic").to("an-another-topic");)
• [x ] Streamiz.Kafka.Net nuget version.
• Apache Kafka version.
• Client configuration.
• Operating system.
• Provide logs (with in debug mode (log4net and StreamConfig.Debug) as necessary in configuration).
• Critical issue.

[LGouellec]

Hey @RossSteele1 ,

I tried to reproduce without success. If you can checkout this branch and test it with your custom code (event, async event processor), would be very helpful to troubleshoot : https://github.com/LGouellec/kafka-streams-dotnet/blob/reproduce/344/test/Streamiz.Kafka.Net.Tests/Reproducer344Tests.cs

But in other words, did you test your topology with a real kafka cluster ?
Don't know if you know, but the Global Table, Stream and Async Stream are run into multiple non-concurrent threads in the real life, TopologyTestDriver try to run sequentially the input records.
https://lgouellec.github.io/kafka-streams-dotnet/threading-model.html

So be careful, because in the real life with a real Kafka Cluster, your join stream-global table is dependant of the order of producing message and order of processing. If your stream input topic event is processed before your event from your global Table, the join won't happen.

Why using a global table and not a KTable because it seems you have the same key. So if your topic are co-partitioned, my recommendation could be to use a KTable.

[RossSteele1]

Hi @LGouellec , thanks so much for the reply!
I was able to recreate the error scenario in your branch, you can take the code here:

`using System;
using System.Threading;
using System.Threading.Tasks;
using Newtonsoft.Json;
using NUnit.Framework;
using Streamiz.Kafka.Net.Mock;
using Streamiz.Kafka.Net.SerDes;
using Streamiz.Kafka.Net.Stream;
using Streamiz.Kafka.Net.Table;

namespace Streamiz.Kafka.Net.Tests;

public class Reproducer344Tests
{
public class Event
{
public string Order { get; set; }
public string Version { get; set; }
}

public class Eventv3
{
    public int Order { get; set; }
    public string Version { get; set; }
}

public class EventEnvelope
{
    public string UUID{ get; set; }
    public string Id { get; set; }
    public DateTime Timestamp { get; set; }
}

public class EventEnvelope<T> : EventEnvelope
{
    public T Event { get; set; }
}

public class AggregateEvent
{
    public EventEnvelope StreamEvent { get; set; }
    public string Order { get; set; }
    public string Version { get; set; }
}

private class Reproducer344TestsProcessor
{
    public async Task<EventEnvelope<Eventv3>> ProcessRawMessage(string id)
    {
        var rawMessage = id;
        try
        {
            var v2Envelope = JsonConvert.DeserializeObject<EventEnvelope<Event>>(rawMessage);
            //External awaited mapper call
            var v3Envelope = MapToV3(v2Envelope);
            rawMessage = JsonConvert.SerializeObject(v3Envelope);

            return JsonConvert.DeserializeObject<EventEnvelope<Eventv3>>(rawMessage);
        }
        catch (Exception e)
        {
            return JsonConvert.DeserializeObject<EventEnvelope<Eventv3>>(rawMessage);
        }
    }
}

public static async Task<EventEnvelope<Eventv3>> MapToV3(EventEnvelope<Event> v2Envelope)
{
    return new EventEnvelope<Eventv3>
    {
        Id = v2Envelope.Id,
        Timestamp = v2Envelope.Timestamp,
        UUID = v2Envelope.UUID,
        Event = new Eventv3
        {
            Order = int.Parse("123"),
            Version = v2Envelope.Event.Version
        }
    };
}

private Topology GetStreamTopology()
{
    var eventProcessor = new Reproducer344TestsProcessor();
    var builder = new StreamBuilder();
    
    var stream = builder
        .Stream<string, string>("test-input")
        .MapValuesAsync(
            async (record, _) =>
            {
                var test = await eventProcessor.ProcessRawMessage(record.Value);
                return test;
            }, null,
            new RequestSerDes<string, string>(new StringSerDes(), new StringSerDes()),
            new ResponseSerDes<string, EventEnvelope<Eventv3>>(new StringSerDes(),
                new JsonSerDes<EventEnvelope<Eventv3>>()));

    var table1 = builder
        .GlobalTable("test-table1", 
    new StringSerDes(),
    new JsonSerDes<EventEnvelope<Event>>(),
    InMemory.As<string, EventEnvelope<Event>>("test-table1-store")
        .WithValueSerdes<JsonSerDes<EventEnvelope<Event>>>());

    var table2 = builder
        .GlobalTable("test-table2", 
    new StringSerDes(),
    new JsonSerDes<EventEnvelope<Event>>(),
    InMemory.As<string, EventEnvelope<Event>>("test-table2-store")
        .WithValueSerdes<JsonSerDes<EventEnvelope<Event>>>());

    stream.Join(table1,
            (_, value) => value.Id,
            (value1, value2) =>
            {
                var aggregateEvent = new AggregateEvent
                {
                    StreamEvent = value1,
                    Order = value2.Event.Order
                };

                return aggregateEvent;
            })
        .Join(table2,
            (_, value) => value.StreamEvent.Id,
                (aggregateEvent, event2) =>
            {
                aggregateEvent.Version = event2.Event.Version;
                return aggregateEvent;
            })
        .Filter((_, value) => value.Order != null && value.Version != null)
        .MapValues((_, value) => value)
        .To<StringSerDes, JsonSerDes<AggregateEvent>>("test-output");

    return builder.Build();
}

[Test]
public void CreateStream_ProducesOrderToOutputTopic()
{
    var stream = GetStreamTopology();
    var config = new StreamConfig<StringSerDes, StringSerDes>();
    config.ApplicationId = "test-reproducer-344";
    config.BootstrapServers = "localhost:9092";
    
    using var driver = new TopologyTestDriver(stream, config);
    var inputTopic = driver.CreateInputTopic<string, string>($"test-input");
    var table1Topic = driver.CreateInputTopic<string, EventEnvelope<Event>, StringSerDes, JsonSerDes<EventEnvelope<Event>>>("test-table1");
    var table2Topic = driver.CreateInputTopic<string, EventEnvelope<Event>, StringSerDes, JsonSerDes<EventEnvelope<Event>>>("test-table2");
    var outputTopic = driver.CreateOuputTopic<string, AggregateEvent, StringSerDes, JsonSerDes<AggregateEvent>>($"test-output");

    table1Topic.PipeInput("id", 
        new EventEnvelope<Event>()
        {
            Id = "id", Timestamp = DateTime.Now, UUID = Guid.NewGuid().ToString(),
            Event = new Event()
            {
                Order = "order1"
            }
        });
    
    table2Topic.PipeInput("id",
        new EventEnvelope<Event>()
        {
            Id = "id", Timestamp = DateTime.Now, UUID = Guid.NewGuid().ToString(),
            Event = new Event {
                Version = "v3"
            }
        });
    
    inputTopic.PipeInput("id", JsonConvert.SerializeObject(new EventEnvelope<Event>()
    {
        Id = "id",
        Timestamp = DateTime.Now,
        UUID = Guid.NewGuid().ToString(),
        Event = new Event
        {
            Version = "v3"
        }
    }));

    var output = outputTopic.ReadKeyValue();
    Assert.IsNotNull(output);
    Assert.AreEqual("v3", output.Message.Value.Version);
    Assert.AreEqual("order1", output.Message.Value.Order);
}

}`

Output is null, compared to the positive test scenario you created; OutputTopic.Pipe.Consumer.Producer.Topics is count 5 in this error state, whereas it was count 6 in passing test, "output-topic" is missing from this list.

Just for more info we have had this working in a live environment originally before any map values was required, but our requirement is that 2 versions of the same event are going to be produced in parallel, mapvaluesasync is taking the event as a string, and returning v3 or upscaling v2 to v3, then returning the new mapped event.

V3 has breaking changes when it tries to map to v2 schema and we haven't been able to find a way to separate events by version, since deserializing v3 causes an error so we can't read metadata. (We will overwrite the schema to v3 hence the upscaler)

[LGouellec]

Hi @RossSteele1 ,

I made couple of changes of your code, and now the unit test is green ✅

rawMessage = JsonConvert.SerializeObject(v3Envelope.Result);
// instead of 
rawMessage = JsonConvert.SerializeObject(v3Envelope);

// the issue was due because you tried to serialize the v3Envelope which is a Task<Envelope> and not the result of the task

Please checkout my branch and test another time.

If you switch the TopologyTestDriver.Mode to ASYNC_CLUSTER_IN_MEMORY (an async mock Kafka cluster), the unit test still green as well :

using var driver = new TopologyTestDriver(stream, config, TopologyTestDriver.Mode.ASYNC_CLUSTER_IN_MEMORY);

But what you need to understand, if you want to join with GlobalKTable, events must be sent before events into the input-topic.
Let me resume with different timeline.

Join Works

Timestamp	Input topic	Async-request topic	Async-response topic	Table1 topic	Table1 topic	Output topic
t0				id;x
t1					id;y
t2	id;v2
t3		id;v2
t4			id;v3
t5						id;v3+x+y

Join doesn't work because id;x comes too late

Timestamp	Input topic	Async-request topic	Async-response topic	Table1 topic	Table1 topic	Output topic
t0					id;y
t1	id;v2
t2		id;v2
t3			id;v3
t4				id;x
t5

Join doesn't work because id;y comes too late

Timestamp	Input topic	Async-request topic	Async-response topic	Table1 topic	Table1 topic	Output topic
t0				id;x
t1	id;v2
t2		id;v2
t3			id;v3
t4					id;y
t5