kafka KafkaConsumer Complete

KafkaMessageConsumer.cs

public void Consume(){
    var config = new ConsumerConfig
    {
       BootstrapServers = _bootstrapServers,
       GroupId = _groupId,
       AutoOffsetReset = _autoOffsetReset
     };      
    
     try{ 
       _consumer = new ConsumerBuilder<String, String>(config).Build();
       _consumer.Subscribe(_topics);
       _isConsuming = true;
       int i = 0;
       while (_isConsuming){
         i++;
         Console.WriteLine(i + ": ");
         ConsumeResult<String, String> consumeResult = _consumer.Consume(_cts.Token);
         Console.WriteLine(consumeResult.Message.Value);
       }
     }
     catch (OperationCanceledException ex){
       Console.WriteLine("Application was ended: " + ex.Message.ToString());
     }
     catch(Exception ex){
       Console.WriteLine("Application Crashed: " + ex.Message);
     }
     finally{                                                               
       if (_consumer != null){
         _consumer.Close();
         ((IDisposable)_consumer).Dispose();
       }
     }
}