Does batch processing respect IgnoreUnhandledMessages?

[cookie-bytes]

Hi,

Is batch processing designed to adhere to the IgnoreUnhandledMessages setting? We seem to be encountering an “UnhandledMessageException”, despite our batch consumer being configured to ignore such messages.

Thanks,

[BEagle1984]

It should respect it, yes.
If it doesn't, that would definitely be a bug.
Let me know if you really are experiencing this. Further details about the scenario or some small sample code would be helpful to troubleshoot.

[miaw7]

Hello @BEagle1984

I was debugging your code a little and can confirm that it throws UnhandledErrorException for batch consumer even if IgnoreUnhandledMessages used. For simple (not batch) consumer it's not reproducible.

In the code i'm using Avro messages. To reproduce this i'm sending to a topic message with a type which can't be recognized by any consumer. For example consumer expects to see Car object but i send to topic Bicycle object.

My code setup:

public class EndpointsConfigurator : IEndpointsConfigurator
{
    public void Configure(IEndpointsConfigurationBuilder builder)
    {
        builder
            .AddKafkaEndpoints(
                endpoints => endpoints
                    .Configure(
                        config =>
                        {
                            config.BootstrapServers = "kafka:9092";
                            config.SecurityProtocol = SecurityProtocol.Plaintext;
                            config.AllowAutoCreateTopics = true;
                        })
                    .AddInbound(
                        endpoint => endpoint
                            .ConsumeFrom("samples-batch")
                            .DeserializeAvro(
                                avro =>
                                {
                                    avro.UseType<GenericRecord>()
                                        .Configure(
                                        schemaRegistry => { schemaRegistry.Url = "http://schema-registry:8081"; });
                                })
                            .IgnoreUnhandledMessages()
                            .DisableMessageValidation()
                            .Configure(
                                config =>
                                {
                                    config.GroupId = "sample-consumer";
                                    config.AutoOffsetReset = AutoOffsetReset.Earliest;
                                    config.AllowAutoCreateTopics = true;
                                })
                            .EnableBatchProcessing(1, TimeSpan.FromSeconds(5)))
                    );
    }
}

GenericRecord i used in the serializer to be sure all events will be unhandled

Consumer:

    public class SampleMessageBatchSubscriber
    {
        public async Task OnBatchReceivedAsync(IAsyncEnumerable<Car> batch)
        {
            int count = 0;

            await foreach (var message in batch)
            {
                count++;
            }
        }
    }

My silverback registration:

        services
            .AddSilverback()
            .WithConnectionToMessageBroker(
                options => options
                    .AddKafka())
            .AddEndpointsConfigurator<EndpointsConfigurator>()
            .AddSingletonSubscriber<SampleMessageBatchSubscriber>();

To publish messages to the topic i'm using kafka-avro-console-producer tool.

What i have noticed in your code
In Publisher.cs

I'm triggering this breakpoint several times per one message. It says handled true, but throwIfUnhandled false as expected.

Then after this method execution my breakpoint being triggered in MessageStreamProvider.cs

And in this method throwIfUnhandled is set to true and it throws an UnhandledMessageException not counting that we have enabled "IgnoreUnhandledMessages".

I assume that it should be false to work as expected

[BEagle1984]

You uploaded the same screenshot twice so I don't see where you placed the second breakpoint.

Would it be possible to send the repro code so I can debug it myself?

edit: btw, thank you for taking the time to dig deeper. ❤️

[miaw7]

Updated the picture, sorry

[miaw7]

@BEagle1984 This is my testing proj
SilverbackTestingProj.zip

As an entry point i'm using docker project inside of the solution.

To publish kafka event i'm using avro console tool inside schema-registry container (attaching shell to container)

This is an example of the command for avro console tool

./kafka-avro-console-producer --broker-list kafka:9092 --topic samples-batch --property schema.registry.url=http://schema-registry:8081/ --property value.schema="{\"type\":\"record\",\"name\":\"Car\",\"namespace\":\"SilverbackTesting.Event\",\"fields\":[{\"name\":\"id\",\"type\":{\"type\":\"string\",\"logicalType\":\"uuid\"}},{\"name\":\"model\",\"type\":\"string\"}]}"

{"id":"b50595c3-4f0b-4c07-885b-c2eb55c0ecf9","model":"somemodel"}

Currently in endpoint configurator i'm using GenericRecord type for deserializer. It will produce UnhandledMessageException with any avro event

[BEagle1984]

Nevermind, sorry.

[BEagle1984]

The bug has been fixed.
The fix will be released with the upcoming v4.4.0.

[cookie-bytes]

Great, thank you