Consumer does not receive messages

[adedw]

I read the documentation and am trying to make a simple consumer. I encountered a problem that the consumer does not receive messages sent via not-KafkaFlow (Tried publishing using Kafka-UI, a Scala service).

Test project SDK Microsoft.NET.Sdk.Worker, TargetFramework net8.0.

Related nugets:

• KafkaFlow.Extensions.Hosting v3.0.7
• KafkaFlow.LogHandler.Microsoft v3.0.7
• KafkaFlow.Serializer.JsonCore v3.0.7

Program.cs

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddKafkaFlowHostedService(kafka => kafka
    .UseMicrosoftLog()
    .AddCluster(cluster => cluster
        .CreateTopicIfNotExists("my-messages", 2, 1)
        .WithBrokers(["localhost:9092"])
        .AddConsumer(consumer => consumer
            .Topic("my-messages")
            .WithGroupId("my-message-consumers")
            .WithBufferSize(100)
            .WithWorkersCount(2)
            .WithAutoOffsetReset(AutoOffsetReset.Earliest)
            .WithManualMessageCompletion()
            .AddMiddlewares(middlewares => middlewares
                .AddDeserializer<JsonCoreDeserializer>()
                .AddTypedHandlers(handlers => handlers.AddHandler<LogMyMessageHandler>())
            ))));

var host = builder.Build();
host.Run();

LogMyMessageHandler.cs

public class LogMyMessageHandler(ILogger<LogMyMessageHandler> logger) : IMessageHandler<Shared.MyMessage>
{
  public Task Handle(IMessageContext context, Shared.MyMessage message)
  {
      logger.LogInformation(EventIds.MessageReceived, "Id: {Id}, text: {Text}", message.Id, message.Text);
      context.ConsumerContext.Complete();
      return Task.CompletedTask;
  }
}

MyMessage.cs

public class MyMessage
{
    [JsonPropertyName("id")] public required Guid Id { get; init; }
    [JsonPropertyName("text")] public required string Text { get; init; }
}

docker-compose.yml

services:
kafka:
  container_name: kafka
  image: bitnami/kafka:3.6.1
  ports:
    - "127.0.0.1:9092:9092"
  environment:
    KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,CONTROLLER:PLAINTEXT
    KAFKA_CFG_LISTENERS: INTERNAL://kafka:9092,CONTROLLER://kafka:9093
    KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka:9092
    KAFKA_CFG_INTER_BROKER_LISTENER_NAME: INTERNAL
    KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
    KAFKA_KRAFT_CLUSTER_ID: LelM2dIFQkiUFvXCEcqRWA
    KAFKA_CFG_NODE_ID: 0
    KAFKA_CFG_PROCESS_ROLES: controller,broker
    KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093

kafka-ui:
  image: provectuslabs/kafka-ui:v0.7.2
  container_name: kafka-ui
  ports:
    - "127.0.0.1:8080:8080"
  environment:
    KAFKA_CLUSTERS_0_NAME: kafka-flow-test
    KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
  depends_on:
    - kafka

Topic state (test1 and test3 added via Kafka-UI, numbers via test kafkaflow producer):

Offset	Partitions	Value
0	0	Text: "test1"
1	0	Text: "9893"
0	1	Text: "5214"
2	0	Text: "6996"
1	1	Text: "3405"
2	1	Text: "4717"
3	0	Text: "4387"
4	0	Text: "478"
3	1	Text: "270"
5	0	Text: "test3"

Kafka logs:

[2024-04-16 19:20:39,017] INFO [GroupCoordinator 0]: Dynamic member with unknown member id joins group my-message-consumers in Empty state. Created a new member id rdkafka-2ca92db8-3506-4eec-a713-0d5fd598c9ec and request the member to rejoin with this id. (kafka.coordinator.group.GroupCoordinator)
[2024-04-16 19:20:39,018] INFO [GroupCoordinator 0]: Preparing to rebalance group my-message-consumers in state PreparingRebalance with old generation 6 (__consumer_offsets-6) (reason: Adding new member rdkafka-2ca92db8-3506-4eec-a713-0d5fd598c9ec with group instance id None; client reason: not provided) (kafka.coordinator.group.GroupCoordinator)
[2024-04-16 19:20:42,018] INFO [GroupCoordinator 0]: Stabilized group my-message-consumers generation 7 (__consumer_offsets-6) with 1 members (kafka.coordinator.group.GroupCoordinator)
[2024-04-16 19:20:42,020] INFO [GroupCoordinator 0]: Assignment received from leader rdkafka-2ca92db8-3506-4eec-a713-0d5fd598c9ec for group my-message-consumers for generation 7. The group has 1 members, 0 of which are static. (kafka.coordinator.group.GroupCoordinator)

Each run, got following consumer logs:

info: KafkaFlow[0]
      Partitions assigned | Data: {"GroupId":"my-message-consumers","ConsumerName":"a31df64e-c966-4e33-87fa-b48aa33e3cc4","Topics":[{"Topic":"my-messages","PartitionsCount":2,"Partitions":[0,1]}]}
info: KafkaFlowTestConsumer.LogMyMessageHandler[1000]
      Id: 4791075c-814b-4a42-8dcb-3b9fc56bea55, text: 9893
info: KafkaFlowTestConsumer.LogMyMessageHandler[1000]
      Id: 4730a3b5-1ed4-4365-a8e8-dd4850994b48, text: 6996
info: KafkaFlowTestConsumer.LogMyMessageHandler[1000]
      Id: ad2b1c54-10c2-4524-a757-0d58c6ecc845, text: 4387
info: KafkaFlowTestConsumer.LogMyMessageHandler[1000]
      Id: ab11f80c-fc49-449d-941f-edb8514ae2db, text: 478

But there is no lag

$ kafka-consumer-groups.sh --bootstrap-server kafka:9092 --describe --group my-message-consumers

GROUP                TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                  HOST            CLIENT-ID
my-message-consumers my-messages     0          -               6               -               rdkafka-c1c000d0-d7a7-4e4c-b08f-04ecbcd1be66 /172.29.0.1     rdkafka
my-message-consumers my-messages     1          4               4               0               rdkafka-c1c000d0-d7a7-4e4c-b08f-04ecbcd1be66 /172.29.0.1     rdkafka

Looks like data from 0 partition, but produced only by KafkaFlow service.

What am i doing wrong?