Почему вы не использовали Exactly Once семантику в Kafka?