How can I use an upsert statement in Flink when using datastream api to read data from Kafka? For example: if I read a message from Kafka, I insert it into the database or else I check if there exists an entry for the same record, I update the record. How can I achieve this?