@Test public void blobReadOptionsTest() throws IOException, StoreException { int logCapacity = 2000; int[] segCapacities = {2000, 1000}; for (int segCapacity : segCapacities) { 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()); int count = logCapacity / segCapacity; for (int i = 0; i < count; i++) { ByteBuffer buffer = ByteBuffer.allocate(availableSegCapacity); log.appendFrom(Channels.newChannel(new ByteBufferInputStream(buffer)), availableSegCapacity); } long offset = Utils.getRandomLong(TestUtils.RANDOM, availableSegCapacity) + firstSegment.getStartOffset(); long size = Utils.getRandomLong(TestUtils.RANDOM, firstSegment.getEndOffset() - offset); long expiresAtMs = Utils.getRandomLong(TestUtils.RANDOM, 1000); long operationTimeMs = System.currentTimeMillis() + TestUtils.RANDOM.nextInt(10000); long crc = TestUtils.RANDOM.nextLong(); MockId id = new MockId("id1"); short accountId = Utils.getRandomShort(TestUtils.RANDOM); short containerId = Utils.getRandomShort(TestUtils.RANDOM); boolean deleted = TestUtils.RANDOM.nextBoolean(); boolean ttlUpdated = TestUtils.RANDOM.nextBoolean(); MessageInfo info = new MessageInfo(id, size, deleted, ttlUpdated, expiresAtMs, crc, accountId, containerId, operationTimeMs); BlobReadOptions options = new BlobReadOptions(log, new Offset(firstSegment.getName(), offset), info); assertEquals("Ref count of log segment should have increased", 1, firstSegment.refCount()); verifyGetters(options, firstSegment, offset, true, info); options.close(); assertEquals("Ref count of log segment should have decreased", 0, firstSegment.refCount()); doSerDeTest(options, firstSegment, log); if (count > 1) { LogSegment secondSegment = log.getNextSegment(firstSegment); options = new BlobReadOptions(log, new Offset(secondSegment.getName(), offset), info); assertEquals("Ref count of log segment should have increased", 1, secondSegment.refCount()); options.close(); assertEquals("Ref count of log segment should have decreased", 0, secondSegment.refCount()); doSerDeTest(options, secondSegment, log); } try { new BlobReadOptions(log, new Offset(firstSegment.getName(), firstSegment.getEndOffset()), new MessageInfo(null, 1, 1, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), operationTimeMs)); fail("Construction should have failed because offset + size > endOffset"); } catch (IllegalArgumentException e) { } } finally { log.close(false); assertTrue(tempDir + " could not be cleaned", StoreTestUtils.cleanDirectory(tempDir, false)); } } }
BlobReadOptions(Log log, Offset offset, MessageInfo info) { segment = log.getSegment(offset.getName()); if (offset.getOffset() + info.getSize() > segment.getEndOffset()) { throw new IllegalArgumentException( "Invalid offset [" + offset + "] and size [" + info.getSize() + "]. Segment end offset: " + "[" + segment.getEndOffset() + "]"); } segmentView = segment.getView(); this.offset = offset; this.info = info; logger.trace("BlobReadOption offset {} size {} MessageInfo {} ", offset, info.getSize(), info); }
BlobReadOptions implements Comparable<BlobReadOptions>, Closeable { BlobReadOptions(Log log, Offset offset, MessageInfo info) { segment = log.getSegment(offset.getName()); if (offset.getOffset() + info.getSize() > segment.getEndOffset()) { throw new IllegalArgumentException( "Invalid offset [" + offset + "] and size [" + info.getSize() + "]. Segment end offset: " + "[" + segment.getEndOffset() + "]"); } segmentView = segment.getView(); this.offset = offset; this.info = info; logger.trace("BlobReadOption offset {} size {} MessageInfo {} ", offset, info.getSize(), info); } }
BlobReadOptions implements Comparable<BlobReadOptions>, Closeable { BlobReadOptions(Log log, Offset offset, MessageInfo info) { segment = log.getSegment(offset.getName()); if (offset.getOffset() + info.getSize() > segment.getEndOffset()) { throw new IllegalArgumentException( "Invalid offset [" + offset + "] and size [" + info.getSize() + "]. Segment end offset: " + "[" + segment.getEndOffset() + "]"); } segmentView = segment.getView(); this.offset = offset; this.info = info; logger.trace("BlobReadOption offset {} size {} MessageInfo {} ", offset, info.getSize(), info); } BlobReadOptions(Log log, Offset offset, MessageInfo info); }
BlobReadOptions implements Comparable<BlobReadOptions>, Closeable { BlobReadOptions(Log log, Offset offset, MessageInfo info) { segment = log.getSegment(offset.getName()); if (offset.getOffset() + info.getSize() > segment.getEndOffset()) { throw new IllegalArgumentException( "Invalid offset [" + offset + "] and size [" + info.getSize() + "]. Segment end offset: " + "[" + segment.getEndOffset() + "]"); } segmentView = segment.getView(); this.offset = offset; this.info = info; logger.trace("BlobReadOption offset {} size {} MessageInfo {} ", offset, info.getSize(), info); } BlobReadOptions(Log log, Offset offset, MessageInfo info); @Override int compareTo(BlobReadOptions o); @Override void close(); }
BlobReadOptions implements Comparable<BlobReadOptions>, Closeable { BlobReadOptions(Log log, Offset offset, MessageInfo info) { segment = log.getSegment(offset.getName()); if (offset.getOffset() + info.getSize() > segment.getEndOffset()) { throw new IllegalArgumentException( "Invalid offset [" + offset + "] and size [" + info.getSize() + "]. Segment end offset: " + "[" + segment.getEndOffset() + "]"); } segmentView = segment.getView(); this.offset = offset; this.info = info; logger.trace("BlobReadOption offset {} size {} MessageInfo {} ", offset, info.getSize(), info); } BlobReadOptions(Log log, Offset offset, MessageInfo info); @Override int compareTo(BlobReadOptions o); @Override void close();  }