How to change SerDes after mapping

[okomen-foreflight]

Hey there!

I have a stream:
var transformed = builder.Stream<string, AvroFleetAggregator, StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(KafkaConstants.FLEET_TRANSFORMED_TOPIC);

And I want to change the underlying types from a <string, AvroFleetAggregator> to a <long, AvroFleetKafkaMessage> type instead. What I've tried to do is to use Map:
var flightIdStream = transformed .Map<long, AvroFleetKafkaMessage>((k, v) => { return new KeyValuePair<long, AvroFleetKafkaMessage>(1, v.PreviousMessages.First()); value.PreviousRawUpdates.Select(x => x.ToFleetFlightMessage()).ToList()); }, "FlightIdStream");
But of course I get a SerDes error:
The key serdes (Streamiz.Kafka.Net.SerDes.ByteArraySerDes) is not compatible to the actual key (System.Int64) for this processor. Change the default key serdes in StreamConfig or provide correct Serdes via method parameters(using the DSL)

How do I change the SerDes it should use?

[LGouellec]

Hi @okomen-foreflight,

All depends what is the next processor in your topology ?
Can you share your entirely topology please ?

Best regards,

[okomen-foreflight]

Hey @LGouellec,

Thanks for making this library!
Here is my entire topology:
I have as a last resort had to push to many intermediary topics because I had to change the SerDes and this was the only way I found to do so.
I would prefer if I could avoid many of the .To() such as after the AvroFleetAggregator which produces to topic KafkaConstants.FLEET_TRANSFORMED_TOPIC only to then read from it again in the transformed stream

`

    var builder = new StreamBuilder();

    var fleetUpdatesStream = builder.Stream<string, AvroFleetKafkaMessage, StringSerDes, SchemaAvroSerDes<AvroFleetKafkaMessage>>(KafkaConstants.FLEET_TAIL_POSITIONS_TOPIC);
    
    var metadataTable = builder.GlobalTable<long, AvroInfraFlightMetadata, Int64SerDes, SchemaAvroSerDes<AvroInfraFlightMetadata>>(
        KafkaConstants.FLEET_METADATA_TOPIC,
        InMemory.As<long, AvroInfraFlightMetadata>());

    var metadataUpdatesStream = builder.Stream<long, AvroInfraFlightMetadata, Int64SerDes, SchemaAvroSerDes<AvroInfraFlightMetadata>>(KafkaConstants.FLEET_UPDATED_METADATA_TOPIC);
    metadataUpdatesStream.LeftJoin(metadataTable, (flightId, _) => flightId, (updatedMetadata, existingMetadata) =>
    {
        ...
        return existingMetadata;
    })
    .To(KafkaConstants.FLEET_METADATA_TOPIC);

    var filteredStream = fleetUpdatesStream.Filter((_, value) =>
    {
        // return _filteringService.Keep(value.ToFleetFlightMessage());
        return true;
    }, "Filtered");

    var aggregatorInit = new AvroFleetAggregator();
    filteredStream
        .GroupByKey<StringSerDes, SchemaAvroSerDes<AvroFleetKafkaMessage>>("PreTransform")
        .Aggregate(() => aggregatorInit, (_, newUpdate, aggregator) =>
        {
            ...
            if (aggregator.PreviousMessages is null)
            {
                aggregator.PreviousMessages = new List<AvroFleetKafkaMessage>();
            }
            aggregator.PreviousMessages.Add(newUpdate);

            return aggregator;
        }, RocksDb.As<string, AvroFleetAggregator, StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(), "Transformed")
        .ToStream()
        .To<StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(KafkaConstants.FLEET_TRANSFORMED_TOPIC);

    var transformed = builder.Stream<string, AvroFleetAggregator, StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(KafkaConstants.FLEET_TRANSFORMED_TOPIC);

    var flightIdStream = transformed
        .Map<long, AvroFleetKafkaMessage>((k, v) =>
        {
            return new KeyValuePair<long, AvroFleetKafkaMessage>(1, v.PreviousMessages.First()); 
        }, "FlightIdStream");
    
    var enrichedStream = flightIdStream.LeftJoin(metadataTable, (flightId, _) => flightId,
        (update, metadata) =>
        {
            if (metadata is null)
                return update;

            // var result = _enrichingService.Enrich(update.ToFleetFlightMessage(), metadata.ToFlightMetadata()).ToFleetKafkaMsg();
            return update;
        }, "EnrichedStream");

    
    var aggregator = new AvroProcessorAggregator();
    enrichedStream
        .GroupByKey("ProcessorGroupByKey")
        .Aggregate(() => aggregator, (_, update, processorAggregator) =>
        {
            ..
            if (processorAggregator.PreviousMessages is null)
                processorAggregator.PreviousMessages = new List<AvroFleetKafkaMessage>();
            processorAggregator.PreviousMessages.Add(update);
            return processorAggregator;
        }, RocksDb.As<long, AvroProcessorAggregator, Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>())
        .ToStream()
        .To<Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>(KafkaConstants.FLEET_PRE_SINK);

    var preSink = builder.Stream<long, AvroProcessorAggregator, Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>(KafkaConstants.FLEET_PRE_SINK);
    preSink
        .ForeachAsync(async (record, _) =>
        {
            ...
            await Task.Delay(500, stoppingToken);
        });

    Topology = builder.Build();`

