Configuración de propiedades de consumidor

Existen muchos ajustes de configuración para el consumidor que controlan aspectos de su comportamiento. Los siguientes valores son algunos de los más importantes.

Hay muchos más valores de configuración disponibles, pero lea la documentación deApache Kafka antes de empezar a experimentar con ellos.

Grupos de consumidores

Un grupo de consumidores es un grupo de consumidores que coopera para consumir mensajes de uno o más temas. Los consumidores de un grupo utilizan todos el mismo valor para la configuración de group.id. Si necesita más de un consumidor para gestionar su carga, puede ejecutar varios consumidores en el mismo grupo de consumidores. Incluso si sólo necesita un consumidor, es habitual especificar también un valor para group.id.

Cada grupo de consumidores tiene un servidor en el clúster que se denomina coordinador responsable de asignar particiones a los consumidores del grupo. Esta responsabilidad se reparte entre los servidores del clúster para repartir la carga. La asignación de particiones a los consumidores puede cambiar en cada reequilibrio del grupo.

Cuando un consumidor se une a un grupo de consumidores, descubre el coordinador del grupo. A continuación, el consumidor indica al coordinador que desea unirse al grupo y el coordinador inicia un reequilibrio de las particiones en el grupo para incluir el nuevo miembro.

Los mensajes de una sola partición son procesados por solo un consumidor en cada grupo. Este mecanismo garantiza que los mensajes de cada partición se procesan en orden. Consulte el diagrama siguiente para ver un ejemplo en el que un tema contiene tres particiones y un grupo de consumidores, que está consumiendo ese tema, contiene dos consumidores. A un consumidor del grupo se le asignan dos particiones y al otro consumidor se le asigna una partición.

Cuando se produce uno de los cambios siguientes en un grupo de consumidores, el grupo se reequilibra cambiando la asignación de particiones a los miembros del grupo para incorporar el cambio:

• Un consumidor se une al grupo.
• Un consumidor abandona el grupo.
• El coordinador considera que un consumidor ha dejado de estar activo.
• Se añaden nuevas particiones a un tema existente.

Si tienes un grupo de consumidores que se reequilibra, ten en cuenta que cualquier consumidor que haya abandonado el grupo tiene sus commits rechazados hasta que se reincorpore al grupo. En este caso, el consumidor deberá volverse a unir al grupo, donde se le podría asignar a una partición distinta de la que estaba consumiendo anteriormente.

Actividad de consumidor

Kafka detecta automáticamente los consumidores fallidos para poder reasignar particiones a los consumidores en funcionamiento. Utiliza dos mecanismos: el sondeo y la pulsación.

Si el lote de mensajes que se devuelven de Consumer.poll(...) es grande o el procesamiento lleva mucho tiempo, el retraso antes de poll() que se vuelva a llamar puede ser significativo o impredecible. En algunos casos, es necesario configurar un intervalo de tiempo máximo de sondeo para que no se elimine a los consumidores de sus grupos sólo porque el procesamiento de mensajes tarda un tiempo. Si este mecanismo es el único disponible, el tiempo que se tarda en detectar un consumidor anómalo también es largo.

Para que la actividad de los consumidores sea más fácil de gestionar, se añadió el latido de fondo en Kafka 0.10.1. El coordinador del grupo espera que los miembros del grupo le envíen latidos regulares para indicar que siguen activos. Una hebra de pulsación de fondo se ejecuta en el consumidor y envía pulsaciones regulares al coordinador. Si el coordinador no recibe una pulsación de un miembro de un grupo dentro del tiempo de espera de sesión, el coordinador elimina el miembro del grupo e inicia un reequilibrio del grupo. El tiempo de espera de la sesión puede ser mucho más corto que el intervalo máximo de sondeo, de modo que el tiempo que se tarda en detectar un consumidor fallido puede ser corto, incluso si el procesamiento de mensajes lleva mucho tiempo.

