How to control the flow of Kafka messages

Code from : confluent-kafka-dotnet/src/Confluent.Kafka/Consumer.cs at 3f48e8944242abf631ea8d2cd5698f6e149e0731 · confluentinc/confluent-kafka-dotnet

internal static IEnumerable<ConsumeResult<K, V>> ConsumeRecords<K, V>(this IConsumer<K, V> consumer, TimeSpan timeout, long maxRecords)
{
    List<ConsumeResult<K, V>> records = new List<ConsumeResult<K, V>>();
    DateTime dt = DateTime.Now;
    TimeSpan ts = TimeSpan.Zero;
    do{
       var r = consumer.Consume(ts);
       if (r != null)
       {  records.Add(r); }
        else{ // TODO : make log(N) to increse progressivly timeout consumption
             ts = (dt.Add(timeout) - DateTime.Now);
        }
        if (records.Count >= maxRecords)
            break;
      } while (dt.Add(timeout) > DateTime.Now);
       return records;
}

Handling Streaming Data with Messaging Systems ( Kafka & Azure Event Hub)

How to control the flow of Kafka messages

Analytics