Come prevenire la rielaborazione dello stesso messaggio Kafka? Dove e come organizzare il controllo di idempotenza sul lato del consumatore?