Consuma messaggi da Kafka Topic - nessuna risposta

Question

var configs = new Dictionary<string, string>
{
    {"bootstrap.servers", MY_SERVER},
    {"security.protocol", "SASL_PLAINTEXT"},
    {"sasl.mechanism", "SCRAM-SHA-256"},
    {"sasl.username", "MY_USERNAME"},
    {"sasl.password", "MY_PWD"},
    {"group.id", "sample_group"} // added
};
var consumerConfig = new ConsumerConfig(configs);    

using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
using (var consumer = new ConsumerBuilder<string, MyModel>(consumerConfig)
           .SetKeyDeserializer(new AvroDeserializer<string>(schemaRegistry, avroSerializerConfig).AsSyncOverAsync())
           .SetValueDeserializer(new AvroDeserializer<MyModel>(schemaRegistry, avroSerializerConfig).AsSyncOverAsync())
           .Build())
{
      consumer.Subscribe(TOPIC_NAME);

      while (true)
      {
          var result = consumer.Consume(); //stuck here
          Console.WriteLine(result);
      }
 }

Come indicato nel codice, non c'è risposta proveniente da consumer.Consume(). Non genera alcun messaggio di errore anche durante consumer.Subscribe()Quale sarà la possibile ragione? (Sono nuovo su Kafka Consumer)

1. Forse non ci sono messaggi in Topic, quindi niente da ricevere?
2. Il codice richiedeva l'assenza di "group.id", quindi ho aggiunto {"group.id", "sample_group"}config e avvolto con ConsumerConfig. Il nome casuale ("sample_group") è consentito per group.id o dovrebbe essere qualcosa recuperato dalle informazioni dell'argomento?
3. qualunque altra cosa?

Answer 1

Il tuo codice sembra a posto e il fatto che non vengano visualizzati errori ed eccezioni è anche un buon segno.

"1. Forse non ci sono messaggi in Topic, quindi niente da ricevere?"

Anche se non ci sono messaggi nell'argomento Kafka, la tua osservazione corrisponde al comportamento previsto. Nel while(true)ciclo si tenta continuamente di recuperare i dati dall'argomento e se non è possibile recuperare nulla, il consumatore riproverà nell'iterazione successiva. I consumatori su argomenti Kafka sono destinati a leggere un argomento in sequenza mentre corrono continuamente. Va benissimo che alcune volte i consumatori abbiano consumato tutti i messaggi e rimangano inattivi per un po 'di tempo fino a quando un nuovo messaggio non arriva nell'argomento. Durante il tempo di attesa il consumatore non si fermerà né si bloccherà.

Tieni presente che i messaggi in un argomento Kafka hanno per impostazione predefinita un periodo di conservazione di 7 giorni. Trascorso questo tempo, i messaggi verranno eliminati.

"2. Il codice richiedeva l'assenza di" group.id ", quindi ho aggiunto {" group.id "," sample_group "} in config e wrap con ConsumerConfig. Il nome casuale (" sample_group ") è consentito per group.id o dovrebbe si tratta di qualcosa recuperato dalle informazioni sull'argomento? "

Sì, il nome "sample_group" è consentito come nome ConsumerGroup. Non ci sono nomi di gruppi di consumatori riservati, quindi questo nome non causerà alcun problema.

"3. qualcos'altro?"

Per impostazione predefinita, un KafkaConsumer legge i messaggi dall'offset "più recente". Ciò significa che, se esegui un ConsumerGroup per la prima volta, non leggerà tutti i messaggi dall'inizio ma piuttosto dalla fine. Controlla le configurazioni dei consumatori nella documentazione .net Kafka-API per qualcosa di simile auto_offset_reset. È possibile impostare questa configurazione su "prima" nel caso in cui si desideri leggere tutti i messaggi dall'inizio. Tieni presente che non appena esegui la tua applicazione con un dato ConsumerGroup per la prima volta, la seconda volta che esegui questa applicazione questa configurazione auto_offset_resetnon avrà alcun impatto perché il ConsumerGroup è ora registrato in Kafka.

Quello che di solito puoi fare, per assicurarti che il consumatore debba effettivamente leggere i messaggi è se avvii il tuo consumatore prima di iniziare a produrre messaggi su quell'argomento. Quindi, (quasi) indipendentemente dalla configurazione, dovresti vedere i dati che fluiscono attraverso la tua applicazione.