In the flink code, I am reading data from a Kafka topic and sinking the results to postgresDB. This works fine. If now I want to check if there exists an entry in the database for the same message that came in from Kafka and then decide if I want to update the row or insert a new row, how can I achieve this?