How to prevent reprocessing of the same Kafka message? Where and how to organize idempotency check on the consumer side?