Using filters with Kafka topic endpoints

[benbadura]

Hello, I'm trying to add a filter that would serve function of logging and measuring time required to finish call to Consume for all consumers registered in my sln.
When it comes to consumers called in memory - everything is working fine - filter is called correctly, however when it comes to kafka consumer - it refuses to trigger.
MT does not raises any errors, it just silently omits that filter. I've tried figuring out what might be the issue, but I'm stuck.

Example of flow with working filter:

controller -> _bus.Publish()-> [LoggingFilter] -> next() -> MessageConsumer.Consume() -> [LoggingFilter]

Example of flow with filter omitted:

API_1 [ controller -> _bus.Publish()-> [LoggingFilter] -> next() -> MessageConsumer.Consume() -> topicProducer.Produce() -> [LoggingFilter] ] -> [API_2] MessageConsumer.Consume()

All help would be really appreciated.

Thanks!

MT Registration

builder.Services.AddMassTransit(busRegistrationConfig =>
{
    busRegistrationConfig.UsingInMemory((context, cfg) =>
    {       
        cfg.ConfigureEndpoints(context);

        //filters registration
        cfg.UseConsumeFilter(typeof(LoggingFilter<>), context);
    });

    busRegistrationConfig.AddConsumers(Assembly.GetExecutingAssembly());

    busRegistrationConfig.AddRider(riderConfig =>
    {
        riderConfig.AddProducer<string, SyncCatalogCompletedEvent>("catalog.sync.completed");

        riderConfig.UsingKafka((context, kafkaFactoryConfig) =>
        {
            kafkaFactoryConfig.Host("localhost:9093");

            kafkaFactoryConfig.TopicEndpoint<string, AuditAction>("audit", "booking-api", endpointConfig =>
            {
                endpointConfig.SetKeyDeserializer(new CustomJsonSingleTypeDeserializer<string>().AsSyncOverAsync());
                endpointConfig.SetValueDeserializer(new CustomJsonMultiTypeDeserializer<AuditAction>().AsSyncOverAsync());
                
                endpointConfig.UseConsumeFilter(typeof(LoggingFilter<>), context);

                endpointConfig.ConfigureConsumer<AuditEventConsumer>(context);
            });
        });
    });
});

Kafka event consumer

    public class AuditEventConsumer : IConsumer<AuditAction>
    {
        private readonly ILogger<AuditEventConsumer> _logger;

        public AuditEventConsumer(ILogger<AuditEventConsumer> logger)
        {
            _logger = logger;
        }

        public async Task Consume(ConsumeContext<AuditAction> context)
        {
            _logger.LogInformation("Received event of type ({type})", nameof(AuditAction));

            return Task.CompletedTask;
        }
    }

LoggingFilter

public class LoggingFilter<T> : IFilter<ConsumeContext<T>> where T : class
    {
        private readonly ILogger<LoggingFilter<T>> _logger;

        public LoggingFilter(ILogger<LoggingFilter<T>> logger)
        {
            _logger = logger;
        }

        public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
        {
            var timer = new Stopwatch();
            timer.Start();

            _logger.LogInformation("[FILTER] Start consuming message ID=({id}) of type {type}", context.MessageId, typeof(T));
            try
            {
                await next.Send(context);
            }
            catch (Exception ex)
            {
                _logger.LogError("[FILTER] Message ID=({id}) of type {type} failed due to exception: {ex}", context.MessageId, typeof(T), ex.Message);
            }
            finally
            {
                timer.Stop();
                _logger.LogInformation("[FILTER] Message ID=({id}) of type {type} finished execution. Ellapsed time {time} ms", context.MessageId, typeof(T), timer.ElapsedMilliseconds);
            }
        }

        public void Probe(ProbeContext context)
        {
        }
    }

[phatboyg]

This came up a couple weeks ago, and seems to be an issue. I believe @NooNameR was looking into it.

[phatboyg]

Since a topic endpoint knows the message type, you could just wire it up directly:

endpointConfig.UseFilter(context.GetService<LoggingFilter<AuditAction>());

[NooNameR]

It's not an easy problem to solve. I'll try to take a deeper look when get some spare time.

[phatboyg]

I actually think it's an easy problem, I'm taking a look at it now.

[phatboyg]

So this actually wasn't bad, endpoints were not validating when the rider was built, so none of the configuration observers were called. I've committed a fix, and it should be in the next develop packages shortly.

[benbadura]

I've tried new develop package and it does seem to work as expected. Thanks a lot for your effort and quick fix!