Message Filtering Feature in Consumer of Brighter

[honkuan86]

Hi all,

I would like to get your expertise opinion on a scenario where we need to filter messages from the consumer of a single Kafka Topic. We have a few services that need to communicate with each other via Kafka, the Topics are setup in a way that each service will have one Topic that they publish their events to, so any events related to the service will be published there, and any other services that are interested to the Topic will subscribe to it. For example: service-a will publish events to service-a-topic and other services will subscribe to service-a-topic.

However, a particular service may not have interests in every messages that are published to the Topic, and the way that we can do now with Brighter is to include a EventType in the body of the message and do some skipping logic in the Handler.

The downside to this approach is a consumer will need to inspect every messages to know if the message is processable which will be costly in terms of performance.

So I would like to get your opinion if we can include a message filtering feature that we can inspect the message header for the EventType (configurable) and filter it before it reach to the Handler, so it will avoid the message deserializing process. Or is the way we setup the Topics is not a good idea ?

[DevJonny]

Hi @honkuan86,

There are two aspects to this, the first is that having one topic per service is not the approach you want to take. The topics should be much more granular, such as one topic per entity. As you've already described you're having to inspect every single message from a topic you're subscribed to see if you're interested in that message or not, again as you've said this is costly in both time and money. So my first bit of advice is revisit how you are breaking up topics, and this issue will most like become moot.

If you really have no other option, and I suggest you take this up with Jason as there are better options, then you have another approach. The filtering can be done in a two step process.

1. Inspect the incoming message as part of your MessageMapper and check the headers to see if you want to process the message. If you don't want to then you could skip deserialising the body, and create an empty/default version of the Event object, as in a specific type that you can check. So something like SkipMessageEvent.

2. Filter out any SkipMessageEvents by creating a attribute/handler pair like the UseInboxAsyncAttribute and UseInboxHandlerAsync.cs. This will be added as a before attribute to your actual handler. The handler part will check each message and see if it's of type SkipMessageEvent and it is return the incoming event rather than calling the base.HandleAsync(..) method. This will stop any more processing of the message, but most importantly will make sure that Kafka moves onto the next message and you don't get stuck at the current offset.

@iancooper anything I've missed or could explain better?

[iancooper]

@jonnyollifflee No, that is good. If you can easily identified from the serialized message, you might not need to convert in the message mapper, just add the custom filter that terminates the pipeline. To terminate, you just don't call the base.Handle(..) method. That makes you the end of the line. So in pseudocode:

if (not an event I want)
return;
else
return base.Handle(...)

But as Jonny says prefer to use one channel per type: https://www.enterpriseintegrationpatterns.com/patterns/messaging/DatatypeChannel.html If you have ordering constraints, it is easier to have MyEvent (type: ThisHappened) and MyEvent (type:thatHappened) than MyThisEvent and MyThatEvent, but this is only because those events must be processed in order with respect to each other. If they don't need to be, separate channels