Change KafkaSinglePortOutputOperator to using latch #53