Konfiguriere kafka-net um das Senden der letzten Nachrichten zu beenden

Question

Ich benutze kafka 0.8.1.1 auf einer Red Hat VM mit kafka-net plugin. Wie kann ich meinen Kunden so konfigurieren, dass er keine früheren Nachrichten von kafka mehr empfängt?

Mein Verbraucher Code:

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

Der Verbraucher in Kafka-net Auto verfolgt derzeit nicht die Offsets verbraucht wird. Sie müssen die Offset-Verfolgung manuell implementieren.

die in Kafkas Version Offset So speichern 0.8.1:

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

den Verbraucher einzustellen, bei einem bestimmten Offset-Punkt zu starten importieren:

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

Beachten Sie den obigen Code wird der Verbraucher festgelegt am Ende des Logs zu konsumieren, effektiv nur neue Nachrichten zu empfangen.