Make sure SDF unbounded wrapper close reader properly

At least it seems like when checkpoint happens, we set the currentReader to empty source reader directly without closing it properly:
https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/io/Read.java#L869-L870