HBASE-18451 PeriodicMemstoreFlusher should inspect the queue before adding a delayed flush request
Signed-off-by: Andrew Purtell <apurt...@apache.org> Project: http://git-wip-us.apache.org/repos/asf/hbase/repo Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/709feeae Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/709feeae Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/709feeae Branch: refs/heads/branch-1.4 Commit: 709feeae97f1d5575f67c97fdb0d09866a0bee61 Parents: 8a7ff12 Author: xcang <xc...@salesforce.com> Authored: Sun Sep 23 23:42:57 2018 -0700 Committer: Andrew Purtell <apurt...@apache.org> Committed: Fri Sep 28 11:50:58 2018 -0700 ---------------------------------------------------------------------- .../hadoop/hbase/regionserver/FlushRequester.java | 6 ++++-- .../hadoop/hbase/regionserver/HRegionServer.java | 13 +++++++------ .../hadoop/hbase/regionserver/MemStoreFlusher.java | 12 ++++++++---- .../hbase/regionserver/TestHeapMemoryManager.java | 7 ++++--- .../hadoop/hbase/regionserver/wal/TestWALReplay.java | 8 ++++---- 5 files changed, 27 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hbase/blob/709feeae/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/FlushRequester.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/FlushRequester.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/FlushRequester.java index c7e155a..243546c 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/FlushRequester.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/FlushRequester.java @@ -32,8 +32,9 @@ public interface FlushRequester { * @param region the Region requesting the cache flush * @param forceFlushAllStores whether we want to flush all stores. e.g., when request from log * rolling. + * @return true if our region is added into the queue, false otherwise */ - void requestFlush(Region region, boolean forceFlushAllStores); + boolean requestFlush(Region region, boolean forceFlushAllStores); /** * Tell the listener the cache needs to be flushed after a delay @@ -42,8 +43,9 @@ public interface FlushRequester { * @param delay after how much time should the flush happen * @param forceFlushAllStores whether we want to flush all stores. e.g., when request from log * rolling. + * @return true if our region is added into the queue, false otherwise */ - void requestDelayedFlush(Region region, long delay, boolean forceFlushAllStores); + boolean requestDelayedFlush(Region region, long delay, boolean forceFlushAllStores); /** * Register a FlushRequestListener http://git-wip-us.apache.org/repos/asf/hbase/blob/709feeae/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java index 962a0e9..dbcd085 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java @@ -38,6 +38,7 @@ import java.net.BindException; import java.net.InetAddress; import java.net.InetSocketAddress; import java.security.PrivilegedExceptionAction; +import java.text.MessageFormat; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; @@ -1661,18 +1662,18 @@ public class HRegionServer extends HasThread implements final StringBuffer whyFlush = new StringBuffer(); for (Region r : this.server.onlineRegions.values()) { if (r == null) continue; - if (((HRegion)r).shouldFlush(whyFlush)) { + if (((HRegion) r).shouldFlush(whyFlush)) { FlushRequester requester = server.getFlushRequester(); if (requester != null) { long randomDelay = (long) RandomUtils.nextInt(RANGE_OF_DELAY) + MIN_DELAY_TIME; - LOG.info(getName() + " requesting flush of " + - r.getRegionInfo().getRegionNameAsString() + " because " + - whyFlush.toString() + - " after random delay " + randomDelay + "ms"); //Throttle the flushes by putting a delay. If we don't throttle, and there //is a balanced write-load on the regions in a table, we might end up //overwhelming the filesystem with too many flushes at once. - requester.requestDelayedFlush(r, randomDelay, false); + if (requester.requestDelayedFlush(r, randomDelay, false)) { + LOG.info(MessageFormat.format("{0} requesting flush of {1} because {2} " + + "after random delay {3} ms", getName(), + r.getRegionInfo().getRegionNameAsString(), whyFlush.toString(), randomDelay)); + } } } } http://git-wip-us.apache.org/repos/asf/hbase/blob/709feeae/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java index 2da782e..9545014 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MemStoreFlusher.java @@ -361,8 +361,7 @@ class MemStoreFlusher implements FlushRequester { } @Override - public void requestFlush(Region r, boolean forceFlushAllStores) { - ((HRegion)r).incrementFlushesQueuedCount(); + public boolean requestFlush(Region r, boolean forceFlushAllStores) { synchronized (regionsInQueue) { if (!regionsInQueue.containsKey(r)) { // This entry has no delay so it will be added at the top of the flush @@ -370,13 +369,15 @@ class MemStoreFlusher implements FlushRequester { FlushRegionEntry fqe = new FlushRegionEntry(r, forceFlushAllStores); this.regionsInQueue.put(r, fqe); this.flushQueue.add(fqe); + ((HRegion)r).incrementFlushesQueuedCount(); + return true; } + return false; } } @Override - public void requestDelayedFlush(Region r, long delay, boolean forceFlushAllStores) { - ((HRegion)r).incrementFlushesQueuedCount(); + public boolean requestDelayedFlush(Region r, long delay, boolean forceFlushAllStores) { synchronized (regionsInQueue) { if (!regionsInQueue.containsKey(r)) { // This entry has some delay @@ -384,7 +385,10 @@ class MemStoreFlusher implements FlushRequester { fqe.requeue(delay); this.regionsInQueue.put(r, fqe); this.flushQueue.add(fqe); + ((HRegion)r).incrementFlushesQueuedCount(); + return true; } + return false; } } http://git-wip-us.apache.org/repos/asf/hbase/blob/709feeae/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java index 00ec9c8..1053ab7 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHeapMemoryManager.java @@ -648,13 +648,14 @@ public class TestHeapMemoryManager { } @Override - public void requestFlush(Region region, boolean forceFlushAllStores) { + public boolean requestFlush(Region region, boolean forceFlushAllStores) { this.listener.flushRequested(flushType, region); + return true; } @Override - public void requestDelayedFlush(Region region, long delay, boolean forceFlushAllStores) { - + public boolean requestDelayedFlush(Region region, long delay, boolean forceFlushAllStores) { + return true; } @Override http://git-wip-us.apache.org/repos/asf/hbase/blob/709feeae/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java ---------------------------------------------------------------------- diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java index 6d57e7b..2ea2629 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplay.java @@ -1119,18 +1119,18 @@ public class TestWALReplay { class TestFlusher implements FlushRequester { @Override - public void requestFlush(Region region, boolean force) { + public boolean requestFlush(Region region, boolean force) { try { region.flush(force); + return true; } catch (IOException e) { throw new RuntimeException("Exception flushing", e); } } @Override - public void requestDelayedFlush(Region region, long when, boolean forceFlushAllStores) { - // TODO Auto-generated method stub - + public boolean requestDelayedFlush(Region region, long when, boolean forceFlushAllStores) { + return true; } @Override