kafka-4576: Log segments close to max size break on fetch by huxihx · Pull Request #2304 · apache/kafka

huxihx · 2017-01-04T03:39:38Z

JVM spec does not make a guarantee that the buffer will be filled up when invoking FileChannel.read, which is called in searchOffsetWithSize to fail the follower replicas' reading data from leader when large log segment is set.

huxihx · 2017-01-04T03:40:13Z

@ijuma @onurkaraman Please have a review on this PR. Thanks

asfbot · 2017-01-04T03:41:19Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/469/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-04T03:41:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/470/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-04T03:41:34Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/468/
Test FAILed (JDK 7 and Scala 2.10).

ijuma

Thanks for the PR. In addition to the other review comments:

It would be good to add unit tests for the new method
FileRecords.readInto also has to be updated.

cc @junrao @hachikuji who may have thoughts on this as well.

ijuma · 2017-01-04T10:04:06Z

+     * @throws IllegalArgumentException If invalid arguments were passed
+     * @throws IOException If some other I/O error occurs
+     */
+    public static void readFullyFrom(FileChannel channel, ByteBuffer buffer, long startPosition) throws IOException {


Maybe the name should simply be readFully. And buffer should maybe be destinationBuffer. And startPosition can maybe just be position for consistency with FileChannel.read.

Finally, we need a mechanism to indicate that EOF has been reached. Maybe a boolean is good enough. The usual way of returning an int is a bit weird for the case where some bytes are read and then EOF is reached.

ijuma · 2017-01-04T10:06:50Z

+        }
+        if (buffer == null) {
+            throw new IllegalArgumentException("Must specify the byte buffer for the channel.");
+        }


I don't think we need the null checks.

ijuma · 2017-01-04T10:07:12Z

