Demonstrate race condition to discuss how to fix #91

[themodernlife]

I was kind of glad to see #91, because I too have ran into this issue while processing LZO files (from S3) via Spark using local config with multiple threads.

I dug into this and the issue seems to be in

hadoop-lzo/src/main/java/com/hadoop/compression/lzo/LzopInputStream.java

Line 349 in e8c11c2

CodecPool.returnDecompressor(decompressor);

.

When using Spark, the exception occurs because the TextInputFormat creates a LineRecordReader that loads the decompressor from the pool. When it gets closed, the LineRecordReader adds it back to the pool... the trouble is that underlying LzopInputStream also adds the decompressor back to the pool, which means the same decompressor is now in the pool twice.

This means that the next 2 calls to getRecordReader are going to use the same decompressor, and that's where you get the race condition.

I attached a unit test that demonstrates the error. The fix here is just to remove the bit that adds the decompressor back to the pool on 349.

Can someone have a look at this as well and let me know what you think? I can clean up the PR and resubmit, but just wanted to add some info about the bug.

Cheers!

[sjlee]

Thanks for the report @themodernlife. We'll look into it.

[themodernlife]

Hey gang, just wondering if you guys have any feedback on this?

[sjlee]

We haven't been able to spend much time on this. At the high level we recognize the issue, but it might be tricky/expensive to fix as users may have code written to work around this and we shouldn't break them...

[rangadi]

not used?

[rangadi]

sorry, can you point to the fix? I must be missing obvious.

[sjlee]

@rangadi, this particular pull request was just to demonstrate the bug itself. I think @themodernlife's suggestion is to not pool decompressors (which he added and commented out in this PR).

[sjlee]

If we simply remove the line that returns the decompressor in LzopInputStream.close(), it would satisfy the use cases mentioned here (via LineRecordReader, or any use case that returns the decompressor). However, I'm pretty certain there are a lot of use cases (correct or not) that are not returning the decompressor, implicitly relying on LzopInputStream.close(). For them, the decompressors would start leaking.

As you suggested, annotating the decompressor as DoNotPool is one option, but we would forgo the benefit of pooling. Decompressors carry native buffers, and not pooling them would have a fairly major performance implication depending on the use cases. So I'm not sure we want to go there.

BTW, it looks like the compressor (LzopOutputStream) has the same issue.

[sjlee]

Now that #103 was merged, would you like to clean up this unit test and update this PR? This would be a useful test.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}