You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by si...@apache.org on 2018/05/03 07:56:29 UTC
[incubator-pulsar] branch master updated: PIP-17: provide
BlockAwareSegmentInputStream implementation and test (#1678)
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new ab0d081 PIP-17: provide BlockAwareSegmentInputStream implementation and test (#1678)
ab0d081 is described below
commit ab0d081da0adb148a5f3aa94a86ab1589f103541
Author: Jia Zhai <zh...@gmail.com>
AuthorDate: Thu May 3 15:56:27 2018 +0800
PIP-17: provide BlockAwareSegmentInputStream implementation and test (#1678)
### Motivation
Implementation of InputStream(BlockAwareSegmentInputStream). when calls the is.read(), it return the following data:
- Block header
- Ledger entries starting from passed in entryId
- Padding in the with bytes 0xFEDCDEAD
This is based on PR #1669 (commit 3c8a846), And PR #1669 should be reviewed and merged first.
Master issue #1511
### Modifications
Add class BlockAwareSegmentInputStream and test for it.
### Result
unit test pass.
---
.../s3offload/BlockAwareSegmentInputStream.java | 70 ++++
.../impl/BlockAwareSegmentInputStreamImpl.java | 210 ++++++++++
.../BlockAwareSegmentInputStreamTest.java | 437 +++++++++++++++++++++
3 files changed, 717 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/BlockAwareSegmentInputStream.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/BlockAwareSegmentInputStream.java
new file mode 100644
index 0000000..c2d34da
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/BlockAwareSegmentInputStream.java
@@ -0,0 +1,70 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.s3offload;
+
+import java.io.IOException;
+import org.apache.bookkeeper.client.api.ReadHandle;
+
+/**
+ * The BlockAwareSegmentInputStream for each cold storage data block.
+ * This interface should be implemented while extends InputStream.
+ * It gets data from ledger, and will be read out the content for a data block.
+ * DataBlockHeader + entries(each with format[[entry_size -- int][entry_id -- long][entry_data]]) + padding
+ */
+public interface BlockAwareSegmentInputStream {
+ /**
+ * Get the ledger, from which this InputStream read data.
+ */
+ ReadHandle getLedger();
+
+ /**
+ * Get start entry id contained in this InputStream.
+ *
+ * @return the start entry id
+ */
+ long getStartEntryId();
+
+ /**
+ * Get block size that could read out from this InputStream.
+ *
+ * @return the block size
+ */
+ int getBlockSize();
+
+ /**
+ * Get entry count that read out from this InputStream
+ *
+ * @return the block entry count
+ */
+ int getBlockEntryCount();
+
+ /**
+ * Get end entry id contained in this InputStream.
+ *
+ * @return the end entry id
+ */
+ long getEndEntryId();
+
+ /**
+ * Get sum of entries data size read from the this InputStream
+ *
+ * @return the block entry bytes count
+ */
+ int getBlockEntryBytesCount();
+}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/impl/BlockAwareSegmentInputStreamImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/impl/BlockAwareSegmentInputStreamImpl.java
new file mode 100644
index 0000000..c83efc3
--- /dev/null
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/s3offload/impl/BlockAwareSegmentInputStreamImpl.java
@@ -0,0 +1,210 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.s3offload.impl;
+
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.common.collect.Lists;
+import com.google.common.primitives.Ints;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.CompositeByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.pulsar.broker.s3offload.BlockAwareSegmentInputStream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The BlockAwareSegmentInputStreamImpl for each cold storage data block.
+ * It gets data from ledger, and will be read out the content for a data block.
+ * DataBlockHeader + entries(each with format[[entry_size -- int][entry_id -- long][entry_data]]) + padding
+ */
+public class BlockAwareSegmentInputStreamImpl extends InputStream implements BlockAwareSegmentInputStream {
+ private static final Logger log = LoggerFactory.getLogger(BlockAwareSegmentInputStreamImpl.class);
+
+ private static final byte[] BLOCK_END_PADDING = Ints.toByteArray(0xFEDCDEAD);
+
+ private final ReadHandle ledger;
+ private final long startEntryId;
+ private final int blockSize;
+
+ // Number of Message entries that read from ledger and been readout from this InputStream.
+ private int blockEntryCount;
+
+ // tracking read status for both header and entries.
+ // Bytes that already been read from this InputStream
+ private int bytesReadOffset = 0;
+ // Byte from this index is all padding byte
+ private int dataBlockFullOffset;
+ private final InputStream dataBlockHeaderStream;
+
+ // how many entries want to read from ReadHandle each time.
+ private static final int ENTRIES_PER_READ = 100;
+ // buf the entry size and entry id.
+ private static final int ENTRY_HEADER_SIZE = 4 /* entry size*/ + 8 /* entry id */;
+ // Keep a list of all entries ByteBuf, each ByteBuf contains 2 buf: entry header and entry content.
+ private List<ByteBuf> entriesByteBuf = null;
+
+ public BlockAwareSegmentInputStreamImpl(ReadHandle ledger, long startEntryId, int blockSize) {
+ this.ledger = ledger;
+ this.startEntryId = startEntryId;
+ this.blockSize = blockSize;
+ this.dataBlockHeaderStream = DataBlockHeaderImpl.of(blockSize, startEntryId).toStream();
+ this.blockEntryCount = 0;
+ this.dataBlockFullOffset = blockSize;
+ this.entriesByteBuf = Lists.newLinkedList();
+ }
+
+ // read ledger entries.
+ private int readEntries() throws IOException {
+ checkState(bytesReadOffset >= DataBlockHeaderImpl.getDataStartOffset());
+ checkState(bytesReadOffset < blockSize);
+
+ // once reach the end of entry buffer, start a new read.
+ if (bytesReadOffset < dataBlockFullOffset && entriesByteBuf.isEmpty()) {
+ entriesByteBuf = readNextEntriesFromLedger(startEntryId + blockEntryCount, ENTRIES_PER_READ);
+ }
+
+ if ((!entriesByteBuf.isEmpty()) && bytesReadOffset + entriesByteBuf.get(0).readableBytes() > blockSize) {
+ // no space for a new entry, set data block full, return end padding
+ if (dataBlockFullOffset == blockSize) {
+ dataBlockFullOffset = bytesReadOffset;
+ }
+ return BLOCK_END_PADDING[(bytesReadOffset++ - dataBlockFullOffset) % BLOCK_END_PADDING.length];
+ } else {
+ // always read from the first ByteBuf in the list, once read all of its content remove it.
+ ByteBuf entryByteBuf = entriesByteBuf.get(0);
+ int ret = entryByteBuf.readByte();
+ bytesReadOffset++;
+
+ if (entryByteBuf.readableBytes() == 0) {
+ entryByteBuf.release();
+ entriesByteBuf.remove(0);
+ blockEntryCount++;
+ }
+
+ return ret;
+ }
+ }
+
+ private List<ByteBuf> readNextEntriesFromLedger(long start, long maxNumberEntries) throws IOException {
+ long end = Math.min(start + maxNumberEntries - 1, ledger.getLastAddConfirmed());
+ try (LedgerEntries ledgerEntriesOnce = ledger.readAsync(start, end).get()) {
+ log.debug("read ledger entries. start: {}, end: {}", start, end);
+
+ List<ByteBuf> entries = Lists.newLinkedList();
+
+ Iterator<LedgerEntry> iterator = ledgerEntriesOnce.iterator();
+ while (iterator.hasNext()) {
+ LedgerEntry entry = iterator.next();
+ int entryLength = (int) entry.getLength();
+ long entryId = entry.getEntryId();
+
+ CompositeByteBuf entryBuf = PooledByteBufAllocator.DEFAULT.compositeBuffer(2);
+ ByteBuf entryHeaderBuf = PooledByteBufAllocator.DEFAULT.buffer(ENTRY_HEADER_SIZE, ENTRY_HEADER_SIZE);
+
+ entryHeaderBuf.writeInt(entryLength).writeLong(entryId);
+ entryBuf.addComponents(true, entryHeaderBuf, entry.getEntryBuffer().retain());
+
+ entries.add(entryBuf);
+ }
+ return entries;
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("Exception when get CompletableFuture<LedgerEntries>. ", e);
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ throw new IOException(e);
+ }
+ }
+
+ @Override
+ public int read() throws IOException {
+ // reading header
+ if (dataBlockHeaderStream.available() > 0) {
+ bytesReadOffset++;
+ return dataBlockHeaderStream.read();
+ }
+
+ // reading Ledger entries.
+ if (bytesReadOffset < blockSize) {
+ return readEntries();
+ }
+
+ // reached end
+ return -1;
+ }
+
+ @Override
+ public void close() throws IOException {
+ super.close();
+ dataBlockHeaderStream.close();
+ if (!entriesByteBuf.isEmpty()) {
+ entriesByteBuf.forEach(buf -> buf.release());
+ entriesByteBuf.clear();
+ }
+ }
+
+ @Override
+ public ReadHandle getLedger() {
+ return ledger;
+ }
+
+ @Override
+ public long getStartEntryId() {
+ return startEntryId;
+ }
+
+ @Override
+ public int getBlockSize() {
+ return blockSize;
+ }
+
+ @Override
+ public int getBlockEntryCount() {
+ return blockEntryCount;
+ }
+
+ @Override
+ public long getEndEntryId() {
+ // return -1 when no entry contained
+ if (blockEntryCount == 0) {
+ return -1;
+ }
+ return startEntryId + blockEntryCount - 1;
+ }
+
+ @Override
+ public int getBlockEntryBytesCount() {
+ return dataBlockFullOffset - DataBlockHeaderImpl.getDataStartOffset() - ENTRY_HEADER_SIZE * blockEntryCount;
+ }
+
+ public static byte[] getBlockEndPadding() {
+ return BLOCK_END_PADDING;
+ }
+
+}
+
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/s3offload/BlockAwareSegmentInputStreamTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/s3offload/BlockAwareSegmentInputStreamTest.java
new file mode 100644
index 0000000..bc2c122
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/s3offload/BlockAwareSegmentInputStreamTest.java
@@ -0,0 +1,437 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.s3offload;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+import static org.testng.internal.junit.ArrayAsserts.assertArrayEquals;
+
+import com.google.common.io.ByteStreams;
+import com.google.common.primitives.Ints;
+import com.google.common.primitives.Longs;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.util.concurrent.DefaultThreadFactory;
+import java.io.ByteArrayInputStream;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.stream.IntStream;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.api.LastConfirmedAndEntry;
+import org.apache.bookkeeper.client.api.LedgerEntries;
+import org.apache.bookkeeper.client.api.LedgerEntry;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.pulsar.broker.s3offload.DataBlockHeader;
+import org.apache.pulsar.broker.s3offload.impl.BlockAwareSegmentInputStreamImpl;
+import org.apache.pulsar.broker.s3offload.impl.DataBlockHeaderImpl;
+import org.testng.annotations.Test;
+import org.testng.collections.Lists;
+
+@Slf4j
+public class BlockAwareSegmentInputStreamTest {
+ @Data
+ class MockLedgerEntry implements LedgerEntry {
+ public byte blockPadding = 0xB;
+ long ledgerId;
+ long entryId;
+ long length;
+ byte entryBytes[];
+ ByteBuf entryBuffer;
+
+ MockLedgerEntry(long ledgerId, long entryId, long length) {
+ this.ledgerId = ledgerId;
+ this.entryId = entryId;
+ this.length = length;
+ this.entryBytes = new byte[(int)length];
+ entryBuffer = Unpooled.wrappedBuffer(entryBytes);
+ entryBuffer.writerIndex(0);
+ IntStream.range(0, (int)length).forEach(i -> entryBuffer.writeByte(blockPadding));
+ }
+
+ @Override
+ public ByteBuffer getEntryNioBuffer() {
+ return null;
+ }
+
+ @Override
+ public LedgerEntry duplicate() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ entryBuffer.release();
+ }
+ }
+
+ @Data
+ class MockLedgerEntries implements LedgerEntries {
+ int ledgerId;
+ int startEntryId;
+ int count;
+ int entrySize;
+ List<LedgerEntry> entries;
+
+ MockLedgerEntries(int ledgerId, int startEntryId, int count, int entrySize) {
+ this.ledgerId = ledgerId;
+ this.startEntryId = startEntryId;
+ this.count = count;
+ this.entrySize = entrySize;
+ this.entries = Lists.newArrayList(count);
+
+ IntStream.range(startEntryId, startEntryId + count).forEach(i ->
+ entries.add(new MockLedgerEntry(ledgerId, i, entrySize)));
+ }
+
+ @Override
+ public void close() {
+ entries.clear();
+ }
+
+ @Override
+ public LedgerEntry getEntry(long entryId) {
+ if (entryId < startEntryId || entryId >= startEntryId + count) {
+ return null;
+ }
+
+ return entries.get(((int)entryId - startEntryId));
+ }
+
+ @Override
+ public Iterator<LedgerEntry> iterator() {
+ return entries.iterator();
+ }
+ }
+
+ class MockReadHandle implements ReadHandle {
+ int ledgerId;
+ int entrySize;
+ int lac;
+ MockReadHandle(int ledgerId, int entrySize, int lac) {
+ this.ledgerId = ledgerId;
+ this.entrySize = entrySize;
+ this.lac = lac;
+ }
+
+ @Override
+ public CompletableFuture<LedgerEntries> readAsync(long firstEntry, long lastEntry) {
+ CompletableFuture<LedgerEntries> future = new CompletableFuture<>();
+ LedgerEntries entries = new MockLedgerEntries(ledgerId,
+ (int)firstEntry,
+ (int)(lastEntry - firstEntry + 1),
+ entrySize);
+
+ future.complete(entries);
+ return future;
+ }
+
+ @Override
+ public CompletableFuture<LedgerEntries> readUnconfirmedAsync(long firstEntry, long lastEntry) {
+ return readAsync(firstEntry, lastEntry);
+ }
+
+ @Override
+ public CompletableFuture<Long> readLastAddConfirmedAsync() {
+ return null;
+ }
+
+ @Override
+ public CompletableFuture<Long> tryReadLastAddConfirmedAsync() {
+ return null;
+ }
+
+ @Override
+ public long getLastAddConfirmed() {
+ return lac;
+ }
+
+ @Override
+ public long getLength() {
+ return (lac + 1) * entrySize;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return true;
+ }
+
+ @Override
+ public CompletableFuture<LastConfirmedAndEntry>
+ readLastAddConfirmedAndEntryAsync(long entryId, long timeOutInMillis, boolean parallel) {
+ return null;
+ }
+
+ @Override
+ public LedgerMetadata getLedgerMetadata() {
+ return null;
+ }
+
+ @Override
+ public long getId() {
+ return ledgerId;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return null;
+ }
+ }
+
+ @Test
+ public void testHaveEndPadding() throws Exception {
+ int ledgerId = 1;
+ int entrySize = 8;
+ int lac = 160;
+ ReadHandle readHandle = new MockReadHandle(ledgerId, entrySize, lac);
+
+ // set block size bigger than to (header + entry) size.
+ int blockSize = 3148 + 5;
+ BlockAwareSegmentInputStreamImpl inputStream = new BlockAwareSegmentInputStreamImpl(readHandle, 0, blockSize);
+ int expectedEntryCount = (blockSize - DataBlockHeaderImpl.getDataStartOffset()) / (entrySize + 4 + 8);
+
+ // verify get methods
+ assertEquals(inputStream.getLedger(), readHandle);
+ assertEquals(inputStream.getStartEntryId(), 0);
+ assertEquals(inputStream.getBlockSize(), blockSize);
+
+ // verify read inputStream
+ // 1. read header. 128
+ byte headerB[] = new byte[DataBlockHeaderImpl.getDataStartOffset()];
+ ByteStreams.readFully(inputStream, headerB);
+ DataBlockHeader headerRead = DataBlockHeaderImpl.fromStream(new ByteArrayInputStream(headerB));
+ assertEquals(headerRead.getBlockLength(), blockSize);
+ assertEquals(headerRead.getFirstEntryId(), 0);
+
+ byte[] entryData = new byte[entrySize];
+ Arrays.fill(entryData, (byte)0xB); // 0xB is MockLedgerEntry.blockPadding
+
+ // 2. read Ledger entries. 201 * 20
+ IntStream.range(0, expectedEntryCount).forEach(i -> {
+ try {
+ byte lengthBuf[] = new byte[4];
+ byte entryIdBuf[] = new byte[8];
+ byte content[] = new byte[entrySize];
+ inputStream.read(lengthBuf);
+ inputStream.read(entryIdBuf);
+ inputStream.read(content);
+
+ assertEquals(entrySize, Ints.fromByteArray(lengthBuf));
+ assertEquals(i, Longs.fromByteArray(entryIdBuf));
+ assertArrayEquals(entryData, content);
+ } catch (Exception e) {
+ fail("meet exception", e);
+ }
+ });
+
+ // 3. read padding
+ int left = blockSize - DataBlockHeaderImpl.getDataStartOffset() - expectedEntryCount * (entrySize + 4 + 8);
+ assertEquals(left, 5);
+ byte padding[] = new byte[left];
+ inputStream.read(padding);
+ ByteBuf paddingBuf = Unpooled.wrappedBuffer(padding);
+ IntStream.range(0, paddingBuf.capacity()/4).forEach(i ->
+ assertEquals(Integer.toHexString(paddingBuf.readInt()),
+ Integer.toHexString(Ints.fromByteArray(inputStream.getBlockEndPadding())))
+ );
+
+ // 4. reach end.
+ assertEquals(inputStream.read(), -1);
+
+ assertEquals(inputStream.getBlockEntryCount(), expectedEntryCount);
+ assertEquals(inputStream.getBlockEntryBytesCount(), entrySize * expectedEntryCount);
+ assertEquals(inputStream.getEndEntryId(), expectedEntryCount - 1);
+
+ inputStream.close();
+ }
+
+ @Test
+ public void testNoEndPadding() throws Exception {
+ int ledgerId = 1;
+ int entrySize = 8;
+ int lac = 120;
+ ReadHandle readHandle = new MockReadHandle(ledgerId, entrySize, lac);
+
+ // set block size equals to (header + entry) size.
+ int blockSize = 2148;
+ BlockAwareSegmentInputStreamImpl inputStream = new BlockAwareSegmentInputStreamImpl(readHandle, 0, blockSize);
+ int expectedEntryCount = (blockSize - DataBlockHeaderImpl.getDataStartOffset()) / (entrySize + 4 + 8);
+
+ // verify get methods
+ assertEquals(inputStream.getLedger(), readHandle);
+ assertEquals(inputStream.getStartEntryId(), 0);
+ assertEquals(inputStream.getBlockSize(), blockSize);
+
+ // verify read inputStream
+ // 1. read header. 128
+ byte headerB[] = new byte[DataBlockHeaderImpl.getDataStartOffset()];
+ ByteStreams.readFully(inputStream, headerB);
+ DataBlockHeader headerRead = DataBlockHeaderImpl.fromStream(new ByteArrayInputStream(headerB));
+ assertEquals(headerRead.getBlockLength(), blockSize);
+ assertEquals(headerRead.getFirstEntryId(), 0);
+
+ byte[] entryData = new byte[entrySize];
+ Arrays.fill(entryData, (byte)0xB); // 0xB is MockLedgerEntry.blockPadding
+
+ // 2. read Ledger entries. 201 * 20
+ IntStream.range(0, expectedEntryCount).forEach(i -> {
+ try {
+ byte lengthBuf[] = new byte[4];
+ byte entryIdBuf[] = new byte[8];
+ byte content[] = new byte[entrySize];
+ inputStream.read(lengthBuf);
+ inputStream.read(entryIdBuf);
+ inputStream.read(content);
+
+ assertEquals(entrySize, Ints.fromByteArray(lengthBuf));
+ assertEquals(i, Longs.fromByteArray(entryIdBuf));
+ assertArrayEquals(entryData, content);
+ } catch (Exception e) {
+ fail("meet exception", e);
+ }
+ });
+
+ // 3. should be no padding
+ int left = blockSize - DataBlockHeaderImpl.getDataStartOffset() - expectedEntryCount * (entrySize + 4 + 8);
+ assertEquals(left, 0);
+
+ // 4. reach end.
+ assertEquals(inputStream.read(), -1);
+
+ assertEquals(inputStream.getBlockEntryCount(), expectedEntryCount);
+ assertEquals(inputStream.getBlockEntryBytesCount(), entrySize * expectedEntryCount);
+ assertEquals(inputStream.getEndEntryId(), expectedEntryCount - 1);
+
+ inputStream.close();
+ }
+
+ @Test
+ public void testReadTillLac() throws Exception {
+ // simulate last data block read.
+ int ledgerId = 1;
+ int entrySize = 8;
+ int lac = 89;
+ ReadHandle readHandle = new MockReadHandle(ledgerId, entrySize, lac);
+
+ // set block size equals to (header + lac_entry) size.
+ int blockSize = DataBlockHeaderImpl.getDataStartOffset() + (1 + lac) * (entrySize + 4 + 8);
+ BlockAwareSegmentInputStreamImpl inputStream = new BlockAwareSegmentInputStreamImpl(readHandle, 0, blockSize);
+ int expectedEntryCount = (blockSize - DataBlockHeaderImpl.getDataStartOffset()) / (entrySize + 4 + 8);
+
+ // verify get methods
+ assertEquals(inputStream.getLedger(), readHandle);
+ assertEquals(inputStream.getStartEntryId(), 0);
+ assertEquals(inputStream.getBlockSize(), blockSize);
+
+ // verify read inputStream
+ // 1. read header. 128
+ byte headerB[] = new byte[DataBlockHeaderImpl.getDataStartOffset()];
+ ByteStreams.readFully(inputStream, headerB);
+ DataBlockHeader headerRead = DataBlockHeaderImpl.fromStream(new ByteArrayInputStream(headerB));
+ assertEquals(headerRead.getBlockLength(), blockSize);
+ assertEquals(headerRead.getFirstEntryId(), 0);
+
+ byte[] entryData = new byte[entrySize];
+ Arrays.fill(entryData, (byte)0xB); // 0xB is MockLedgerEntry.blockPadding
+
+ // 2. read Ledger entries. 96 * 20
+ IntStream.range(0, expectedEntryCount).forEach(i -> {
+ try {
+ byte lengthBuf[] = new byte[4];
+ byte entryIdBuf[] = new byte[8];
+ byte content[] = new byte[entrySize];
+ inputStream.read(lengthBuf);
+ inputStream.read(entryIdBuf);
+ inputStream.read(content);
+
+ assertEquals(entrySize, Ints.fromByteArray(lengthBuf));
+ assertEquals(i, Longs.fromByteArray(entryIdBuf));
+ assertArrayEquals(entryData, content);
+ } catch (Exception e) {
+ fail("meet exception", e);
+ }
+ });
+
+ // 3. should have no padding
+ int left = blockSize - DataBlockHeaderImpl.getDataStartOffset() - expectedEntryCount * (entrySize + 4 + 8);
+ assertEquals(left, 0);
+
+ // 4. reach end.
+ assertEquals(inputStream.read(), -1);
+
+ assertEquals(inputStream.getBlockEntryCount(), expectedEntryCount);
+ assertEquals(inputStream.getBlockEntryBytesCount(), entrySize * expectedEntryCount);
+ assertEquals(inputStream.getEndEntryId(), expectedEntryCount - 1);
+
+ inputStream.close();
+ }
+
+ @Test
+ public void testNoEntryPutIn() throws Exception {
+ // simulate first entry size over the block size budget, it shouldn't be added.
+ // 2 entries, each with bigger size than block size, so there should no entry added into block.
+ int ledgerId = 1;
+ int entrySize = 1000;
+ int lac = 1;
+ ReadHandle readHandle = new MockReadHandle(ledgerId, entrySize, lac);
+
+ // set block size not able to hold one entry
+ int blockSize = DataBlockHeaderImpl.getDataStartOffset() + entrySize;
+ BlockAwareSegmentInputStreamImpl inputStream = new BlockAwareSegmentInputStreamImpl(readHandle, 0, blockSize);
+ int expectedEntryCount = 0;
+
+ // verify get methods
+ assertEquals(inputStream.getLedger(), readHandle);
+ assertEquals(inputStream.getStartEntryId(), 0);
+ assertEquals(inputStream.getBlockSize(), blockSize);
+
+ // verify read inputStream
+ // 1. read header. 128
+ byte headerB[] = new byte[DataBlockHeaderImpl.getDataStartOffset()];
+ ByteStreams.readFully(inputStream, headerB);
+ DataBlockHeader headerRead = DataBlockHeaderImpl.fromStream(new ByteArrayInputStream(headerB));
+ assertEquals(headerRead.getBlockLength(), blockSize);
+ assertEquals(headerRead.getFirstEntryId(), 0);
+
+
+ // 2. since no entry put in, it should only get padding after header.
+ byte padding[] = new byte[blockSize - DataBlockHeaderImpl.getDataStartOffset()];
+ inputStream.read(padding);
+ ByteBuf paddingBuf = Unpooled.wrappedBuffer(padding);
+ IntStream.range(0, paddingBuf.capacity()/4).forEach(i ->
+ assertEquals(Integer.toHexString(paddingBuf.readInt()),
+ Integer.toHexString(Ints.fromByteArray(inputStream.getBlockEndPadding())))
+ );
+
+ // 3. reach end.
+ assertEquals(inputStream.read(), -1);
+
+ assertEquals(inputStream.getBlockEntryCount(), 0);
+ assertEquals(inputStream.getBlockEntryBytesCount(), 0);
+ assertEquals(inputStream.getEndEntryId(), -1);
+
+ inputStream.close();
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
sijie@apache.org.