Consume from different partitions in parallel

[sfiodorov]

Hi.
My setup is:

 public static IServiceCollection AddKafkaFlow(this IServiceCollection serviceCollection)
        {
            return
             serviceCollection.AddKafkaFlowHostedService(kafka => kafka
                 // use serilog handler
                 .UseMicrosoftLog()
                 //.UseConsoleLog()
                 .AddCluster(cluster => cluster
                     .WithBrokers(new[] { "localhost:9092" })

                     // admin stuff
                     .EnableAdminMessages("kafka-flow.admin")
                     .EnableTelemetry("kafka-flow.admin")
                     .AddConsumer(consumer => consumer
                     .WithStatisticsIntervalMs(10)

                         .Topic(SharedTypes.Constants.TopicName)
                         .WithGroupId("sample-group")
                         .WithBufferSize(100)
                         .WithWorkersCount(2)
                         .WithAutoOffsetReset(KafkaFlow.AutoOffsetReset.Earliest)
                         .WithPartitionsAssignedHandler((x, y) =>
                         {
                             foreach (var tp in y)
                             {
                                 Console.WriteLine($"ASSIGNED: {tp.Topic}:{tp.Partition.Value}");
                             }
                         })
                         // .WithConsumerConfig(new ConsumerConfig() { Debug = "topic,consumer" })
                         .AddMiddlewares(middlewares => middlewares
                             // .AddSerializer<KafkaFlow.Serializer.NewtonsoftJsonSerializer>()
                             .AddSingleTypeSerializer<TestMessage, KafkaFlow.Serializer.NewtonsoftJsonSerializer>()
                         .AddTypedHandlers(handlers => handlers
                             .AddHandler<SampleConsumer>()
                             .WhenNoHandlerFound((context) =>
                             {
                                 Console.WriteLine($"No message handler found to the message {context.Message.Value.GetType().FullName} ");
                             })
                         )
                         )
                     )
                 )
             );
        }

Consumer:

 public class SampleConsumer : IMessageHandler<TestMessage>
    {
        private readonly ILogger<SampleConsumer> _logger;

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

        public async Task Handle(IMessageContext context, TestMessage message)
        {
            _logger.LogInformation($"PARTITION: {context.ConsumerContext.Partition},Consumed message id: {message.Id}, payload: {message.Text}");
            await Task.Delay(5000);


        }
    }

Topic with 5 partitions, each has messages.
My expectation is that each partition will be processed in parallel by the assigned group partition.
So I expect to see logs that all partitioning are processed.
But I see only log from the consumer attached to partition 0.
When partition 0 is processed I am getting logs about partition 1.
What am I missing in the configuration?
My target is to process partitions in parallel to get horizontal scale

[filipeesch]

Hi @sfiodorov, the Confluent driver that controls how the messages will be read from the topic/partition. In a development environment, some strange behaviors can happen, like yours. In a production environment, the "things" happen differently, you have many producers, many application instances, more partitions, and messages begin produced in different partitions at the same time. I think this problem will not happen. Putting the consumer with two workers, KafkaFlow will process two messages at the same time, even if the topic has just one partition.

[sfiodorov]

@filipeesch thanks for the clarification about the original issue. You are correct this is by the design of librdkafka.

But not sure i understand about the workers you mentioned.
If my consumer is defined with 10 workers and the topic has only one partition. You say that 10 messages from the same partition will be processed simultaneously. In such a case i will lose the strict ordering promised by Kafka, am i right?
How should I define the number of workers in the case of one partition and in the case of multiple partitions?

[filipeesch]

KafkaFlow, by default, uses the message key to choose a worker to process the message, so, the message order is guaranteed when they share the same key, the message partition is irrelevant. This animation can help a little to understand: https://github.com/Farfetch/kafka-flow/wiki/consumers#how-it-works

[sfiodorov]

Now its clear. Thanks.