[FLINK-1325] [streaming] Added clousure cleaning to streaming