Consumidor de Kafka: ¿cómo comprobar si todos los mensajes en la partición del tema se han consumido completamente?
¿Existen API o atributos que se puedan utilizar o comparar para determinar si se han consumido todos los mensajes en una partición de un tema? Estamos trabajando en una prueba que utilizará otro consumidor del mismo grupo de consumidores para comprobar si la partición del tema aún tiene algún mensaje o no. Uno de nuestros servicios de aplicaciones también utiliza Kafka para procesar eventos internos. ¿Existe alguna manera de sincronizar el progreso del consumo de mensajes?
davy.ai
Sí, existen varias APIs y atributos que se pueden utilizar para determinar si todos los mensajes en una partición del tema han sido consumidos.
El método KafkaConsumer.position() de puede utilizarse para obtener el desplazamiento del último mensaje consumido en una partición del tema. Luego, se puede comparar este desplazamiento con el método KafkaConsumer.endOffsets() para determinar si todos los mensajes en la partición del tema han sido consumidos.
Si está usando Kafka Streams, puede utilizar el método KafkaStreams.allMetadata() para obtener el desplazamiento actual de todas las particiones en el tema. Luego, se puede comparar este desplazamiento con el método KafkaStreams.endOffsets() para determinar si todos los mensajes en la partición del tema han sido consumidos.
Para sincronizar el progreso del consumo de mensajes, puede utilizar los métodos KafkaConsumer.commitSync() o KafkaConsumer.commitAsync() para confirmar el desplazamiento actual del grupo de consumidores. Esto asegurará que todos los consumidores en el grupo estén sincronizados y hayan consumido los mismos mensajes.