+            throw new IllegalArgumentException("Must specify the byte buffer for the channel.");
+        }
+        if (startPosition < 0) {
+            throw new IllegalArgumentException("The file position cannot be negative.");


We should include startPosition in the message.

ijuma · 2017-01-04T10:17:36Z

+        if (startPosition < 0) {
+            throw new IllegalArgumentException("The file position cannot be negative.");
+        }
+        final int totalBytesToRead = buffer.capacity();


This is not correct, it should be buffer.remaining. It seems simpler to just do while (buffer.remaining()) with an early exit in the case read returns -1.

ijuma · 2017-01-04T10:17:53Z

+            int bytesRead = channel.read(buffer, currentPosition);
+            if (bytesRead == -1) {
+                log.warn(String.format("Read incomplete data from the given file channel. Current file channel size: %d, and wanted to read at position: %d",
+                        channel.size(), currentPosition));


What's the purpose of this warning? It doesn't seem needed.

ijuma · 2017-01-04T10:20:22Z

+            totalBytesRead += bytesRead;
+            currentPosition += bytesRead;
+        }
+        buffer.flip(); // Make the buffer ready for read


A bit unclear why we need this. In my mind, readFully should be as close to FileChannel.read as possible with the exception that it attempts to fill the buffer while end of file is not reached.

ijuma · 2017-01-04T10:24:00Z

Note that there are a number of test failures caused by this change.

huxihx · 2017-01-04T14:44:27Z

Thanks for the comments. Will work on these comments asap.

Addressed Ijuma's comments 1. Refined some variaiable names 2. Fix code bug to have all unit test cases passed

huxihx · 2017-01-05T06:14:15Z

@ijuma Thanks for the comments. Refined the code as per your suggestions. Now all the unit test cases should pass. Since we already have about 7 unit test cases (such as FileRecordsTest.testRead), do you think we still need to add new test cases for this method? Thanks again.

asfbot · 2017-01-05T06:53:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/513/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-05T06:55:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/512/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-05T07:00:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/511/
Test PASSed (JDK 7 and Scala 2.10).

ijuma

Thanks for the updates. Looks good overall, just a few nits.

ijuma · 2017-01-05T10:27:22Z

+    public static void readFully(FileChannel channel, ByteBuffer destinationBuffer, long position) throws IOException {
+        if (channel == null) {
+            throw new IllegalArgumentException("Cannot read an empty file channel.");
+        }


I don't think we need this null check either.

ijuma · 2017-01-05T10:28:33Z

    }

+    /**
+     * Read data from the channel to the given byte buffer


Maybe add at the end until there are no bytes remaining in the buffer or the channel has reached end of stream.

ijuma · 2017-01-05T10:31:14Z

+     * @param position The file position at which the transfer is to begin; must be non-negative
+     *
+     * @throws IllegalArgumentException If invalid arguments were passed
+     * @throws IOException If some other I/O error occurs


@throws IOException If an I/O error occurs. See {@link FileChannel#read(ByteBuffer, long)} for detailed information on the exception cases.

ijuma · 2017-01-05T10:41:33Z

+            throw new IllegalArgumentException("Cannot read an empty file channel.");
+        }
+        if (position < 0) {
+            throw new IllegalArgumentException("The file position cannot be negative, but we got " + position);


Nit: "The file channel position..."

ijuma · 2017-01-05T10:44:42Z

@@ -113,7 +113,7 @@ public FileChannel channel() {
     * @throws IOException
     */
    public ByteBuffer readInto(ByteBuffer buffer, int position) throws IOException {


Worth clarifying the docs, similar to the readFully documentation.

ijuma · 2017-01-05T10:47:08Z

+            throw new IllegalArgumentException("The file position cannot be negative, but we got " + position);
+        }
+        long currentPosition = position;
+        while (destinationBuffer.hasRemaining()) {


I think we probably want a do/while loop here. There should be no difference in behaviour, but it seems to model the problem better (i.e. we first do a read and then we check if there is still space remaining in the buffer. Maybe:

long currentPosition = position; int bytesRead; do { bytesRead = channel.read(destinationBuffer, currentPosition); currentPosition += bytesRead; } while (bytesRead != -1 && destinationBuffer.hasRemaining());

…ystic/kafka into kafka4576_FileChannel_read

adjusted some comments as per Ijuma's suggestion.

…nel_read

huxihx · 2017-01-05T12:09:27Z

@ijuma already addressed your comments, please take some time to review again. Appreciated.

ijuma · 2017-01-05T12:23:14Z

-     * @return The same buffer
-     * @throws IOException
+     * Read data from the channel to the given byte buffer until there are no bytes remaining in the buffer or
+     * the channel has reached end of stream


Hmm, the idea was to extend the existing definition to explain in more detail what happens, not to change to be the same as readFully.

sorry, but I do not catch your meaning. Do you mean we should modify the comments for readInto method to explain why we use readFully instead of naive Channel.read?

The previous comment said "Read log entries into the given buffer". This is more specific than "Read data from the channel to the given byte buffer..." and we should keep that. We could perhaps say:

"Read log entries into the given buffer until there are no bytes remaining in the buffer or the channel has reached end of stream."

Thank you for clarifying this. Already addressed. Please check.

asfbot · 2017-01-05T12:47:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/528/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-05T12:49:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/526/
Test PASSed (JDK 7 and Scala 2.10).

Restored the comments for FileRecords.readInto method and tweaked some expression.

ijuma

LGTM. Will wait for @hachikuji or @junrao to take a look as well before merging.

asfbot · 2017-01-05T13:39:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/530/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-05T13:41:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/529/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-05T13:46:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/528/
Test PASSed (JDK 7 and Scala 2.10).

hachikuji · 2017-01-05T17:44:00Z

The fix LGTM. Would be nice to have some test cases for readFully. Maybe that wouldn't be too bad using a mock for FileChannel?

ijuma · 2017-01-05T17:49:58Z

Good point about the test, I had asked for it and then forgot to reply to @amethystic's question. I agree with @hachikuji that it would be useful to have a simple test with a mocked FileChannel.

huxihx · 2017-01-06T03:04:49Z

@ijuma sorry but not familiar with the mock things. Could you give an example on how to get started with that?

ijuma · 2017-01-06T11:36:04Z

@amethystic, if you search for EasyMock in our tests, you'll find many examples. The idea is to mock the FileChannel.read method to verify that readFully works correctly for the various cases: end of stream is reached before buffer is full, buffer is filled with a single read, buffer is filled after multiple reads, etc.

The existing unit tests don't exercise these different paths (since they already passed before your change).

huxihx · 2017-01-20T09:15:36Z

What readFully must guarantee is that it never leads to a situation where channel still has something to read but buffer does not get filled up. As a utility method, readFully should not assume channel's EOF is an exceptional case. The enclosing method could take over that responsibility.

junrao · 2017-01-20T21:11:07Z

@amethystic : I think this issue is in FileRecords.readInto(). In this method, the expectation is that the buffer may not be filled if there is not enough bytes in the channel. So, for that, we probably need a slightly different util function that reads until the min(Buffer capacity and the remaining bytes in channel). In other places, we do expect the buffer will be filled up.

ijuma · 2017-01-21T09:50:56Z

+            currentPosition += bytesRead;
+        } while (bytesRead != -1 && destinationBuffer.hasRemaining());
+        if (bytesRead == -1) {
+            throw new IOException(String.format("Reached EOF before filling up the buffer. Current file channel position = %d, buffer remaining = %d",


It would be nice if the calling code could pass a string with context for what is being read. That way we don't lose information like:

if (buf.hasRemaining()) - throw new KafkaException("Failed to read magic byte from FileChannel " + channel);

For the case above, we could pass "magic byte" and then the error message would read: "Reached EOF before filling the buffer with magic byte..." or something along those lines.

Thanks for the feedback. Personally, I agree @junrao 's suggestion that we might need another read function to take care of the cases when reading EOF is expected. As for @ijuma 's "magic byte", I am not sure if I fully understand how it could be used, but I still think we have put too much responsibilities to readFully which otherwise should do only one thing: READING CHANNEL TO THE BUFFER. So I strongly recommend we refine the enclosing readInto to have all of these things pass the tests.

What do you think?

I am not sure what is your concrete suggestion. Our aim here is to provide internal utility methods that do the right thing concisely and with good error messages.

I agree that it may make sense to have two methods. One could be readFullyOrFail and the other could simply be readFully maybe. The former would throw the exception on EOF while the other would not.

To clarify my suggestion, I meant that the readFullyOrFail variant could take a string parameter that would be included in the IOException message. It's useful to know that we failed to read "magic string", for example.

Okay, I will consider to combine all the suggestion and submit a fix asap. Thank for the comments.

Addressed Jun's comments 1. Renamed readFully to readFullyOrFail 2. Added a new method 'readFully' that do not throw IOException when encountering EOF 3. Added two unit test cases to test readFully 3. Added two unit test cases to test readFully 3. Added two unit test cases to test readFully

asfbot · 2017-01-22T05:12:06Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1092/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-22T05:12:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1090/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-22T05:17:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1090/
Test PASSed (JDK 7 and Scala 2.10).

ijuma · 2017-01-22T11:02:03Z

-        channel.read(logHeaderBuffer, position);
-        if (logHeaderBuffer.hasRemaining())
-            return null;
+        Utils.readFullyOrFail(channel, logHeaderBuffer, position);


@junrao, is it OK to throw an exception instead of returning null if we reach EOF before filling the buffer here?

ijuma

Thanks for the updates and for the tests. I submitted a PR to your branch with some improvements here:

huxihx#1

I also left a few comments regarding the tests and one question for Jun regarding the implementation.

ijuma · 2017-01-22T14:54:13Z

+            // Scenario 1: test single read
+            Utils.readFullyOrFail(channel, perfectBuffer, 0);
+            assertTrue("Buffer should be filled up.", !perfectBuffer.hasRemaining());
+            // Scenario 2: test multiple reads


I am not sure what you mean by test multiple reads here. I think we're testing two things here:

That the first read doesn't cause a side-effect to the channel that could prevent the second read from succeeding.

That we can read into a smaller buffer than what's in the file channel.

"multiple reads" is a little unclear as it sounds like we are causing the underlying file channel to do multiple reads, which I don't think we are.

ijuma · 2017-01-22T14:55:31Z

+    }
+
+    @Test
+    public void testReadFullyOrFailThrowIOExceptionForEOF() throws IOException {


Does this add anything over testReadFullyOrFailWithRealFile? If not, we can simply remove it.

ijuma · 2017-01-22T14:58:25Z

+            assertTrue("Buffer should be filled up.", !perfectBuffer.hasRemaining());
+            // Scenario 2: test multiple reads
+            Utils.readFullyOrFail(channel, smallBuffer, 0);
+            assertTrue("Buffer should be filled up.", !smallBuffer.hasRemaining());


Maybe we should add one case where position > 0.

ijuma · 2017-01-22T15:00:07Z

+
+        EasyMock.replay(channelMock);
+        Utils.readFully(channelMock, buffer, 0L);
+        Assert.assertTrue("The buffer should be filled up", !buffer.hasRemaining());


Seems like assertFalse would be more appropriate here. There are a few cases like that. Also, it would be good to verify the buffer contents.

ijuma · 2017-01-22T15:06:21Z

    }

+    @Test
+    public void testReadFullyOrFailWithRealFile() throws IOException {


It would be good to verify that the the buffer is populated correctly.

ijuma · 2017-01-22T15:09:54Z

+
+        EasyMock.replay(channelMock);
+        Utils.readFullyOrFail(channelMock, buffer, 0L);
+        Assert.assertTrue("The buffer should be filled up", !buffer.hasRemaining());


It would be good to verify that the buffer contents are correct as well.

ijuma · 2017-01-22T15:10:51Z

+                }
+            });
+            totalMockedBytesRead += mockedBytesRead;
+        }


Seems like this is the same in testReadFullyOrFailWithMultiReads. Maybe we can extract it to a helper method.

ijuma · 2017-01-22T15:12:23Z

+    }
+
+    @Test
+    public void testReadFullyCanFillUpBuffer() throws IOException {


Is this name correct? It seems like the buffer is never filled in this test.

Implementation and documentation improvements

asfbot · 2017-01-23T02:08:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1099/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-23T02:17:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1097/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-23T03:27:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1097/
Test FAILed (JDK 8 and Scala 2.12).

Merged Ijuma's fix and addressed Ijuma's comments.

huxihx · 2017-01-23T03:48:25Z

@ijuma already merged your fix and addressed the comments. Thanks for the feedback and review again:)

