@Test public void storeMessageReadSetTest() throws IOException, StoreException { int logCapacity = 2000; int segCapacity = 1000; Log log = new Log(tempDir.getAbsolutePath(), logCapacity, StoreTestUtils.DEFAULT_DISK_SPACE_ALLOCATOR, createStoreConfig(segCapacity, setFilePermissionEnabled), metrics); try { LogSegment firstSegment = log.getFirstSegment(); int availableSegCapacity = (int) (segCapacity - firstSegment.getStartOffset()); byte[] srcOfTruth = TestUtils.getRandomBytes(2 * availableSegCapacity); ReadableByteChannel dataChannel = Channels.newChannel(new ByteBufferInputStream(ByteBuffer.wrap(srcOfTruth))); log.appendFrom(dataChannel, availableSegCapacity); log.appendFrom(dataChannel, availableSegCapacity); LogSegment secondSegment = log.getNextSegment(firstSegment); Offset firstSegOffset1 = new Offset(firstSegment.getName(), firstSegment.getStartOffset()); Offset firstSegOffset2 = new Offset(firstSegment.getName(), firstSegment.getStartOffset() + availableSegCapacity / 2); Offset secondSegOffset1 = new Offset(secondSegment.getName(), secondSegment.getStartOffset()); Offset secondSegOffset2 = new Offset(secondSegment.getName(), secondSegment.getStartOffset() + availableSegCapacity / 2); List<MockId> mockIdList = new ArrayList<>(); MockId mockId = new MockId("id1"); mockIdList.add(mockId); BlobReadOptions ro1 = new BlobReadOptions(log, firstSegOffset2, new MessageInfo(mockId, availableSegCapacity / 3, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000))); mockId = new MockId("id2"); mockIdList.add(mockId); BlobReadOptions ro2 = new BlobReadOptions(log, secondSegOffset1, new MessageInfo(mockId, availableSegCapacity / 4, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000))); mockId = new MockId("id3"); mockIdList.add(mockId); BlobReadOptions ro3 = new BlobReadOptions(log, secondSegOffset2, new MessageInfo(mockId, availableSegCapacity / 2, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000))); mockId = new MockId("id4"); mockIdList.add(mockId); BlobReadOptions ro4 = new BlobReadOptions(log, firstSegOffset1, new MessageInfo(mockId, availableSegCapacity / 5, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000))); mockId = new MockId("id5"); mockIdList.add(mockId); BlobReadOptions ro5 = new BlobReadOptions(log, firstSegOffset2, new MessageInfo(mockId, availableSegCapacity / 6, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000))); List<BlobReadOptions> options = new ArrayList<>(Arrays.asList(ro1, ro2, ro3, ro4, ro5)); MessageReadSet readSet = new StoreMessageReadSet(options); assertEquals(readSet.count(), options.size()); assertEquals(readSet.getKeyAt(0), mockIdList.get(3)); assertEquals(readSet.sizeInBytes(0), availableSegCapacity / 5); assertEquals(readSet.getKeyAt(1), mockIdList.get(0)); assertEquals(readSet.sizeInBytes(1), availableSegCapacity / 3); assertEquals(readSet.getKeyAt(2), mockIdList.get(4)); assertEquals(readSet.sizeInBytes(2), availableSegCapacity / 6); assertEquals(readSet.getKeyAt(3), mockIdList.get(1)); assertEquals(readSet.sizeInBytes(3), availableSegCapacity / 4); assertEquals(readSet.getKeyAt(4), mockIdList.get(2)); assertEquals(readSet.sizeInBytes(4), availableSegCapacity / 2); ByteBuffer readBuf = ByteBuffer.allocate(availableSegCapacity / 5); ByteBufferOutputStream stream = new ByteBufferOutputStream(readBuf); if (doDataPrefetch) { readSet.doPrefetch(0, 0, Long.MAX_VALUE); } long written = readSet.writeTo(0, Channels.newChannel(stream), 0, Long.MAX_VALUE); assertEquals("Return value from writeTo() is incorrect", availableSegCapacity / 5, written); assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, 0, availableSegCapacity / 5)); readBuf = ByteBuffer.allocate(availableSegCapacity / 3); stream = new ByteBufferOutputStream(readBuf); WritableByteChannel channel = Channels.newChannel(stream); long currentReadOffset = 0; if (doDataPrefetch) { readSet.doPrefetch(1, currentReadOffset, availableSegCapacity / 3); } while (currentReadOffset < availableSegCapacity / 3) { written = readSet.writeTo(1, channel, currentReadOffset, 1); assertEquals("Return value from writeTo() is incorrect", 1, written); currentReadOffset++; } long startOffset = availableSegCapacity / 2; long endOffset = availableSegCapacity / 2 + availableSegCapacity / 3; assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, (int) startOffset, (int) endOffset)); readBuf = ByteBuffer.allocate(availableSegCapacity / 2); stream = new ByteBufferOutputStream(readBuf); channel = Channels.newChannel(stream); currentReadOffset = 0; if (doDataPrefetch) { readSet.doPrefetch(4, currentReadOffset, availableSegCapacity / 2); } while (currentReadOffset < availableSegCapacity / 2) { written = readSet.writeTo(4, channel, currentReadOffset, availableSegCapacity / 6); long expectedWritten = Math.min(availableSegCapacity / 2 - currentReadOffset, availableSegCapacity / 6); assertEquals("Return value from writeTo() is incorrect", expectedWritten, written); currentReadOffset += availableSegCapacity / 6; } startOffset = availableSegCapacity + availableSegCapacity / 2; endOffset = startOffset + availableSegCapacity / 2; assertArrayEquals(readBuf.array(), Arrays.copyOfRange(srcOfTruth, (int) startOffset, (int) endOffset)); readBuf = ByteBuffer.allocate(1); stream = new ByteBufferOutputStream(readBuf); channel = Channels.newChannel(stream); written = readSet.writeTo(0, channel, readSet.sizeInBytes(0), 1); assertEquals("No data should have been written", 0, written); try { readSet.sizeInBytes(options.size()); fail("Reading should have failed because index is out of bounds"); } catch (IndexOutOfBoundsException e) { } try { readSet.writeTo(options.size(), channel, 100, 100); fail("Reading should have failed because index is out of bounds"); } catch (IndexOutOfBoundsException e) { } try { readSet.getKeyAt(options.size()); fail("Getting key should have failed because index is out of bounds"); } catch (IndexOutOfBoundsException e) { } } finally { log.close(false); } }
StoreMessageReadSet(List<BlobReadOptions> readOptions) { Collections.sort(readOptions); this.readOptions = readOptions; }
StoreMessageReadSet implements MessageReadSet { StoreMessageReadSet(List<BlobReadOptions> readOptions) { Collections.sort(readOptions); this.readOptions = readOptions; } }
StoreMessageReadSet implements MessageReadSet { StoreMessageReadSet(List<BlobReadOptions> readOptions) { Collections.sort(readOptions); this.readOptions = readOptions; } StoreMessageReadSet(List<BlobReadOptions> readOptions); }
StoreMessageReadSet implements MessageReadSet { StoreMessageReadSet(List<BlobReadOptions> readOptions) { Collections.sort(readOptions); this.readOptions = readOptions; } StoreMessageReadSet(List<BlobReadOptions> readOptions); @Override long writeTo(int index, WritableByteChannel channel, long relativeOffset, long maxSize); @Override void writeTo(AsyncWritableChannel channel, Callback<Long> callback); @Override int count(); @Override long sizeInBytes(int index); @Override StoreKey getKeyAt(int index); @Override void doPrefetch(int index, long relativeOffset, long size); @Override ByteBuf getPrefetchedData(int index); }
StoreMessageReadSet implements MessageReadSet { StoreMessageReadSet(List<BlobReadOptions> readOptions) { Collections.sort(readOptions); this.readOptions = readOptions; } StoreMessageReadSet(List<BlobReadOptions> readOptions); @Override long writeTo(int index, WritableByteChannel channel, long relativeOffset, long maxSize); @Override void writeTo(AsyncWritableChannel channel, Callback<Long> callback); @Override int count(); @Override long sizeInBytes(int index); @Override StoreKey getKeyAt(int index); @Override void doPrefetch(int index, long relativeOffset, long size); @Override ByteBuf getPrefetchedData(int index);  }