Best way to upsetting/delete records in Ignite from Kafka Stream?

[John Smith]

Hi, I have a bunch of Json records in Kafka. I would like to either UPSERT
or DELETE a record from my Ignite cache based on the "type" specified in
the Json record. What's the best way to do this or what feature of
Kafka/Ignite I should use?