refactor FilteredDataSink to use BufferedDataSink. Fix up BufferedDataSink.close to close once the buffer is flushed.