You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zh...@apache.org on 2018/06/05 10:16:22 UTC
[04/50] [abbrv] hbase git commit: HBASE-18116 Replication source
in-memory accounting should not include bulk transfer hfiles
HBASE-18116 Replication source in-memory accounting should not include bulk transfer hfiles
Signed-off-by: Andrew Purtell <ap...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/a11701ec
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/a11701ec
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/a11701ec
Branch: refs/heads/HBASE-19064
Commit: a11701ecc5094d0c8cd96a290d86e52df83a9707
Parents: 0968668
Author: Xu Cang <xc...@salesforce.com>
Authored: Thu May 31 20:00:04 2018 -0700
Committer: Andrew Purtell <ap...@apache.org>
Committed: Fri Jun 1 11:15:47 2018 -0700
----------------------------------------------------------------------
.../regionserver/ReplicationSource.java | 1 +
.../regionserver/ReplicationSourceShipper.java | 19 ++++++++++++++++-
.../ReplicationSourceWALReader.java | 22 +++++++++++++++-----
.../regionserver/TestGlobalThrottler.java | 20 ++++++++----------
4 files changed, 45 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/a11701ec/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
index 4051efe..d21d83c 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java
@@ -598,6 +598,7 @@ public class ReplicationSource implements ReplicationSourceInterface {
}
@Override
+ //offsets totalBufferUsed by deducting shipped batchSize.
public void postShipEdits(List<Entry> entries, int batchSize) {
if (throttler.isEnabled()) {
throttler.addPushSize(batchSize);
http://git-wip-us.apache.org/repos/asf/hbase/blob/a11701ec/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
index 11fd660..123ecbe 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceShipper.java
@@ -124,6 +124,18 @@ public class ReplicationSourceShipper extends Thread {
}
/**
+ * get batchEntry size excludes bulk load file sizes.
+ * Uses ReplicationSourceWALReader's static method.
+ */
+ private int getBatchEntrySizeExcludeBulkLoad(WALEntryBatch entryBatch) {
+ int totalSize = 0;
+ for(Entry entry : entryBatch.getWalEntries()) {
+ totalSize += entryReader.getEntrySizeExcludeBulkLoad(entry);
+ }
+ return totalSize;
+ }
+
+ /**
* Do the shipping logic
*/
private void shipEdits(WALEntryBatch entryBatch) {
@@ -139,6 +151,7 @@ public class ReplicationSourceShipper extends Thread {
return;
}
int currentSize = (int) entryBatch.getHeapSize();
+ int sizeExcludeBulkLoad = getBatchEntrySizeExcludeBulkLoad(entryBatch);
while (isActive()) {
try {
try {
@@ -175,7 +188,11 @@ public class ReplicationSourceShipper extends Thread {
// Log and clean up WAL logs
updateLogPosition(entryBatch);
- source.postShipEdits(entries, currentSize);
+ //offsets totalBufferUsed by deducting shipped batchSize (excludes bulk load size)
+ //this sizeExcludeBulkLoad has to use same calculation that when calling
+ //acquireBufferQuota() in ReplicatinoSourceWALReader because they maintain
+ //same variable: totalBufferUsed
+ source.postShipEdits(entries, sizeExcludeBulkLoad);
// FIXME check relationship between wal group and overall
source.getSourceMetrics().shipBatch(entryBatch.getNbOperations(), currentSize,
entryBatch.getNbHFiles());
http://git-wip-us.apache.org/repos/asf/hbase/blob/a11701ec/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
index 64fd48d..f685a9b 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceWALReader.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.hadoop.hbase.wal.WALKey;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.yetus.audience.InterfaceStability;
import org.slf4j.Logger;
@@ -168,10 +169,12 @@ class ReplicationSourceWALReader extends Thread {
if (edit == null || edit.isEmpty()) {
return false;
}
- long entrySize = getEntrySize(entry);
+ long entrySize = getEntrySizeIncludeBulkLoad(entry);
+ long entrySizeExcludeBulkLoad = getEntrySizeExcludeBulkLoad(entry);
batch.addEntry(entry);
updateBatchStats(batch, entry, entrySize);
- boolean totalBufferTooLarge = acquireBufferQuota(entrySize);
+ boolean totalBufferTooLarge = acquireBufferQuota(entrySizeExcludeBulkLoad);
+
// Stop if too many entries or too big
return totalBufferTooLarge || batch.getHeapSize() >= replicationBatchSizeCapacity ||
batch.getNbEntries() >= replicationBatchCountCapacity;
@@ -296,11 +299,20 @@ class ReplicationSourceWALReader extends Thread {
return entryBatchQueue.take();
}
- private long getEntrySize(Entry entry) {
+ private long getEntrySizeIncludeBulkLoad(Entry entry) {
WALEdit edit = entry.getEdit();
- return edit.heapSize() + calculateTotalSizeOfStoreFiles(edit);
+ WALKey key = entry.getKey();
+ return edit.heapSize() + sizeOfStoreFilesIncludeBulkLoad(edit) +
+ key.estimatedSerializedSizeOf();
}
+ public static long getEntrySizeExcludeBulkLoad(Entry entry) {
+ WALEdit edit = entry.getEdit();
+ WALKey key = entry.getKey();
+ return edit.heapSize() + key.estimatedSerializedSizeOf();
+ }
+
+
private void updateBatchStats(WALEntryBatch batch, Entry entry, long entrySize) {
WALEdit edit = entry.getEdit();
batch.incrementHeapSize(entrySize);
@@ -353,7 +365,7 @@ class ReplicationSourceWALReader extends Thread {
* @param edit edit to count row keys from
* @return the total size of the store files
*/
- private int calculateTotalSizeOfStoreFiles(WALEdit edit) {
+ private int sizeOfStoreFilesIncludeBulkLoad(WALEdit edit) {
List<Cell> cells = edit.getCells();
int totalStoreFilesSize = 0;
http://git-wip-us.apache.org/repos/asf/hbase/blob/a11701ec/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
index d3b4e8e..fecce02 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/replication/regionserver/TestGlobalThrottler.java
@@ -64,6 +64,8 @@ public class TestGlobalThrottler {
HBaseClassTestRule.forClass(TestGlobalThrottler.class);
private static final Logger LOG = LoggerFactory.getLogger(TestGlobalThrottler.class);
+ private static final int REPLICATION_SOURCE_QUOTA = 200;
+ private static int numOfPeer = 0;
private static Configuration conf1;
private static Configuration conf2;
@@ -84,7 +86,7 @@ public class TestGlobalThrottler {
conf1.set(HConstants.ZOOKEEPER_ZNODE_PARENT, "/1");
conf1.setLong("replication.source.sleepforretries", 100);
// Each WAL is about 120 bytes
- conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, 200);
+ conf1.setInt(HConstants.REPLICATION_SOURCE_TOTAL_BUFFER_KEY, REPLICATION_SOURCE_QUOTA);
conf1.setLong("replication.source.per.peer.node.bandwidth", 100L);
utility1 = new HBaseTestingUtility(conf1);
@@ -109,6 +111,7 @@ public class TestGlobalThrottler {
admin1.addPeer("peer1", rpc, null);
admin1.addPeer("peer2", rpc, null);
admin1.addPeer("peer3", rpc, null);
+ numOfPeer = admin1.getPeersCount();
}
@AfterClass
@@ -140,7 +143,10 @@ public class TestGlobalThrottler {
if (size > 0) {
testQuotaNonZero = true;
}
- if (size > 600) {
+ //the reason here doing "numOfPeer + 1" is because by using method addEntryToBatch(), even the
+ // batch size (after added last entry) exceeds quota, it still keeps the last one in the batch
+ // so total used buffer size can be one "replication.total.buffer.quota" larger than expected
+ if (size > REPLICATION_SOURCE_QUOTA * (numOfPeer + 1)) {
// We read logs first then check throttler, so if the buffer quota limiter doesn't
// take effect, it will push many logs and exceed the quota.
testQuotaPass = false;
@@ -181,13 +187,5 @@ public class TestGlobalThrottler {
Assert.assertTrue(testQuotaNonZero);
}
- private List<Integer> getRowNumbers(List<Cell> cells) {
- List<Integer> listOfRowNumbers = new ArrayList<>(cells.size());
- for (Cell c : cells) {
- listOfRowNumbers.add(Integer.parseInt(Bytes
- .toString(c.getRowArray(), c.getRowOffset() + ROW.length,
- c.getRowLength() - ROW.length)));
- }
- return listOfRowNumbers;
- }
+
}