[okomen-foreflight]

It seems to be the LeftJoin that does it. Should I produce it to an output topic? Even though I am using it straight afterwards?

Furthermore it seems when I produce to output topics, the entire stream becomes slower compared to just never producing to output topics. I imagine I should increase the amount of stream threads? Is there any way to know how many threads to increase it to? Like a rule of thumb or something

[LGouellec]

Do you try this one ? I let multiple comments and question ..

var builder = new StreamBuilder();

var fleetUpdatesStream = builder.Stream<string, AvroFleetKafkaMessage, StringSerDes, SchemaAvroSerDes<AvroFleetKafkaMessage>>(KafkaConstants.FLEET_TAIL_POSITIONS_TOPIC);

var metadataTable = builder.GlobalTable<long, AvroInfraFlightMetadata, Int64SerDes, SchemaAvroSerDes<AvroInfraFlightMetadata>>(
    KafkaConstants.FLEET_METADATA_TOPIC,
    InMemory.As<long, AvroInfraFlightMetadata>());

var metadataUpdatesStream = builder.Stream<long, AvroInfraFlightMetadata, Int64SerDes, SchemaAvroSerDes<AvroInfraFlightMetadata>>(KafkaConstants.FLEET_UPDATED_METADATA_TOPIC);
metadataUpdatesStream.LeftJoin(metadataTable, (flightId, _) => flightId, (updatedMetadata, existingMetadata) =>
{
    ...
        return existingMetadata;
})
.To(KafkaConstants.FLEET_METADATA_TOPIC);

var filteredStream = fleetUpdatesStream.Filter((_, value) =>
{
    // return _filteringService.Keep(value.ToFleetFlightMessage());
    return true;
}, "Filtered");

var aggregatorInit = new AvroFleetAggregator();
filteredStream
    .GroupByKey<StringSerDes, SchemaAvroSerDes<AvroFleetKafkaMessage>>("PreTransform")
    .Aggregate(() => aggregatorInit, (_, newUpdate, aggregator) =>
    {
        ...
            if (aggregator.PreviousMessages is null)
        {
            aggregator.PreviousMessages = new List<AvroFleetKafkaMessage>();
        }
        aggregator.PreviousMessages.Add(newUpdate);

        return aggregator;
    }, RocksDb.As<string, AvroFleetAggregator, StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(), "Transformed")
    .ToStream()
    .To<StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(KafkaConstants.FLEET_TRANSFORMED_TOPIC);  // Why do you need this ?, why you don't continue your topolgy ? 

var transformed = builder.Stream<string, AvroFleetAggregator, StringSerDes, SchemaAvroSerDes<AvroFleetAggregator>>(KafkaConstants.FLEET_TRANSFORMED_TOPIC);

