Filter stream by message header

[bmmptlgc]

I am new to Streamiz.Kafka.Net and I am trying to write a streaming application the will consume messages from a multi Avro schema Kafka topic, filter messages on a message type header and then publish the non filtered messages to a topic dedicated to a single message type.

For example, lets imagine source-topic has message types A, B and C (header message-type will contain the value). Considering that we want to filter out messages of type B, the end result would be all messages with message-type A being published to destination-topic-A and all messages with message-type C being published to destination-topic-C.

var sourceStream = builder.Stream<string, GenericRecord>(topicConfig.TopicName);

sourceStream
    .Filter((key, value, context) =>
    {
        var messageTypeHeader = context.Headers.FirstOrDefault(h => h.Key == "message-type!")?.GetValueBytes();
        if (messageTypeHeader == null)
            return false;

        var messageType = Encoding.UTF8.GetString(messageTypeHeader);
        return topicConfig.MessageTypes.Contains(messageType);
    })
    .To((key, value, context) =>
    {
        var messageTypeHeader = context.Headers.FirstOrDefault(h => h.Key == KnownHeaders.MessageType)?.GetValueBytes();
        if (messageTypeHeader == null)
            return null;

        var messageType = Encoding.UTF8.GetString(messageTypeHeader);
        return $"destination-topic-{messageType}";
    }, new StringSerDes(), new SchemaAvroSerDes<GenericRecord>());

Something like that wouldn't work, because Filter's Func parameter doesn't have a context parameter. Is there another way to achieve my goal?

[LGouellec]

Hi @bmmptlgc,
Did you try something like that ?

var predicateHeader = (Headers headers, string headerKey, string headerValue) =>
            {
                var messageTypeHeader = headers.FirstOrDefault(h => h.Key == headerKey)?.GetValueBytes();
                if (messageTypeHeader == null)
                    return false;

                var messageType = Encoding.UTF8.GetString(messageTypeHeader);
                return messageType.Equals(headerValue);
            };
            
            var sourceStream = builder.Stream<string, GenericRecord>("input");

            var branches = sourceStream
                .Branch(
                    (k,v) => predicateHeader(StreamizMetadata.GetCurrentHeadersMetadata(), "message-type", "orders"),
                    (k,v) => predicateHeader(StreamizMetadata.GetCurrentHeadersMetadata(), "message-type", "sales"),
                    (k,v) => predicateHeader(StreamizMetadata.GetCurrentHeadersMetadata(), "message-type", "test"),
                    (s, record) => true);
                
            // only messages with the header message-type = orders
            branches[0].To(...);
            // only messages with the header message-type = sales
            branches[1].To(...);
            // only messages with the header message-type = test
            branches[2].To(...);
            // dlq 
            branches[3].To("dlq-topic");

Btw, I plan to add the context in the filter and branch DSL processors in a short term

[bmmptlgc]

@LGouellec Thanks for the quick reply. I'll give this a try. I just need to make branching dynamic, because the message-types to split to single topics is set in configuration.

I noticed I had a mistake in my question and I just edited it.

I assume (s, record) => true would be a branch for all source messages types that do not need to be published (like message type B in my example above). Is that correct?

And then in your example, I would comment out branches[3].To("dlq-topic"); since I just want to ignore those messages.

But I will definitely refactor once context is available in `the filter, since it's a more elegant and readable solution.

[LGouellec]

I assume (s, record) => true would be a branch for all source messages types that do not need to be published (like message type B in my example above).

Correct

But I will definitely refactor once context is available in `the filter, since it's a more elegant and readable solution.

If you would like to use Filter now, you can with the StreamizMetadata.GetCurrentHeadersMetadata().

You should have to test, but normally it's fine
Example :

var predicateHeader = (Headers headers, string headerKey, string headerValue) =>
            {
                var messageTypeHeader = headers.FirstOrDefault(h => h.Key == headerKey)?.GetValueBytes();
                if (messageTypeHeader == null)
                    return false;

                var messageType = Encoding.UTF8.GetString(messageTypeHeader);
                return messageType.Equals(headerValue);
            };
            
            var sourceStream = builder.Stream<string, GenericRecord>("input");

            sourceStream
                .Filter((k,v) => predicateHeader(StreamizMetadata.GetCurrentHeadersMetadata(), "message-type", "orders"))
                .To("orders");
                
            sourceStream
                .Filter((k,v) => predicateHeader(StreamizMetadata.GetCurrentHeadersMetadata(), "message-type", "sales"))
                .To("sales");

           // etc ...

[LGouellec]

Subscribe this issue : #334

[bmmptlgc]

Hi @LGouellec, StreamizMetadata.GetCurrentHeadersMetadata() inside Filter is always null.

[LGouellec]

HI @bmmptlgc ,

Sorry I forgot something. Please enable this configuration and I should be fine :

config.FollowMetadata = true;