@Test public void writeFromTest() throws IOException, StoreException { String currSegmentName = "log_current"; LogSegment segment = getSegment(currSegmentName, STANDARD_SEGMENT_SIZE, true); try { long writeStartOffset = segment.getStartOffset(); byte[] bufOne = TestUtils.getRandomBytes(STANDARD_SEGMENT_SIZE / 3); byte[] bufTwo = TestUtils.getRandomBytes(STANDARD_SEGMENT_SIZE / 2); segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(bufOne))), writeStartOffset, bufOne.length); assertEquals("End offset is not as expected", writeStartOffset + bufOne.length, segment.getEndOffset()); readAndEnsureMatch(segment, writeStartOffset, bufOne); segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(bufTwo))), writeStartOffset, bufTwo.length); assertEquals("End offset is not as expected", writeStartOffset + bufTwo.length, segment.getEndOffset()); readAndEnsureMatch(segment, writeStartOffset, bufTwo); segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(bufOne))), writeStartOffset, bufOne.length); assertEquals("End offset is not as expected", writeStartOffset + bufTwo.length, segment.getEndOffset()); readAndEnsureMatch(segment, writeStartOffset, bufOne); readAndEnsureMatch(segment, writeStartOffset + bufOne.length, Arrays.copyOfRange(bufTwo, bufOne.length, bufTwo.length)); for (int i = 0; i < 10; i++) { long offset = writeStartOffset + Utils.getRandomLong(TestUtils.RANDOM, segment.sizeInBytes() - bufOne.length - writeStartOffset); segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(bufOne))), offset, bufOne.length); readAndEnsureMatch(segment, offset, bufOne); } ByteBuffer failBuf = ByteBuffer.wrap(TestUtils.getRandomBytes((int) (STANDARD_SEGMENT_SIZE - writeStartOffset + 1))); long writeOverFlowCount = metrics.overflowWriteError.getCount(); try { segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(failBuf)), writeStartOffset, failBuf.remaining()); fail("WriteFrom should have failed because data won't fit"); } catch (IndexOutOfBoundsException e) { assertEquals("Write overflow should have been reported", writeOverFlowCount + 1, metrics.overflowWriteError.getCount()); assertEquals("Position of buffer has changed", 0, failBuf.position()); } long[] invalidOffsets = {writeStartOffset - 1, STANDARD_SEGMENT_SIZE, STANDARD_SEGMENT_SIZE + 1}; ByteBuffer buffer = ByteBuffer.wrap(TestUtils.getRandomBytes(1)); for (long invalidOffset : invalidOffsets) { try { segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(buffer)), invalidOffset, buffer.remaining()); fail("WriteFrom should have failed because offset provided for write is invalid"); } catch (IndexOutOfBoundsException e) { assertEquals("Position of buffer has changed", 0, buffer.position()); } } segment.close(false); try { segment.writeFrom(Channels.newChannel(new ByteBufferInputStream(buffer)), writeStartOffset, buffer.remaining()); fail("WriteFrom should have failed because segments are closed"); } catch (ClosedChannelException e) { assertEquals("Position of buffer has changed", 0, buffer.position()); } } finally { closeSegmentAndDeleteFile(segment); } }
void writeFrom(ReadableByteChannel channel, long offset, long size) throws IOException { if (offset < startOffset || offset >= capacityInBytes) { throw new IndexOutOfBoundsException( "Provided offset [" + offset + "] is out of bounds for the segment [" + file.getAbsolutePath() + "] with capacity [" + capacityInBytes + "]"); } if (offset + size > capacityInBytes) { metrics.overflowWriteError.inc(); throw new IndexOutOfBoundsException( "Cannot write to segment [" + file.getAbsolutePath() + "] from offset [" + offset + "] for size [" + size + "] because it exceeds the capacity [" + capacityInBytes + "]"); } long bytesWritten = 0; while (bytesWritten < size) { bytesWritten += fileChannel.transferFrom(channel, offset + bytesWritten, size - bytesWritten); } if (offset + size > endOffset.get()) { endOffset.set(offset + size); } }
LogSegment implements Read, Write { void writeFrom(ReadableByteChannel channel, long offset, long size) throws IOException { if (offset < startOffset || offset >= capacityInBytes) { throw new IndexOutOfBoundsException( "Provided offset [" + offset + "] is out of bounds for the segment [" + file.getAbsolutePath() + "] with capacity [" + capacityInBytes + "]"); } if (offset + size > capacityInBytes) { metrics.overflowWriteError.inc(); throw new IndexOutOfBoundsException( "Cannot write to segment [" + file.getAbsolutePath() + "] from offset [" + offset + "] for size [" + size + "] because it exceeds the capacity [" + capacityInBytes + "]"); } long bytesWritten = 0; while (bytesWritten < size) { bytesWritten += fileChannel.transferFrom(channel, offset + bytesWritten, size - bytesWritten); } if (offset + size > endOffset.get()) { endOffset.set(offset + size); } } }
LogSegment implements Read, Write { void writeFrom(ReadableByteChannel channel, long offset, long size) throws IOException { if (offset < startOffset || offset >= capacityInBytes) { throw new IndexOutOfBoundsException( "Provided offset [" + offset + "] is out of bounds for the segment [" + file.getAbsolutePath() + "] with capacity [" + capacityInBytes + "]"); } if (offset + size > capacityInBytes) { metrics.overflowWriteError.inc(); throw new IndexOutOfBoundsException( "Cannot write to segment [" + file.getAbsolutePath() + "] from offset [" + offset + "] for size [" + size + "] because it exceeds the capacity [" + capacityInBytes + "]"); } long bytesWritten = 0; while (bytesWritten < size) { bytesWritten += fileChannel.transferFrom(channel, offset + bytesWritten, size - bytesWritten); } if (offset + size > endOffset.get()) { endOffset.set(offset + size); } } LogSegment(String name, File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics,
      boolean writeHeader);   LogSegment(String name, File file, StoreConfig config, StoreMetrics metrics);   LogSegment(File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics, FileChannel fileChannel); }
LogSegment implements Read, Write { void writeFrom(ReadableByteChannel channel, long offset, long size) throws IOException { if (offset < startOffset || offset >= capacityInBytes) { throw new IndexOutOfBoundsException( "Provided offset [" + offset + "] is out of bounds for the segment [" + file.getAbsolutePath() + "] with capacity [" + capacityInBytes + "]"); } if (offset + size > capacityInBytes) { metrics.overflowWriteError.inc(); throw new IndexOutOfBoundsException( "Cannot write to segment [" + file.getAbsolutePath() + "] from offset [" + offset + "] for size [" + size + "] because it exceeds the capacity [" + capacityInBytes + "]"); } long bytesWritten = 0; while (bytesWritten < size) { bytesWritten += fileChannel.transferFrom(channel, offset + bytesWritten, size - bytesWritten); } if (offset + size > endOffset.get()) { endOffset.set(offset + size); } } LogSegment(String name, File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics,
      boolean writeHeader);   LogSegment(String name, File file, StoreConfig config, StoreMetrics metrics);   LogSegment(File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics, FileChannel fileChannel); @Override int appendFrom(ByteBuffer buffer); @Override void appendFrom(ReadableByteChannel channel, long size); @Override void readInto(ByteBuffer buffer, long position); @Override String toString(); }
LogSegment implements Read, Write { void writeFrom(ReadableByteChannel channel, long offset, long size) throws IOException { if (offset < startOffset || offset >= capacityInBytes) { throw new IndexOutOfBoundsException( "Provided offset [" + offset + "] is out of bounds for the segment [" + file.getAbsolutePath() + "] with capacity [" + capacityInBytes + "]"); } if (offset + size > capacityInBytes) { metrics.overflowWriteError.inc(); throw new IndexOutOfBoundsException( "Cannot write to segment [" + file.getAbsolutePath() + "] from offset [" + offset + "] for size [" + size + "] because it exceeds the capacity [" + capacityInBytes + "]"); } long bytesWritten = 0; while (bytesWritten < size) { bytesWritten += fileChannel.transferFrom(channel, offset + bytesWritten, size - bytesWritten); } if (offset + size > endOffset.get()) { endOffset.set(offset + size); } } LogSegment(String name, File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics,
      boolean writeHeader);   LogSegment(String name, File file, StoreConfig config, StoreMetrics metrics);   LogSegment(File file, long capacityInBytes, StoreConfig config, StoreMetrics metrics, FileChannel fileChannel); @Override int appendFrom(ByteBuffer buffer); @Override void appendFrom(ReadableByteChannel channel, long size); @Override void readInto(ByteBuffer buffer, long position); @Override String toString();  }