Fast clean unused direct memory when decompress

[asfimport]

When use NonBlockedDecompressorStream which call:

SnappyDecompressor.setInput

public synchronized void setInput(byte[] buffer, int off, int len) {
 SnappyUtil

public synchronized void setInput(byte[] buffer, int off, int len) {
 SnappyUtil.validateBuffer(buffer, off, len);

 if (inputBuffer.capacity() - inputBuffer.position() < len) {
 ByteBuffer newBuffer = ByteBuffer.allocateDirect(inputBuffer.position() + len);
 inputBuffer.rewind();
 newBuffer.put(inputBuffer);
 inputBuffer = newBuffer; 
 } else {
 inputBuffer.limit(inputBuffer.position() + len);
 }
 inputBuffer.put(buffer, off, len);
}

.validateBuffer(buffer, off, len);

 if (inputBuffer.capacity() - inputBuffer.position() < len) {
 ByteBuffer newBuffer = ByteBuffer.allocateDirect(inputBuffer.position() + len);
 inputBuffer.rewind();
 newBuffer.put(inputBuffer);
 inputBuffer = newBuffer; 
 } else {
 inputBuffer.limit(inputBuffer.position() + len);
 }
 inputBuffer.put(buffer, off, len);
}

If we do not get any full gc for old gen.we may failed by off-heap memory leak
 

Reporter: zhoukang

Related issues:

• Snappy Decompressor/Compressor may cause direct memory leak (duplicates)

PRs and other links:

• GitHub Pull Request #492

_{Note: This issue was originally created as PARQUET-1320. Please see the migration documentation for further details.}