asfbot · 2017-01-23T04:36:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1102/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-23T04:38:36Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1100/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-23T06:02:01Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1100/
Test FAILed (JDK 7 and Scala 2.10).

junrao

@amethystic : Thanks for the patch. LGTM. Just a couple of minor comments.

junrao · 2017-01-24T00:34:31Z

+                break;
+            }
+            // If the remaining byte number is less than a step,
+            // directly assign the last mocked bytes read to fix `bufferSize` exactly


fix => fit?

junrao · 2017-01-24T00:34:36Z

+     * @param channel File channel containing the data to read from
+     * @param destinationBuffer The buffer into which bytes are to be transferred
+     * @param position The file position at which the transfer is to begin; it must be non-negative
+     * @param description A description of what is being read, this will be included in the EOFException if we


Incomplete sentence after "if we"?

huxihx · 2017-01-24T03:22:20Z

Sorry for the typos. Thanks.

Fixed some typos

asfbot · 2017-01-24T03:46:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1136/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-24T04:32:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1134/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-24T06:39:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1134/
Test FAILed (JDK 7 and Scala 2.10).

`FileChannel.read` may not fill the destination buffer even if there are enough bytes in the channel to do so. Add a couple of utility methods that ensure this and use them from all the relevant places. Author: huxi <huxi@zhenrongbao.com> Author: amethystic <huxi_2b@hotmail.com> Author: Ismael Juma <ismael@juma.me.uk> Reviewers: Jun Rao <junrao@gmail.com>, Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes #2304 from amethystic/kafka4576_FileChannel_read (cherry picked from commit 337f576) Signed-off-by: Ismael Juma <ismael@juma.me.uk>

