Esto es lo que Kafka FAQ tiene que decir sobre el tema de exactamente-una vez:

¿Cómo obtengo exactamente un mensaje de Kafka?

Exactamente una semántica tiene dos partes: evitar la duplicación durante la producción de datos y evitar duplicados durante el consumo de datos.

Hay dos enfoques para obtener semántica exactamente una vez durante la producción de datos:

• Utilice un solo escritor por partición y cada vez que obtenga una comprobación de errores de red el último mensaje en esa partición para ver si su última escritura tuvo éxito
• Incluya una clave primaria (UUID o algo así) en el mensaje y deduplice en el consumidor.

Si hace una de estas cosas, el registro que aloja Kafka estará libre de duplicados. Sin embargo, la lectura sin duplicados también depende de la cooperación del consumidor. Si el consumidor está chequeando periódicamente su posición, entonces si falla y se reinicia, se reiniciará desde el checkpointed posición. Por lo tanto, si la salida de datos y el punto de control no se escriben atómicamente, también será posible obtener duplicados aquí. Este problema es particular para su sistema de almacenamiento. Por ejemplo, si está utilizando una base de datos, podría confirmarlos juntos en una transacción. El HDFS loader Camus que LinkedIn escribió hace algo como esto para las cargas de Hadoop. La otra alternativa que no requiere una transacción es almacenar el desplazamiento con los datos cargados y deduplicar utilizando el combinación de tema / partición / desplazamiento.

Creo que hay dos mejoras que harían esto mucho más fácil:

• La idempotencia del productor podría hacerse de forma automática y mucho más barata integrando opcionalmente el soporte para esto en el servidor.
• El consumidor de alto nivel existente no expone mucho del control más fino de las compensaciones (por ejemplo, para restablecer su posición). Estaremos trabajando en eso pronto

Estoy de acuerdo con el deduplicado de RaGe en el lado del consumidor. Y usamos Redis para desduplicar el mensaje de Kafka.

Supongamos que la clase Message tiene un miembro llamado 'uniqId', que es llenado por el lado productor y se garantiza que es único. Usamos una cadena aleatoria de 12 longitudes. (regexp es '^[A-Za-z0-9]{12}$')

El lado consumidor usa SETNX de Redis para deduplicar y EXPIRAR para purgar las claves caducadas automáticamente. Código de ejemplo:

Message msg = ... // eg. ConsumerIterator.next().message().fromJson();
Jedis jedis = ... // eg. JedisPool.getResource();
String key = "SPOUT:" + msg.uniqId; // prefix name at will
String val = Long.toString(System.currentTimeMillis());
long rsps = jedis.setnx(key, val);
if (rsps <= 0) {
    log.warn("kafka dup: {}", msg.toJson()); // and other logic
} else {
    jedis.expire(key, 7200); // 2 hours is ok for production environment;
}

El código anterior detectó mensajes duplicados varias veces cuando Kafka (versión 0.8.x) tuvo situaciones. Con nuestro registro de auditoría de balance de entrada/salida, no se perdió ningún mensaje o ocurrió dup.