Как защититься от повторной обработки одного Kafka-сообщения? Где и как организовать проверку идемпотентности на consumer-стороне?