Aggregate bulk of messages in a unique one: TumblingWindow behavoir.

[alessandroros]

Hello, first of all thanks for this great project. Is helping me a lot on "managing" my kafka streams with dotnet. I would like to ask help for accomplishing this stuff. Suppose you have a lot of this messagges incoming in a topic:

public class Message
{
    public DateTime? TimeMsgReceived { get; set; }
    
    public Asset Asset { get; set; }
    
    public Gps Gps { get; set; }
    
    public Param Param { get; set; }
}

I would like to aggregate all of this data in a message like this:

public class AggregatedMessage 
{
    public DateTime? TimeMsgReceived { get; set; }
    
    public Asset Asset { get; set; }
    
    public Gps Gps { get; set; }

    public List<Param> Params { get; set; }
}

Where it's the same but instead of having a single param it has a list of Param. All this messages have a key that is unique for the group of messages I would like to aggregate.
I developed this code based on documentation/issues/discussions found on this repo 😄

var streamConfig = new StreamConfig
        {
            ApplicationId = configuration["Kafka:ApplicationId"],
            BootstrapServers = configuration["Kafka:ConsumerSettings:BootstrapServers"],
            SaslMechanism = SaslMechanism.Plain,
            SaslUsername = configuration["Kafka:ConsumerSettings:SaslUsername"],
            SaslPassword = configuration["Kafka:ConsumerSettings:SaslPassword"],
            SecurityProtocol = SecurityProtocol.SaslSsl,
            AutoOffsetReset = AutoOffsetReset.Earliest,
            StateDir = Path.Combine("."),
            CommitIntervalMs = 5000,
            Guarantee = ProcessingGuarantee.AT_LEAST_ONCE,
            MetricsRecording = MetricsRecordingLevel.DEBUG,
            
            Logger = LoggerFactory.Create(b =>
            {
                b.SetMinimumLevel(LogLevel.Debug);
                b.AddConsole();
            }),
            ReplicationFactor = 3,
            DefaultKeySerDes = new StringSerDes(),
            DefaultValueSerDes = new JsonSerDes<Message>()
        };

var stringSerdes = new StringSerDes();
var messageSerdes = new JsonSerDes<Message>();
var builder = new StreamBuilder();
var messageStream = builder.Stream<string, Message>(inputTopic, stringSerdes, messageSerdes);
messageStream.Filter((k, v) => CheckFilterCriteria(allowedVehicle, allowedSignals, v))
            .Map((key, value) => new KeyValuePair<string, Message>(CreateMessageKey(value), value))
            .GroupByKey()
            .WindowedBy(TumblingWindowOptions.Of(200000))
            .Aggregate<MessageAggregated,JsonSerDes<MessageAggregated>>(
                () => new MessageAggregated
                {
                    Params = new List<Param>()
                },
                (key, message, aggregatedMessage) =>
                {
                    aggregatedMessage.Asset = message.Asset;
                    aggregatedMessage.TimeMsgReceived = message.TimeMsgReceived;
                    aggregatedMessage.Gps = message.Gps;
                    aggregatedMessage.Params.Add(message.Param);
                    return aggregatedMessage;
                }
            )
            .ToStream()
            .Map((key, messageAggregated) => new KeyValuePair<string, MessageAggregated>(key.Key, messageAggregated))
            .To<StringSerDes,JsonSerDes<MessageAggregated>>(groupedOutputTopic);

When I send 3 messages in a time <200000ms, I see 3 messages. One with the first message, the second with 2 messages aggregated and the third with 3 messages aggregated correctly. Is this the correct behavoir of the TumblingWindow? I expected to get just the third aggregated message and not even the prevoius messages. Is this correct? I though that the "window time" was waiting for "new messages" before starting the aggregation.
Any help would be really appreciated 😃

[alessandroros]

An update on this, maybe I need the suppress feature in order to achieve what I mean i think #70 and here #134.

[LGouellec]

Hi @alessandroros,

Thanks for this discussion.
Yes this exactly what you need, the Supress(..) feature. For now, it's not present in the 1.5.0 roadmap.
Because before releasing this feature, I need to deliver the cache functionality and it's planned for 1.5.0

[alessandroros]

Hi @LGouellec,
thanks a lot for the reply. Really appriciated. When the cache functionality will be released, is there a way to achieve what I would like to do with the caching feature? Just for knowing a workaround when the 1.5.0 will be released.

[LGouellec]

The caching feature would permit to not send to the downstream processors all intermediate result events.
Meaning , without caching feature with a word count use case, you can have :
Input topic
key1:toto
key2:titi
key1:toto
key2:titi

Output topic :
key1:1
key2:1
key1:2
key2:2

With the caching feature, you could have something like this with the same input records :
Output topic :
key1:2
key2:2

In JAVA, we have a 10Mb cache enabled per default and the cache is flushed every commit.interval.ms. I consider this solution or maybe add a new configuration if you want to flush the cache more regularly than the commit.

[alessandroros]

Great news. That's perfect. Thank you a lot. That's more than enough for what I need to do 👍