var flightIdStream = transformed
    .Map<long, AvroFleetKafkaMessage>((k, v) =>
    {
        return new KeyValuePair<long, AvroFleetKafkaMessage>(1, v.PreviousMessages.First());
    }, "FlightIdStream");

var enrichedStream = flightIdStream.LeftJoin(metadataTable, (flightId, _) => flightId,
    (update, metadata) =>
    {
        if (metadata is null)
            return update;

        // var result = _enrichingService.Enrich(update.ToFleetFlightMessage(), metadata.ToFlightMetadata()).ToFleetKafkaMsg();
        return update;
    }, "EnrichedStream");


var aggregator = new AvroProcessorAggregator();
enrichedStream
    .GroupByKey<Int64SerDes, SchemaAvroSerDes<AvroFleetKafkaMessage>>("ProcessorGroupByKey") // Change to the correct serdes
    .Aggregate(() => aggregator, (_, update, processorAggregator) =>
    {
        ..

        if (processorAggregator.PreviousMessages is null)
            processorAggregator.PreviousMessages = new List<AvroFleetKafkaMessage>();
        processorAggregator.PreviousMessages.Add(update);
        return processorAggregator;
    }, RocksDb.As<long, AvroProcessorAggregator, Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>())
    .ToStream()
    .To<Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>(KafkaConstants.FLEET_PRE_SINK); // Why do you need this ?, why you don't continue your topolgy ? 

var preSink = builder.Stream<long, AvroProcessorAggregator, Int64SerDes, SchemaAvroSerDes<AvroProcessorAggregator>>(KafkaConstants.FLEET_PRE_SINK);
preSink
    .ForeachAsync(async (record, _) =>
    {
        ...
            await Task.Delay(500, stoppingToken);
    });

Topology = builder.Build();`

Regarding the performance, exactly if you use multiple intermediate topic, the performance can decreased .. The number of threads depends how much partition do you have in the source topic. You can't have much threads than the number of partitions.
Max number of threads per instance = Number of source partitions / Instance number.

If your source topic has 10 partitions, and you scale your application at 2. Your max number of threads per instance = 5.

Hope it helps !

[okomen-foreflight]

Hey there! Ahh that makes sense with the threads. Thank you.

So regarding the "change to the correct SerDes" what would the correct one be?
Above the enrichedStream i am mapping to a long and a AvroFleetKafkaMessage and then I make an enrichedStream using a FeftJoin where I just return update which is also an AvroFleetKafkaMessage.

[LGouellec]

The correct serdes could be :
Key : Int64SerDes
Value : AvroFleetKafkaMessage's serdes, so maybe SchemaAvroSerDes if you want to use Schema Reegistry.

In this situation, the serdes at the GroupByKey level is actually the serdes before the groupByKey operation, so the LeftJoin processor, which is actually the same than Map(..)

[okomen-foreflight]

I see, I would like to use Schema registry, but I can't seem to change the SerDes for my mapping function. And since the GroupByKey uses the LeftJoin which uses the map, it seems it's the map function that I need to modify. So how do I map with while changing it to use SchemaAvroSerDes?
So instead of .Map<long, AvroFleetKafkaMessage>((k, v) => what should it then be to change the SerDes if I want to use Schema Registry?

[LGouellec]

Hi @okomen-foreflight ,

In this particulier case, you can't set a serdes to Map processor because SerDes is only used to publish or consume a Kafka topic, and a Map processor is not backed by a kafka topic.

GroupByKey yes, because when you use Map, you change the key, so I need to repartition the data, that's why the groupByKey is backed by a kafka topic sometimes.

A strong recommendation is to get the topology description via :

var streamBuilder = new StreamBuilder();
...
var topology = streamBuilder.Build();
var topologyDescription = topology.Describe().ToString();

and put the description in this tools : https://zz85.github.io/kafka-streams-viz/

You can visualise all the internal topics and the list of processors