ijuma · 2017-01-24T11:03:26Z

Thanks for the updates, I merged to trunk and 0.10.2 after tweaking the tests a little.

`FileChannel.read` may not fill the destination buffer even if there are enough bytes in the channel to do so. Add a couple of utility methods that ensure this and use them from all the relevant places. Author: huxi <huxi@zhenrongbao.com> Author: amethystic <huxi_2b@hotmail.com> Author: Ismael Juma <ismael@juma.me.uk> Reviewers: Jun Rao <junrao@gmail.com>, Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes apache#2304 from amethystic/kafka4576_FileChannel_read

kafka-4576: Log segments close to max size break on fetch

f32fe29

JVM spec does not make a guarantee that the buffer will be filled up when invoking FileChannel.read, which is called in searchOffsetWithSize to fail the follower replicas' reading data from leader when large log segment is set.

ijuma requested changes Jan 4, 2017

View reviewed changes

kafka-4576: Log segments close to max size break on fetch

ed9bbe0

Addressed Ijuma's comments 1. Refined some variaiable names 2. Fix code bug to have all unit test cases passed

ijuma reviewed Jan 5, 2017

View reviewed changes

amethystic added 3 commits January 5, 2017 19:45

Merge branch 'kafka4576_FileChannel_read' of https://github.com/ameth…

