Konfiguracja kafka-net, aby przestać wysyłać najnowsze wiadomości

Question

Używam kafka 0.8.1.1 na maszynie Red Hat VM z wtyczką kafka-net. Jak mogę skonfigurować mojego klienta, aby przestał otrzymywać wcześniejsze wiadomości od kafka?

Mój kod konsument:

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

Konsument w Kafka-net aktualnie nie automatycznego śledzenia przesunięcia konsumowane. Będziesz musiał ręcznie zaimplementować śledzenie przesunięcia.

aby zapisać przesunięcie w wersji 0.8.1: kafka

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(); 

Aby ustawić konsumenta, aby rozpocząć importowanie w określonym przesunięciem punktu:

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); 

Uwaga powyższy kod będzie ustawiony konsumenta aby rozpocząć konsumpcję na samym końcu dziennika, skutecznie tylko odbierając nowe wiadomości.