Puede configurar el intervalo máximo de sondeo utilizando la propiedad max.poll.interval.ms y el tiempo de espera de sesión utilizando la propiedad session.timeout.ms. No es necesario que utilice estos valores a menos que tarde más de 5 minutos en procesar un lote de mensajes.

Gestión de desplazamientos

Para cada grupo de consumidores, Kafka mantiene el desplazamiento confirmado para cada partición que se consume. Cuando un consumidor procesa un mensaje, no lo elimina de la partición. En su lugar, sólo actualiza su desplazamiento actual utilizando un proceso que se llama confirmar el desplazamiento.

Event Streams retiene la información de desplazamientos confirmados durante 7 días.

Control de la velocidad de consumo de mensajes

Si tiene problemas con la gestión de mensajes provocados por la inundación de mensajes, puede establecer una opción de consumidor para controlar la velocidad de consumo de mensajes. Utilice fetch.max.bytes y max.poll.records para controlar el volumen de datos que puede devolver una llamada a poll().

Gestión del reequilibrio de consumidores

Cuando se añaden o eliminan consumidores de un grupo, se produce un reequilibrio del grupo y los consumidores no pueden consumir mensajes. Esto provoca que todos los consumidores de un grupo de consumidores no estén disponibles durante un breve periodo.

Si se le notifica con el callback "on partitions revoked", utilice un ConsumerRebalanceListener para consignar manualmente los offsets (si no está utilizando auto-commit) y para pausar el procesamiento posterior hasta que se le notifique el reequilibrio exitoso mediante el callback "on partition assigned".

Fragmentos de código

Estos fragmentos de código están en un nivel alto para ilustrar los conceptos implicados. Para obtener ejemplos completos, consulte los ejemplos de Event Streams en GitHub.

Para conectar un consumidor a Event Streams, debe crear credenciales de servicio. Para obtener información sobre cómo obtener estas credenciales, consulte Conexión a Event Streams.

En el código de consumidor, primero debe crear el conjunto de propiedades de configuración. Todas las conexiones a Event Streams están aseguradas usando TLS y autenticación usuario-contraseña, por lo que necesitas al menos estas propiedades. Sustituya BOOTSTRAP_ENDPOINTS, USER y PASSWORD por los de sus propias credenciales de servicio:

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

Para consumir mensajes, también es necesario especificar deserializadores para las claves y los valores, como en el siguiente ejemplo.

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

Estos deserializadores deben coincidir con los serializadores utilizados por los productores.

A continuación, utilice un KafkaConsumer para consumir mensajes, donde cada mensaje está representado por un ConsumerRecord. La forma más común de consumir mensajes es poner el consumidor en un grupo de consumidores estableciendo el ID de grupo y después llamar a subscribe() para ver una lista de temas. Al consumidor se le asignan algunas particiones para consumir, aunque si existen más consumidores en el grupo que particiones del tema, es posible que al consumidor no se le asigne ninguna partición. A continuación, llame a poll() en bucle, recibiendo un lote de mensajes para procesar, donde cada mensaje está representado por 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());
}

Este bucle consumidor se ejecuta eternamente pero puede ser interrumpido desde otro hilo llamando a Consumer.wakeup() para conseguir un cierre ordenado.

Para confirmar los desplazamientos manualmente, primero es necesario establecer la configuración de enable.auto.commit en false. A continuación, utilice Consumer.commmitSync() o bien Consumer.commitAsync() para actualizar el desplazamiento confirmado del consumidor periódicamente. Por simplicidad, este ejemplo procesa los registros para cada partición y confirma el último desplazamiento por separado.

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

Manejo de excepciones

Cualquier aplicación robusta que utiliza el cliente Kafka necesita gestionar excepciones para determinadas situaciones previstas. En algunos casos, las excepciones no se generan directamente porque algunos métodos son asíncronos y entregan sus resultados utilizando un Future o una devolución de llamada. Consulte el código de ejemplo en GitHub que muestra ejemplos completos.

Maneje la siguiente lista de excepciones en el código: