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/07/04 07:00:13 UTC
[hbase] branch branch-2.1 updated: HBASE-22582 The Compaction
writer may access the lastCell whose memory has been released when
appending fileInfo in the final (#341)
This is an automated email from the ASF dual-hosted git repository.
openinx pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-2.1 by this push:
new 6d5e120 HBASE-22582 The Compaction writer may access the lastCell whose memory has been released when appending fileInfo in the final (#341)
6d5e120 is described below
commit 6d5e120596587ee007934739efddf04b63437dfa
Author: openinx <op...@gmail.com>
AuthorDate: Thu Jul 4 14:27:21 2019 +0800
HBASE-22582 The Compaction writer may access the lastCell whose memory has been released when appending fileInfo in the final (#341)
---
.../hadoop/hbase/io/hfile/HFileWriterImpl.java | 7 ++
.../hadoop/hbase/mob/DefaultMobStoreCompactor.java | 15 +--
.../hadoop/hbase/regionserver/StoreScanner.java | 5 +-
.../hbase/regionserver/compactions/Compactor.java | 28 +++--
.../hbase/regionserver/TestCompactorMemLeak.java | 137 +++++++++++++++++++++
5 files changed, 169 insertions(+), 23 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
index cfc3dd9..104694a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/HFileWriterImpl.java
@@ -55,6 +55,8 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.io.Writable;
+import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+
/**
* Common functionality needed by all versions of {@link HFile} writers.
*/
@@ -773,6 +775,11 @@ public class HFileWriterImpl implements HFile.Writer {
}
}
+ @VisibleForTesting
+ public Cell getLastCell() {
+ return lastCell;
+ }
+
protected void finishFileInfo() throws IOException {
if (lastCell != null) {
// Make a copy. The copy is stuffed into our fileinfo map. Needs a clean
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
index 01c195a..3b00cc4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/mob/DefaultMobStoreCompactor.java
@@ -310,13 +310,10 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
// logging at DEBUG level
if (LOG.isDebugEnabled()) {
if ((now - lastMillis) >= COMPACTION_PROGRESS_LOG_INTERVAL) {
- LOG.debug("Compaction progress: "
- + compactionName
- + " "
- + progress
- + String.format(", rate=%.2f kB/sec", (bytesWrittenProgressForLog / 1024.0)
- / ((now - lastMillis) / 1000.0)) + ", throughputController is "
- + throughputController);
+ String rate = String.format("%.2f",
+ (bytesWrittenProgressForLog / 1024.0) / ((now - lastMillis) / 1000.0));
+ LOG.debug("Compaction progress: {} {}, rate={} KB/sec, throughputController is {}",
+ compactionName, progress, rate, throughputController);
lastMillis = now;
bytesWrittenProgressForLog = 0;
}
@@ -329,6 +326,10 @@ public class DefaultMobStoreCompactor extends DefaultCompactor {
throw new InterruptedIOException(
"Interrupted while control throughput of compacting " + compactionName);
} finally {
+ // Clone last cell in the final because writer will append last cell when committing. If
+ // don't clone here and once the scanner get closed, then the memory of last cell will be
+ // released. (HBASE-22582)
+ ((ShipperListener) writer).beforeShipped();
throughputController.finish(compactionName);
if (!finished && mobFileWriter != null) {
abortWriter(mobFileWriter);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
index 3efa9ad..dfc4251 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreScanner.java
@@ -1004,9 +1004,8 @@ public class StoreScanner extends NonReversedNonLazyKeyValueScanner
protected void checkScanOrder(Cell prevKV, Cell kv,
CellComparator comparator) throws IOException {
// Check that the heap gives us KVs in an increasing order.
- assert prevKV == null || comparator == null
- || comparator.compare(prevKV, kv) <= 0 : "Key " + prevKV
- + " followed by a " + "smaller key " + kv + " in cf " + store;
+ assert prevKV == null || comparator == null || comparator.compare(prevKV, kv) <= 0 : "Key "
+ + prevKV + " followed by a smaller key " + kv + " in cf " + store;
}
protected boolean seekToNextRow(Cell c) throws IOException {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
index 83690a9..90370ee 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java
@@ -377,11 +377,12 @@ public abstract class Compactor<T extends CellSink> {
long now = 0;
boolean hasMore;
ScannerContext scannerContext =
- ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
+ ScannerContext.newBuilder().setBatchLimit(compactionKVMax).build();
throughputController.start(compactionName);
- KeyValueScanner kvs = (scanner instanceof KeyValueScanner)? (KeyValueScanner)scanner : null;
- long shippedCallSizeLimit = (long) numofFilesToCompact * this.store.getColumnFamilyDescriptor().getBlocksize();
+ KeyValueScanner kvs = (scanner instanceof KeyValueScanner) ? (KeyValueScanner) scanner : null;
+ long shippedCallSizeLimit =
+ (long) numofFilesToCompact * this.store.getColumnFamilyDescriptor().getBlocksize();
try {
do {
hasMore = scanner.next(cells, scannerContext);
@@ -429,7 +430,7 @@ public abstract class Compactor<T extends CellSink> {
}
// Clone the cells that are in the writer so that they are freed of references,
// if they are holding any.
- ((ShipperListener)writer).beforeShipped();
+ ((ShipperListener) writer).beforeShipped();
// The SHARED block references, being read for compaction, will be kept in prevBlocks
// list(See HFileScannerImpl#prevBlocks). In case of scan flow, after each set of cells
// being returned to client, we will call shipped() which can clear this list. Here by
@@ -448,13 +449,10 @@ public abstract class Compactor<T extends CellSink> {
// logging at DEBUG level
if (LOG.isDebugEnabled()) {
if ((now - lastMillis) >= COMPACTION_PROGRESS_LOG_INTERVAL) {
- LOG.debug("Compaction progress: "
- + compactionName
- + " "
- + progress
- + String.format(", rate=%.2f kB/sec", (bytesWrittenProgressForLog / 1024.0)
- / ((now - lastMillis) / 1000.0)) + ", throughputController is "
- + throughputController);
+ String rate = String.format("%.2f",
+ (bytesWrittenProgressForLog / 1024.0) / ((now - lastMillis) / 1000.0));
+ LOG.debug("Compaction progress: {} {}, rate={} KB/sec, throughputController is {}",
+ compactionName, progress, rate, throughputController);
lastMillis = now;
bytesWrittenProgressForLog = 0;
}
@@ -463,9 +461,13 @@ public abstract class Compactor<T extends CellSink> {
} while (hasMore);
} catch (InterruptedException e) {
progress.cancel();
- throw new InterruptedIOException("Interrupted while control throughput of compacting "
- + compactionName);
+ throw new InterruptedIOException(
+ "Interrupted while control throughput of compacting " + compactionName);
} finally {
+ // Clone last cell in the final because writer will append last cell when committing. If
+ // don't clone here and once the scanner get closed, then the memory of last cell will be
+ // released. (HBASE-22582)
+ ((ShipperListener) writer).beforeShipped();
throughputController.finish(compactionName);
}
progress.complete();
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactorMemLeak.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactorMemLeak.java
new file mode 100644
index 0000000..d334902
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestCompactorMemLeak.java
@@ -0,0 +1,137 @@
+/**
+ * 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.hadoop.hbase.regionserver;
+
+import static org.apache.hadoop.hbase.regionserver.DefaultStoreEngine.DEFAULT_COMPACTOR_CLASS_KEY;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtility;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.KeyValue.KeyOnlyKeyValue;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.io.hfile.HFileWriterImpl;
+import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequestImpl;
+import org.apache.hadoop.hbase.regionserver.compactions.DefaultCompactor;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+@Category({ RegionServerTests.class, MediumTests.class })
+public class TestCompactorMemLeak {
+
+ private static final HBaseTestingUtility UTIL = new HBaseTestingUtility();
+ private static final Configuration CONF = UTIL.getConfiguration();
+ private static final AtomicBoolean IS_LAST_CELL_ON_HEAP = new AtomicBoolean(false);
+ private static final byte[] FAMILY = Bytes.toBytes("f");
+ private static final byte[] QUALIFIER = Bytes.toBytes("q");
+ private static final byte[] VALUE = Bytes.toBytes("value");
+
+ @Rule
+ public TestName name = new TestName();
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestCompactorMemLeak.class);
+
+ @BeforeClass
+ public static void setUp() throws Exception {
+ IS_LAST_CELL_ON_HEAP.set(false);
+ // Must use OFF-HEAP BucketCache here.
+ CONF.setFloat(HConstants.HFILE_BLOCK_CACHE_SIZE_KEY, 0.1f);
+ CONF.set(HConstants.BUCKET_CACHE_IOENGINE_KEY, "offheap");
+ // 32MB for BucketCache.
+ CONF.setFloat(HConstants.BUCKET_CACHE_SIZE_KEY, 32);
+ // Use the MyCompactor
+ CONF.set(DEFAULT_COMPACTOR_CLASS_KEY, MyCompactor.class.getName());
+ UTIL.startMiniCluster();
+ }
+
+ @AfterClass
+ public static void tearDown() throws Exception {
+ IS_LAST_CELL_ON_HEAP.set(false);
+ UTIL.shutdownMiniCluster();
+ }
+
+ private void assertMajorCompactionOK(TableName tableName) {
+ List<HRegion> regions = UTIL.getHBaseCluster().getRegionServerThreads().get(0).getRegionServer()
+ .getRegions(tableName);
+ Assert.assertEquals(regions.size(), 1);
+ HRegion region = regions.get(0);
+ Assert.assertEquals(region.getStores().size(), 1);
+ HStore store = region.getStore(FAMILY);
+ Assert.assertEquals(store.getStorefilesCount(), 1);
+ }
+
+ @Test
+ public void testMemLeak() throws IOException, InterruptedException {
+ TableName tableName = TableName.valueOf(name.getMethodName());
+ Table table = UTIL.createTable(tableName, FAMILY);
+
+ // Put and Flush #1
+ Put put = new Put(Bytes.toBytes("row1")).addColumn(FAMILY, QUALIFIER, VALUE);
+ table.put(put);
+ UTIL.getAdmin().flush(tableName);
+
+ // Put and Flush #2
+ put = new Put(Bytes.toBytes("row2")).addColumn(FAMILY, QUALIFIER, VALUE);
+ table.put(put);
+ UTIL.getAdmin().flush(tableName);
+
+ // Major compact
+ UTIL.getAdmin().majorCompact(tableName);
+ Thread.sleep(6000);
+ assertMajorCompactionOK(tableName);
+
+ // The last cell before Compactor#commitWriter must be an heap one.
+ Assert.assertTrue(IS_LAST_CELL_ON_HEAP.get());
+ }
+
+ public static class MyCompactor extends DefaultCompactor {
+
+ public MyCompactor(Configuration conf, HStore store) {
+ super(conf, store);
+ }
+
+ @Override
+ protected List<Path> commitWriter(StoreFileWriter writer, FileDetails fd,
+ CompactionRequestImpl request) throws IOException {
+ HFileWriterImpl writerImpl = (HFileWriterImpl) writer.writer;
+ Cell cell = writerImpl.getLastCell();
+ // The cell should be backend with an KeyOnlyKeyValue.
+ IS_LAST_CELL_ON_HEAP.set(cell instanceof KeyOnlyKeyValue);
+ return super.commitWriter(writer, fd, request);
+ }
+ }
+}