If I remove ".setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)" it works, but problem is if I run multiple instances of flink job it end up producing duplicate events. Wondering if I am missing any additional/required configuration, but I am clueless as of now, it would be great if anyone has any example configuration on this "DeliveryGuarantee._*EXACTLY_ONCE".*_