You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by op...@apache.org on 2019/04/10 06:09:50 UTC
[hbase] branch branch-2 updated: HBASE-22185
RAMQueueEntry#writeToCache should freeBlock if any exception encountered
instead of the IOException catch block
This is an automated email from the ASF dual-hosted git repository.
openinx pushed a commit to branch branch-2
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2 by this push:
new 2e798f8 HBASE-22185 RAMQueueEntry#writeToCache should freeBlock if any exception encountered instead of the IOException catch block
2e798f8 is described below
commit 2e798f87b79b888b593a68f85d6aa41f5c0eea8d
Author: huzheng <op...@gmail.com>
AuthorDate: Mon Apr 8 23:15:46 2019 +0800
HBASE-22185 RAMQueueEntry#writeToCache should freeBlock if any exception encountered instead of the IOException catch block
---
.../hadoop/hbase/io/hfile/bucket/BucketCache.java | 49 ++++++++++++----------
.../hbase/io/hfile/bucket/TestBucketCache.java | 39 +++++++++++++++++
2 files changed, 67 insertions(+), 21 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
index 70bf9f4..2882545 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
@@ -1511,30 +1511,37 @@ public class BucketCache implements BlockCache, HeapSize {
this.accessCounter = accessCounter;
}
- public BucketEntry writeToCache(final IOEngine ioEngine,
- final BucketAllocator bucketAllocator,
- final UniqueIndexMap<Integer> deserialiserMap,
- final LongAdder realCacheSize) throws CacheFullException, IOException,
- BucketAllocatorException {
+ private BucketEntry getBucketEntry(IOEngine ioEngine, long offset, int len) {
+ if (ioEngine.usesSharedMemory()) {
+ if (UnsafeAvailChecker.isAvailable()) {
+ return new UnsafeSharedMemoryBucketEntry(offset, len, accessCounter, inMemory);
+ } else {
+ return new SharedMemoryBucketEntry(offset, len, accessCounter, inMemory);
+ }
+ } else {
+ return new BucketEntry(offset, len, accessCounter, inMemory);
+ }
+ }
+
+ public BucketEntry writeToCache(final IOEngine ioEngine, final BucketAllocator bucketAllocator,
+ final UniqueIndexMap<Integer> deserialiserMap, final LongAdder realCacheSize)
+ throws IOException {
int len = data.getSerializedLength();
// This cacheable thing can't be serialized
- if (len == 0) return null;
+ if (len == 0) {
+ return null;
+ }
long offset = bucketAllocator.allocateBlock(len);
- BucketEntry bucketEntry = ioEngine.usesSharedMemory()
- ? UnsafeAvailChecker.isAvailable()
- ? new UnsafeSharedMemoryBucketEntry(offset, len, accessCounter, inMemory)
- : new SharedMemoryBucketEntry(offset, len, accessCounter, inMemory)
- : new BucketEntry(offset, len, accessCounter, inMemory);
- bucketEntry.setDeserialiserReference(data.getDeserializer(), deserialiserMap);
+ boolean succ = false;
+ BucketEntry bucketEntry;
try {
+ bucketEntry = getBucketEntry(ioEngine, offset, len);
+ bucketEntry.setDeserialiserReference(data.getDeserializer(), deserialiserMap);
if (data instanceof HFileBlock) {
// If an instance of HFileBlock, save on some allocations.
- HFileBlock block = (HFileBlock)data;
+ HFileBlock block = (HFileBlock) data;
ByteBuff sliceBuf = block.getBufferReadOnly();
ByteBuffer metadata = block.getMetaData();
- if (LOG.isTraceEnabled()) {
- LOG.trace("Write offset=" + offset + ", len=" + len);
- }
ioEngine.write(sliceBuf, offset);
ioEngine.write(metadata, offset + len - metadata.limit());
} else {
@@ -1542,12 +1549,12 @@ public class BucketCache implements BlockCache, HeapSize {
data.serialize(bb, true);
ioEngine.write(bb, offset);
}
- } catch (IOException ioe) {
- // free it in bucket allocator
- bucketAllocator.freeBlock(offset);
- throw ioe;
+ succ = true;
+ } finally {
+ if (!succ) {
+ bucketAllocator.freeBlock(offset);
+ }
}
-
realCacheSize.add(len);
return bucketEntry;
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java
index 924dd02..5363d70 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestBucketCache.java
@@ -48,15 +48,19 @@ import org.apache.hadoop.hbase.io.hfile.HFileContext;
import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.BucketSizeInfo;
import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.IndexStatistics;
+import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMQueueEntry;
+import org.apache.hadoop.hbase.nio.ByteBuff;
import org.apache.hadoop.hbase.testclassification.IOTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
+import org.mockito.Mockito;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
@@ -454,4 +458,39 @@ public class TestBucketCache {
CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
block1Buffer);
}
+
+ @Test
+ public void testFreeBlockWhenIOEngineWriteFailure() throws IOException {
+ // initialize an block.
+ int size = 100, offset = 20;
+ int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
+ ByteBuffer buf = ByteBuffer.allocate(length);
+ HFileContext meta = new HFileContextBuilder().build();
+ HFileBlock block = new HFileBlock(BlockType.DATA, size, size, -1, buf, HFileBlock.FILL_HEADER,
+ offset, 52, -1, meta);
+
+ // initialize an mocked ioengine.
+ IOEngine ioEngine = Mockito.mock(IOEngine.class);
+ Mockito.when(ioEngine.usesSharedMemory()).thenReturn(false);
+ // Mockito.doNothing().when(ioEngine).write(Mockito.any(ByteBuffer.class), Mockito.anyLong());
+ Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuffer.class),
+ Mockito.anyLong());
+ Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuff.class),
+ Mockito.anyLong());
+
+ // create an bucket allocator.
+ long availableSpace = 1024 * 1024 * 1024L;
+ BucketAllocator allocator = new BucketAllocator(availableSpace, null);
+
+ BlockCacheKey key = new BlockCacheKey("dummy", 1L);
+ RAMQueueEntry re = new RAMQueueEntry(key, block, 1, true);
+
+ Assert.assertEquals(0, allocator.getUsedSize());
+ try {
+ re.writeToCache(ioEngine, allocator, new UniqueIndexMap<>(), null);
+ Assert.fail();
+ } catch (Exception e) {
+ }
+ Assert.assertEquals(0, allocator.getUsedSize());
+ }
}