[BEAM-2790] Use byte[] instead of ByteBuffer to read from Hadoop FS

[iemejia]

Follow this checklist to help us incorporate your contribution quickly and easily:

• Make sure there is a JIRA issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a JIRA issue. Your pull request should address just this issue, without pulling in other changes.
• Each commit in the pull request should have a meaningful subject line and body.
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue.
• Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
• Run mvn clean verify to make sure basic checks pass. A more thorough check will be performed on your pull request automatically.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

---

[iemejia]

Since the contained InputStream on some of the Hadoop FileSystem implementations may not implement ByteBufferReadable we should take a conservative approach and read directly from the byte array.
R: @lukecwik

[coveralls]

Coverage decreased (-0.01%) to 69.971% when pulling 76f2d91 on iemejia:BEAM-2790-fix-s3-read into 9b175cc on apache:master.

[lukecwik]

dst make have a limit on how much to read and its start position may not be zero.
dst may not have array accessible because the bytes may be off heap.

You should use position and limit to write into the bytebuffer or you will not respect what the user has asked for.
You should only use array() if you checked hasArray() and correctly calculate the offsets/limits with position(), arrayOffset() and remaining. Otherwise you should use a fixed size buffer and copy the bytes twice.

[steveloughran]

I'd recommend looking at PositionedReadable.readFully(), which will handle partial reads of the dest buffer by reading in more data, etc. On a simple() read it could return any value, including 0.

[iemejia]

@lukecwik Not sure if I understand, if you look at TextSource, it defines a fixed size for the read buffer, so we have a limit. TextBasedReader already defines it to be 8192.

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java

Lines 89 to 90 in 5181e61

	private static final int READ_BUFFER_SIZE = 8192;
	private final ByteBuffer readBuffer = ByteBuffer.allocate(READ_BUFFER_SIZE);

And the method that iterates to read (by delegating to the HadoopSeekableByteChannel) is already moving the buffer start position to zero on each iteration so we cannot get off heap, no?

beam/sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextSource.java

Lines 228 to 240 in 5181e61

	private boolean tryToEnsureNumberOfBytesInBuffer(int minCapacity) throws IOException {
	// While we aren't at EOF or haven't fulfilled the minimum buffer capacity,
	// attempt to read more bytes.
	while (buffer.size() <= minCapacity && !eof) {
	eof = inChannel.read(readBuffer) == -1;
	readBuffer.flip();
	buffer = buffer.concat(ByteString.copyFrom(readBuffer));
	readBuffer.clear();
	}
	// Return true if we were able to honor the minimum buffer capacity request
	return buffer.size() >= minCapacity;
	}
	}

[lukecwik]

That is only true for TextSource as it is implemented today. We can't say that all user implemented sources will always use in memory byte buffers with a fixed size or that the user will consume and reset the ByteBuffer on each read (for example they may detect that not enough was read and ask for more).

[lukecwik]

Add a test which exercises various bytebuffer setups.

[echauchot]

@lukecwik @steveloughran, Ismaël is off for some weeks, I will continue this PR. For now I'm just starting to dig into the subject.

[echauchot]

I did the fix in the code to deal with the offsets/limits to have the same behavior than read(ByteBuffer). I'm adding the corner tests to HadoopFilesystem.

[iemejia]

I am closing this one because I suppose @echauchot will do a new one. See you in october.