88357de

…ystic/kafka into kafka4576_FileChannel_read

kafka-4576: Log segments close to max size break on fetch

4877475

adjusted some comments as per Ijuma's suggestion.

Merge remote-tracking branch 'upstream/trunk' into kafka4576_FileChan…

12ba2d9

…nel_read

ijuma reviewed Jan 5, 2017

View reviewed changes

kafka-4576: Log segments close to max size break on fetch

f9b9dd5

Restored the comments for FileRecords.readInto method and tweaked some expression.

ijuma approved these changes Jan 5, 2017

View reviewed changes

ijuma reviewed Jan 21, 2017

View reviewed changes

ijuma reviewed Jan 22, 2017

View reviewed changes

Implementation and documentation improvements

beb77ec

ijuma reviewed Jan 22, 2017

View reviewed changes

Merge pull request #1 from ijuma/kafka-4576-read-fully

e6543e1

Implementation and documentation improvements

kafka-4576: Log segments close to max size break on fetch

f622e79

Merged Ijuma's fix and addressed Ijuma's comments.

junrao reviewed Jan 24, 2017

View reviewed changes

kafka-4576: Log segments close to max size break on fetch

d5c8ae9

Fixed some typos

asfgit closed this in 337f576 Jan 24, 2017

huxihx deleted the kafka4576_FileChannel_read branch January 24, 2017 11:35

Conversation

huxihx commented Jan 4, 2017

Uh oh!

huxihx commented Jan 4, 2017

Uh oh!

asfbot commented Jan 4, 2017

Uh oh!

asfbot commented Jan 4, 2017

Uh oh!

asfbot commented Jan 4, 2017

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Jan 4, 2017

Uh oh!

huxihx commented Jan 4, 2017

Uh oh!

huxihx commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

huxihx commented Jan 5, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

ijuma left a comment

Choose a reason for hiding this comment

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

asfbot commented Jan 5, 2017

Uh oh!

hachikuji commented Jan 5, 2017

Uh oh!

ijuma commented Jan 5, 2017

Uh oh!