[FLINK-3325] [cep] Close input stream in CEPPatternOperator.restoreState

flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CEPPatternOperator.java

@@ -133,5 +133,7 @@ public class CEPPatternOperator<IN> extends AbstractCEPPatternOperator<IN> {
 		for (int i = 0; i <numberPriorityQueueEntries; i++) {
 			priorityQueue.offer(streamRecordSerializer.deserialize(div));
 		}
+
+		div.close();
 	}
 }