Configurazione delle proprietà del consumatore

Esistono molte impostazioni di configurazione per il consumer che controllano gli aspetti del suo funzionamento. Le seguenti impostazioni sono alcune delle più importanti.

Sono disponibili molte altre impostazioni di configurazione, ma leggi la documentazione diApache Kafka prima di iniziare a provarle.

Gruppi di consumer

Un gruppo di consumatori è un gruppo di consumatori che collabora per consumare i messaggi di uno o più argomenti. I consumatori in un gruppo utilizzano tutti lo stesso valore per la configurazione group.id. Se ti serve più di un consumatore per gestire il tuo carico di lavoro, puoi eseguire più consumatori nello stesso gruppo di consumatori. Anche se si ha bisogno di un solo consumatore, è normale specificare anche un valore per group.id.

Ogni gruppo di consumatori ha un server nel cluster, chiamato coordinatore, responsabile dell'assegnazione delle partizioni ai consumatori del gruppo. Questa responsabilità è distribuita tra i server nel cluster per uniformare il carico. L?assegnazione di partizioni ai consumatori può cambiare a ogni riequilibratura dei gruppi.

Quando si unisce a un gruppo di consumatori, un consumatore scopre il coordinatore per il gruppo. Il consumatore comunica quindi al coordinatore di volersi unire al gruppo e il coordinatore avvia un ribilanciamento delle partizioni del gruppo per includere il nuovo membro.

I messaggi provenienti da una singola partizione vengono elaborati da un solo consumatore in ogni gruppo. Questo meccanismo garantisce che i messaggi di ogni partizione vengano elaborati in ordine. Consultare il diagramma riportato di seguito per un esempio in cui un argomento contiene tre partizioni e un gruppo di consumatori, che utilizza tale argomento, contiene due consumatori. A un consumer nel gruppo vengono assegnate due partizioni e all'altro consumer viene assegnata una partizione.

Quando in un gruppo di consumatori si verifica una delle seguenti modifiche, il gruppo si riequilibra spostando l'assegnazione delle partizioni ai membri del gruppo per adattarsi alla modifica:

• Un consumatore si unisce al gruppo.
• Un consumatore lascia il gruppo.
• Un consumatore è considerato non più in vita dal coordinatore.
• Vengono aggiunte nuove partizioni a un tema esistente.

Se avete un gruppo di consumatori che viene riequilibrato, sappiate che qualsiasi consumatore che ha lasciato il gruppo ha i suoi commit che vengono rifiutati fino a quando non si ricongiunge al gruppo. in questo caso, il consumatore deve unirsi nuovamente al gruppo, dove gli potrebbe essere assegnata una partizione diversa da quella da cui stava consumando in precedenza.

Attività del consumatore

Kafka rileva automaticamente i consumatori in errore in modo da poter riassegnare le partizioni ai consumatori funzionanti. Utilizza due meccanismi: il polling e l'heartbeat.

Se il batch di messaggi restituito da Consumer.poll(...) è grande o l'elaborazione richiede tempo, il ritardo prima che poll() venga richiamato di nuovo può essere significativo o imprevedibile. In alcuni casi, è necessario configurare un lungo intervallo di polling massimo in modo che i consumatori non vengano rimossi dai loro gruppi solo perché l'elaborazione dei messaggi sta impiegando del tempo. Se questo meccanismo è l'unico disponibile, anche il tempo impiegato per rilevare un consumer non riuscito è lungo.

Per rendere lo stato di attività del consumatore facile da gestire, in Kafka 0.10.1 è stato aggiunto un heartbeat di background. Il coordinatore del gruppo prevede che i membri del gruppo gli inviino degli heartbeat regolari per indicare che rimangono attivi. Un thread di heartbeat in background viene eseguito nel consumatore e invia regolarmente heartbeat al coordinatore. Se il coordinatore non riceve un heartbeat da un membro del gruppo entro il timeout della sessione, il coordinatore rimuove il membro dal gruppo e avvia un ribilanciamento del gruppo. Il timeout di sessione può essere molto più breve dell'intervallo massimo di polling, in modo che il tempo necessario per rilevare un consumatore fallito possa essere breve, anche se l'elaborazione dei messaggi richiede molto tempo.

