Son mesajları göndermeyi durdurmak için kafka-net'i yapılandır

Question

Kafka-net eklentisi ile Red Hat VM üzerinde kafka 0.8.1.1 kullanıyorum. Tüketicimi kafka'dan daha önceki iletileri almayı durdurmak için nasıl yapılandırabilirim?

Benim tüketici kodu:

var options = new KafkaOptions(new Uri("tcp://199.53.249.150:9092"), new Uri("tcp://199.53.249.151:9092")); 

Stopwatch sp = new Stopwatch(); 
var router = new BrokerRouter(options); 
var consumer = new Consumer(new ConsumerOptions("Test", router)); 

ThreadStart start2 =() => 
{ 
    while (true) 
    { 
     sp.Start(); 
     foreach (var message in consumer.Consume()) 
     { 
      if (MessageDecoderReceiver.MessageBase(message.Value) != null) 
      { 
       PrintMessage(MessageDecoderReceiver.MessageBase(message.Value).ToString()); 
      } 
      else 
      { 
       Console.WriteLine(message.Value); 
      } 
     } 
     sp.Stop(); 
    } 
}; 
var thread2 = new Thread(start2); 
thread2.Start(); 

Answer 1

Kafka-net Tüketici anda otomatik uzaklıklar tüketildiğini takip etmez. Ofset izlemeyi manuel olarak uygulamanız gerekecektir. Yukarıdaki kod tüketiciyi ayarlayacaktır

var offsets = consumer.GetTopicOffsetAsync(IntegrationConfig.IntegrationTopic).Result 
        .Select(x => new OffsetPosition(x.PartitionId, x.Offsets.Max())).ToArray(); 

var consumer = new Consumer(new ConsumerOptions(IntegrationConfig.IntegrationTopic, router), offsets); 

Not:

var commit = new OffsetCommitRequest 
      { 
       ConsumerGroup = consumerGroup, 
       OffsetCommits = new List<OffsetCommit> 
          { 
           new OffsetCommit 
            { 
             PartitionId = partitionId, 
             Topic = IntegrationConfig.IntegrationTopic, 
             Offset = offset, 
             Metadata = metadata 
            } 
          } 
      }; 

var commitResponse = conn.Connection.SendAsync(commit).Result.FirstOrDefault(); 

belirli ofset noktada ithalatına başlamak tüketiciyi ayarlamak için:

kafka sürümü 0.8.1 de ofset Mağaza Günlüğün en sonunda tüketmeye başlamak, sadece yeni mesajları almak.