delayed kafkastream close