Repository: kafka Updated Branches: refs/heads/0.10.2 4218c0247 -> 928f69ce8
KAFKA-4576; Log segments close to max size break on fetch `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 <[email protected]> Author: amethystic <[email protected]> Author: Ismael Juma <[email protected]> Reviewers: Jun Rao <[email protected]>, Jason Gustafson <[email protected]>, Ismael Juma <[email protected]> Closes #2304 from amethystic/kafka4576_FileChannel_read (cherry picked from commit 337f576f5979bf8924c5707b338cf4d3c76a53fd) Signed-off-by: Ismael Juma <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/kafka/repo Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/928f69ce Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/928f69ce Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/928f69ce Branch: refs/heads/0.10.2 Commit: 928f69ce883c8e30050d2e6f5e72e77e6740bdb9 Parents: 4218c02 Author: huxi <[email protected]> Authored: Tue Jan 24 10:04:42 2017 +0000 Committer: Ismael Juma <[email protected]> Committed: Tue Jan 24 11:02:37 2017 +0000 ---------------------------------------------------------------------- .../kafka/common/record/FileLogInputStream.java | 13 +- .../apache/kafka/common/record/FileRecords.java | 9 +- .../org/apache/kafka/common/utils/Utils.java | 52 +++++++ .../apache/kafka/common/utils/UtilsTest.java | 140 ++++++++++++++++++- 4 files changed, 201 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/kafka/blob/928f69ce/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java b/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java index ae393b0..dd9cc84 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java +++ b/clients/src/main/java/org/apache/kafka/common/record/FileLogInputStream.java @@ -18,6 +18,7 @@ package org.apache.kafka.common.record; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.errors.CorruptRecordException; +import org.apache.kafka.common.utils.Utils; import java.io.IOException; import java.nio.ByteBuffer; @@ -56,9 +57,7 @@ public class FileLogInputStream implements LogInputStream<FileLogInputStream.Fil return null; logHeaderBuffer.rewind(); - channel.read(logHeaderBuffer, position); - if (logHeaderBuffer.hasRemaining()) - return null; + Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(); @@ -117,9 +116,7 @@ public class FileLogInputStream implements LogInputStream<FileLogInputStream.Fil try { byte[] magic = new byte[1]; ByteBuffer buf = ByteBuffer.wrap(magic); - channel.read(buf, position + Records.LOG_OVERHEAD + Record.MAGIC_OFFSET); - if (buf.hasRemaining()) - throw new KafkaException("Failed to read magic byte from FileChannel " + channel); + Utils.readFullyOrFail(channel, buf, position + Records.LOG_OVERHEAD + Record.MAGIC_OFFSET, "magic byte"); return magic[0]; } catch (IOException e) { throw new KafkaException(e); @@ -136,9 +133,7 @@ public class FileLogInputStream implements LogInputStream<FileLogInputStream.Fil return record; ByteBuffer recordBuffer = ByteBuffer.allocate(recordSize); - channel.read(recordBuffer, position + Records.LOG_OVERHEAD); - if (recordBuffer.hasRemaining()) - throw new IOException("Failed to read full record from channel " + channel); + Utils.readFullyOrFail(channel, recordBuffer, position + Records.LOG_OVERHEAD, "full record"); recordBuffer.rewind(); record = new Record(recordBuffer); http://git-wip-us.apache.org/repos/asf/kafka/blob/928f69ce/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java index 8a33dca..960b716 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java +++ b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java @@ -108,14 +108,17 @@ public class FileRecords extends AbstractRecords implements Closeable { } /** - * Read log entries into a given buffer. + * Read log entries into the given buffer until there are no bytes remaining in the buffer or the end of the file + * is reached. + * * @param buffer The buffer to write the entries to * @param position Position in the buffer to read from * @return The same buffer - * @throws IOException + * @throws IOException If an I/O error occurs, see {@link FileChannel#read(ByteBuffer, long)} for details on the + * possible exceptions */ public ByteBuffer readInto(ByteBuffer buffer, int position) throws IOException { - channel.read(buffer, position + this.start); + Utils.readFully(channel, buffer, position + this.start); buffer.flip(); return buffer; } http://git-wip-us.apache.org/repos/asf/kafka/blob/928f69ce/clients/src/main/java/org/apache/kafka/common/utils/Utils.java ---------------------------------------------------------------------- diff --git a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java index ac8d078..afa85bd 100755 --- a/clients/src/main/java/org/apache/kafka/common/utils/Utils.java +++ b/clients/src/main/java/org/apache/kafka/common/utils/Utils.java @@ -12,6 +12,7 @@ */ package org.apache.kafka.common.utils; +import java.io.EOFException; import java.io.IOException; import java.io.InputStream; import java.io.Closeable; @@ -803,4 +804,55 @@ public class Utils { return Crc32.crc32(buffer.array(), buffer.arrayOffset() + start, size); } + /** + * Read data from the channel to the given byte buffer until there are no bytes remaining in the buffer. If the end + * of the file is reached while there are bytes remaining in the buffer, an EOFException is thrown. + * + * @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 it is thrown + * + * @throws IllegalArgumentException If position is negative + * @throws EOFException If the end of the file is reached while there are remaining bytes in the destination buffer + * @throws IOException If an I/O error occurs, see {@link FileChannel#read(ByteBuffer, long)} for details on the + * possible exceptions + */ + public static void readFullyOrFail(FileChannel channel, ByteBuffer destinationBuffer, long position, + String description) throws IOException { + if (position < 0) { + throw new IllegalArgumentException("The file channel position cannot be negative, but it is " + position); + } + int expectedReadBytes = destinationBuffer.remaining(); + readFully(channel, destinationBuffer, position); + if (destinationBuffer.hasRemaining()) { + throw new EOFException(String.format("Failed to read `%s` from file channel `%s`. Expected to read %d bytes, " + + "but reached end of file after reading %d bytes. Started read from position %d.", + description, channel, expectedReadBytes, expectedReadBytes - destinationBuffer.remaining(), position)); + } + } + + /** + * Read data from the channel to the given byte buffer until there are no bytes remaining in the buffer or the end + * of the file has been reached. + * + * @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 + * + * @throws IllegalArgumentException If position is negative + * @throws IOException If an I/O error occurs, see {@link FileChannel#read(ByteBuffer, long)} for details on the + * possible exceptions + */ + public static void readFully(FileChannel channel, ByteBuffer destinationBuffer, long position) throws IOException { + if (position < 0) { + throw new IllegalArgumentException("The file channel position cannot be negative, but it is " + position); + } + long currentPosition = position; + int bytesRead; + do { + bytesRead = channel.read(destinationBuffer, currentPosition); + currentPosition += bytesRead; + } while (bytesRead != -1 && destinationBuffer.hasRemaining()); + } } http://git-wip-us.apache.org/repos/asf/kafka/blob/928f69ce/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java ---------------------------------------------------------------------- diff --git a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java index 46400b4..194cad6 100755 --- a/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java +++ b/clients/src/test/java/org/apache/kafka/common/utils/UtilsTest.java @@ -16,18 +16,27 @@ */ package org.apache.kafka.common.utils; +import java.io.EOFException; +import java.nio.channels.FileChannel; +import java.nio.file.StandardOpenOption; import java.util.Arrays; import java.util.Collections; import java.io.Closeable; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Random; +import org.apache.kafka.test.TestUtils; +import org.easymock.EasyMock; +import org.easymock.IAnswer; import org.junit.Test; + +import static org.apache.kafka.common.utils.Utils.formatAddress; import static org.apache.kafka.common.utils.Utils.getHost; import static org.apache.kafka.common.utils.Utils.getPort; -import static org.apache.kafka.common.utils.Utils.formatAddress; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -158,6 +167,135 @@ public class UtilsTest { } } + @Test + public void testReadFullyOrFailWithRealFile() throws IOException { + try (FileChannel channel = FileChannel.open(TestUtils.tempFile().toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE)) { + // prepare channel + String msg = "hello, world"; + channel.write(ByteBuffer.wrap(msg.getBytes()), 0); + channel.force(true); + assertEquals("Message should be written to the file channel", channel.size(), msg.length()); + + ByteBuffer perfectBuffer = ByteBuffer.allocate(msg.length()); + ByteBuffer smallBuffer = ByteBuffer.allocate(5); + ByteBuffer largeBuffer = ByteBuffer.allocate(msg.length() + 1); + // Scenario 1: test reading into a perfectly-sized buffer + Utils.readFullyOrFail(channel, perfectBuffer, 0, "perfect"); + assertFalse("Buffer should be filled up", perfectBuffer.hasRemaining()); + assertEquals("Buffer should be populated correctly", msg, new String(perfectBuffer.array())); + // Scenario 2: test reading into a smaller buffer + Utils.readFullyOrFail(channel, smallBuffer, 0, "small"); + assertFalse("Buffer should be filled", smallBuffer.hasRemaining()); + assertEquals("Buffer should be populated correctly", "hello", new String(smallBuffer.array())); + // Scenario 3: test reading starting from a non-zero position + smallBuffer.clear(); + Utils.readFullyOrFail(channel, smallBuffer, 7, "small"); + assertFalse("Buffer should be filled", smallBuffer.hasRemaining()); + assertEquals("Buffer should be populated correctly", "world", new String(smallBuffer.array())); + // Scenario 4: test end of stream is reached before buffer is filled up + try { + Utils.readFullyOrFail(channel, largeBuffer, 0, "large"); + fail("Expected EOFException to be raised"); + } catch (EOFException e) { + // expected + } + } + } + + /** + * Tests that `readFullyOrFail` behaves correctly if multiple `FileChannel.read` operations are required to fill + * the destination buffer. + */ + @Test + public void testReadFullyOrFailWithPartialFileChannelReads() throws IOException { + FileChannel channelMock = EasyMock.createMock(FileChannel.class); + final int bufferSize = 100; + ByteBuffer buffer = ByteBuffer.allocate(bufferSize); + StringBuilder expectedBufferContent = new StringBuilder(); + fileChannelMockExpectReadWithRandomBytes(channelMock, expectedBufferContent, bufferSize); + EasyMock.replay(channelMock); + Utils.readFullyOrFail(channelMock, buffer, 0L, "test"); + assertEquals("The buffer should be populated correctly", expectedBufferContent.toString(), + new String(buffer.array())); + assertFalse("The buffer should be filled", buffer.hasRemaining()); + EasyMock.verify(channelMock); + } + + /** + * Tests that `readFullyOrFail` behaves correctly if multiple `FileChannel.read` operations are required to fill + * the destination buffer. + */ + @Test + public void testReadFullyWithPartialFileChannelReads() throws IOException { + FileChannel channelMock = EasyMock.createMock(FileChannel.class); + final int bufferSize = 100; + StringBuilder expectedBufferContent = new StringBuilder(); + fileChannelMockExpectReadWithRandomBytes(channelMock, expectedBufferContent, bufferSize); + EasyMock.replay(channelMock); + ByteBuffer buffer = ByteBuffer.allocate(bufferSize); + Utils.readFully(channelMock, buffer, 0L); + assertEquals("The buffer should be populated correctly.", expectedBufferContent.toString(), + new String(buffer.array())); + assertFalse("The buffer should be filled", buffer.hasRemaining()); + EasyMock.verify(channelMock); + } + + @Test + public void testReadFullyIfEofIsReached() throws IOException { + final FileChannel channelMock = EasyMock.createMock(FileChannel.class); + final int bufferSize = 100; + final String fileChannelContent = "abcdefghkl"; + ByteBuffer buffer = ByteBuffer.allocate(bufferSize); + EasyMock.expect(channelMock.size()).andReturn((long) fileChannelContent.length()); + EasyMock.expect(channelMock.read(EasyMock.anyObject(ByteBuffer.class), EasyMock.anyInt())).andAnswer(new IAnswer<Integer>() { + @Override + public Integer answer() throws Throwable { + ByteBuffer buffer = (ByteBuffer) EasyMock.getCurrentArguments()[0]; + buffer.put(fileChannelContent.getBytes()); + return -1; + } + }); + EasyMock.replay(channelMock); + Utils.readFully(channelMock, buffer, 0L); + assertEquals("abcdefghkl", new String(buffer.array(), 0, buffer.position())); + assertEquals(buffer.position(), channelMock.size()); + assertTrue(buffer.hasRemaining()); + EasyMock.verify(channelMock); + } + + /** + * Expectation setter for multiple reads where each one reads random bytes to the buffer. + * + * @param channelMock The mocked FileChannel object + * @param expectedBufferContent buffer that will be updated to contain the expected buffer content after each + * `FileChannel.read` invocation + * @param bufferSize The buffer size + * @throws IOException If an I/O error occurs + */ + private void fileChannelMockExpectReadWithRandomBytes(final FileChannel channelMock, + final StringBuilder expectedBufferContent, + final int bufferSize) throws IOException { + final int step = 20; + final Random random = new Random(); + int remainingBytes = bufferSize; + while (remainingBytes > 0) { + final int mockedBytesRead = remainingBytes < step ? remainingBytes : random.nextInt(step); + final StringBuilder sb = new StringBuilder(); + EasyMock.expect(channelMock.read(EasyMock.anyObject(ByteBuffer.class), EasyMock.anyInt())).andAnswer(new IAnswer<Integer>() { + @Override + public Integer answer() throws Throwable { + ByteBuffer buffer = (ByteBuffer) EasyMock.getCurrentArguments()[0]; + for (int i = 0; i < mockedBytesRead; i++) + sb.append("a"); + buffer.put(sb.toString().getBytes()); + expectedBufferContent.append(sb); + return mockedBytesRead; + } + }); + remainingBytes -= mockedBytesRead; + } + } + private static class TestCloseable implements Closeable { private final int id; private final IOException closeException;