È possibile configurare l'intervallo massimo di polling mediante la proprietà max.poll.interval.ms e il timeout della sessione mediante la proprietà session.timeout.ms. Non è necessario utilizzare queste impostazioni a meno che non ci vogliano più di 5 minuti per elaborare un gruppo di messaggi.

Gestione degli offset

Per ogni gruppo di consumatori, Kafka mantiene l'offset impegnato per ogni partizione consumata. Quando un consumatore elabora un messaggio, non lo rimuove dalla partizione. Invece, si limita ad aggiornare l'offset corrente utilizzando un processo che viene chiamato "committing" dell'offset.

Event Streams conserva le informazioni sugli offset di cui è stato eseguito il commit per 7 giorni.

Controllo della velocità di consumo dei messaggi

Se si verificano problemi di gestione dei messaggi causati dall'ingolfamento degli stessi, è possibile impostare un'opzione consumer per controllare la velocità di consumo dei messaggi. Utilizza fetch.max.bytes e max.poll.records per controllare quanti dati può restituire una chiamata a poll().

Gestione del ribilanciamento dei consumatori

Quando i consumatori vengono aggiunti o rimossi da un gruppo, si verifica un riequilibrio del gruppo e i consumatori non sono in grado di consumare i messaggi. In questo modo tutti i consumatori di un gruppo di consumatori non sono disponibili per un breve periodo.

Se si riceve la notifica con il callback "on partitions revoked", utilizzare un ConsumerRebalanceListener per impegnare manualmente gli offset (se non si utilizza l'autocommit) e per sospendere l'ulteriore elaborazione fino alla notifica dell'avvenuto ribilanciamento tramite il callback "on partition assigned".

Frammenti di codice

Questi frammenti di codice sono di alto livello per illustrare i concetti coinvolti. Per esempi completi, vedi gli esempi di Event Streams in GitHub.

Per connettere un consumer a Event Streams, devi creare le credenziali del servizio. Per informazioni su come ottenere queste credenziali, vedi Connessione a Event Streams.

Nel codice consumer, è necessario prima creare la serie di proprietà di configurazione. Tutte le connessioni a Event Streams sono protette dall'uso di TLS e dall'autenticazione utente-password, per cui sono necessarie almeno queste proprietà. Sostituire BOOTSTRAP_ENDPOINTS, USER e PASSWORD con le credenziali del proprio servizio:

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

Per utilizzare i messaggi, è anche necessario specificare deserializzatori per le chiavi e i valori, come nel seguente esempio.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

Questi deserializer devono corrispondere ai serializer utilizzati dai producer.

Usa quindi un KafkaConsumer per consumare i messaggi, dove ogni messaggio è rappresentato da un ConsumerRecord. Il modo più comune per consumare i messaggi consiste nell'inserire il consumatore in un gruppo di consumatori impostando l'ID gruppo e richiamare quindi subscribe() per un elenco di argomenti. Al consumatore vengono assegnate alcune partizioni da consumare, anche se se nel gruppo esistono più consumatori che partizioni nell'argomento, al consumatore potrebbe non essere assegnata alcuna partizione. Richiama quindi poll() in un loop, ricevendo un batch di messaggi da elaborare, dove ogni messaggio è rappresentato da un ConsumerRecord.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

Questo ciclo consumer viene eseguito per sempre, ma può essere interrotto da un altro thread chiamando Consumer.wakeup() per ottenere una chiusura ordinata.

Per eseguire il commit degli offset manualmente, è prima necessario impostare la configurazione enable.auto.commit su false. Usa quindi Consumer.commmitSync() o Consumer.commitAsync() per aggiornare periodicamente l'offset di cui è stato eseguito il commit del consumatore. Per semplicità, questo esempio elabora i record per ciascuna partizione ed esegue il commit dell'ultimo offset separatamente.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

Gestione delle eccezioni

Qualsiasi solida applicazione che utilizza il client Kafka deve gestire le eccezioni per delle specifiche situazioni previste. In alcuni casi, le eccezioni non vengono lanciate direttamente, perché alcuni metodi sono asincroni e forniscono i loro risultati utilizzando un Future o un callback. Controlla il codice di esempio in GitHub che mostra gli esempi completi.

Gestire il seguente elenco di eccezioni nel codice: