CODEWITHSUNDEEP
c#kafka-consumer-api
Chris Dunlop
Chris Dunlop

Reputation: 145

How does the offset for a topic work in Kafka (Kafka_net)

I have a basic producer app and a consumer app. if I run both and have both start consuming on their respective topics, I have a great working system. My thought was that if I started the producer and sent a message that I would be able to then start the consumer and have it pick up that message. I was wrong.

Unless both are up and running, I lose messages (or they do not get consumed).

my consumer app looks like this for comsuming...

Uri uri = new Uri("http://localhost:9092");

KafkaOptions options = new KafkaOptions(uri);                
BrokerRouter brokerRouter = new BrokerRouter(options);
Consumer consumer = new Consumer(new ConsumerOptions(receiveTopic, brokerRouter));
             

List<OffsetResponse> offset = consumer.GetTopicOffsetAsync(receiveTopic, 100000).Result;

IEnumerable<OffsetPosition> t = from x in offset select new OffsetPosition(x.PartitionId, x.Offsets.Max());
consumer.SetOffsetPosition(t.ToArray());
IEnumerable<KafkaNet.Protocol.Message> msgs = consumer.Consume();

foreach (KafkaNet.Protocol.Message msg in msgs)
{
    do some stuff here based on the message received
}

unless I have the code between the lines, it starts at the beginning every time I start the application. What is the proper way to manage topic offsets so messages are consumed after a disconnect happens?

If I run

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic chat-message-reply-XXX consumer-property fetch-size=40000000 --from-beginning

I can see the messages, but when I connect my application to that topic, the consumer.Consume() does not pick up the messages it has not already seen. I have tried this with and without runing the above bat file to see if that makes any difference. When I look at the consumer.SetOffsetPosition(t.ToArray()) call (t specifically) it shows that the offset is the count of all messages for the topic.

Please help,

Upvotes: 2

Views: 2623

Answers (2)

Chris Dunlop
Chris Dunlop

Reputation: 145

I switched over to use Confluent's C# .NET package and it now works.

Upvotes: 0

ndogac
ndogac

Reputation: 1245

Set auto.offset.reset configuration in your ConsumerOptions to earliest. When the consumer group starts the consume messages, it will consume from the latest offset because the default value for auto.offset.reset is latest.

But I looked at kafka-net API now, it does not have a AutoOffsetReset property, and it seems pretty insufficient with its configuration in consumers. It also lacks documentation with method summaries.

I would suggest you use Confluent .NET Kafka Nuget package because it is owned by Confluent itself.

Also, why are calling GetTopicOffsets and setting that offset back again in consumer. I think when you configure your consumer, you should just start reading messages with Consume().

Try this:

static void Main(string[] args)
{
    var uri = new Uri("http://localhost:9092");

    var kafkaOptions = new KafkaOptions(uri);                
    var brokerRouter = new BrokerRouter(kafkaOptions);
    var consumerOptions = new ConsumerOptions(receivedTopic, brokerRouter);
    var consumer = new Consumer(consumerOptions);

    foreach (var msg in consumer.Consume())
    {
        var value = Encoding.UTF8.GetString(msg.Value);

        // Process value here
    }
}

In addition, enable logs in your KafkaOptions and ConsumerOptions, they will help you a lot:

var kafkaOptions = new KafkaOptions(uri)
            {
                Log = new ConsoleLog()
            };
var consumerOptions = new ConsumerOptions(topic, brokerRouter)
            {
                Log = new ConsoleLog()
            });

Upvotes: 1

Related Questions