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 2021/11/18 15:26:23 UTC
[hbase] 05/06: HBASE-26448 Make sure we do not flush a region too frequently (#3847)
This is an automated email from the ASF dual-hosted git repository.
zhangduo pushed a commit to branch HBASE-26233
in repository https://gitbox.apache.org/repos/asf/hbase.git
commit 42ec40c8667a9d04aef99dd035ed1aebfd964fa7
Author: Duo Zhang <zh...@apache.org>
AuthorDate: Tue Nov 16 00:06:54 2021 +0800
HBASE-26448 Make sure we do not flush a region too frequently (#3847)
Signed-off-by: Xiaolin Ha <ha...@apache.org>
---
.../RegionReplicationFlushRequester.java | 142 +++++++++++++++++++++
.../regionreplication/RegionReplicationSink.java | 24 ++--
.../TestRegionReplicationFlushRequester.java | 84 ++++++++++++
3 files changed, 239 insertions(+), 11 deletions(-)
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationFlushRequester.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationFlushRequester.java
new file mode 100644
index 0000000..960f57e
--- /dev/null
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationFlushRequester.java
@@ -0,0 +1,142 @@
+/**
+ * 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.regionreplication;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.util.Threads;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hbase.thirdparty.io.netty.util.HashedWheelTimer;
+import org.apache.hbase.thirdparty.io.netty.util.Timeout;
+
+/**
+ * A helper class for requesting flush on a given region.
+ * <p/>
+ * In general, we do not want to trigger flush too frequently for a region, so here we will add
+ * something like a rate control, i.e, the interval of the two flush request should not be too
+ * small.
+ */
+@InterfaceAudience.Private
+class RegionReplicationFlushRequester {
+
+ /**
+ * The timer for executing delayed flush request task.
+ * <p/>
+ * It will be shared across all the instances {@link RegionReplicationFlushRequester}. Created on
+ * demand to save one extra thread as not every user uses region replication.
+ */
+ private static volatile HashedWheelTimer TIMER;
+
+ /**
+ * The minimum interval between two flush requests
+ */
+ public static final String MIN_INTERVAL_SECS =
+ "hbase.region.read-replica.sink.flush.min-interval.secs";
+
+ public static final int MIN_INTERVAL_SECS_DEFAULT = 30;
+
+ private final Runnable flushRequester;
+
+ private final long minIntervalSecs;
+
+ private long lastRequestNanos;
+
+ private long pendingFlushRequestSequenceId;
+
+ private long lastFlushedSequenceId;
+
+ private Timeout pendingFlushRequest;
+
+ RegionReplicationFlushRequester(Configuration conf, Runnable flushRequester) {
+ this.flushRequester = flushRequester;
+ this.minIntervalSecs = conf.getInt(MIN_INTERVAL_SECS, MIN_INTERVAL_SECS_DEFAULT);
+ }
+
+ private static HashedWheelTimer getTimer() {
+ HashedWheelTimer timer = TIMER;
+ if (timer != null) {
+ return timer;
+ }
+ synchronized (RegionReplicationFlushRequester.class) {
+ timer = TIMER;
+ if (timer != null) {
+ return timer;
+ }
+ timer = new HashedWheelTimer(
+ new ThreadFactoryBuilder().setNameFormat("RegionReplicationFlushRequester-Timer-pool-%d")
+ .setDaemon(true).setUncaughtExceptionHandler(Threads.LOGGING_EXCEPTION_HANDLER).build(),
+ 500, TimeUnit.MILLISECONDS);
+ TIMER = timer;
+ }
+ return timer;
+ }
+
+ private void request() {
+ flushRequester.run();
+ lastRequestNanos = System.nanoTime();
+ }
+
+ private synchronized void flush(Timeout timeout) {
+ pendingFlushRequest = null;
+ if (pendingFlushRequestSequenceId >= lastFlushedSequenceId) {
+ request();
+ }
+ }
+
+ /**
+ * Request a flush for the given region.
+ * <p/>
+ * The sequence id of the edit which we fail to replicate. A flush must happen after this sequence
+ * id to recover the failure.
+ */
+ synchronized void requestFlush(long sequenceId) {
+ // if there is already a flush task, just reuse it.
+ if (pendingFlushRequest != null) {
+ pendingFlushRequestSequenceId = Math.max(sequenceId, pendingFlushRequestSequenceId);
+ return;
+ }
+ // check last flush time
+ long elapsedSecs = TimeUnit.NANOSECONDS.toSeconds(System.nanoTime() - lastRequestNanos);
+ if (elapsedSecs >= minIntervalSecs) {
+ request();
+ return;
+ }
+ // schedule a timer task
+ HashedWheelTimer timer = getTimer();
+ pendingFlushRequest =
+ timer.newTimeout(this::flush, minIntervalSecs - elapsedSecs, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Record that we have already finished a flush with the given {@code sequenceId}.
+ * <p/>
+ * We can cancel the pending flush request if the failed sequence id is less than the given
+ * {@code sequenceId}.
+ */
+ synchronized void recordFlush(long sequenceId) {
+ this.lastFlushedSequenceId = sequenceId;
+ // cancel the pending flush request if it is necessary, i.e, we have already finished a flush
+ // with higher sequence id.
+ if (sequenceId > pendingFlushRequestSequenceId && pendingFlushRequest != null) {
+ pendingFlushRequest.cancel();
+ pendingFlushRequest = null;
+ }
+ }
+}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationSink.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationSink.java
index 68aa508..9095870 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationSink.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/regionreplication/RegionReplicationSink.java
@@ -118,7 +118,7 @@ public class RegionReplicationSink {
private final RegionReplicationBufferManager manager;
- private final Runnable flushRequester;
+ private final RegionReplicationFlushRequester flushRequester;
private final AsyncClusterConnection conn;
@@ -136,7 +136,7 @@ public class RegionReplicationSink {
private volatile long pendingSize;
- private long lastFlushSequenceNumber;
+ private long lastFlushedSequenceId;
private boolean sending;
@@ -154,7 +154,7 @@ public class RegionReplicationSink {
this.primary = primary;
this.tableDesc = td;
this.manager = manager;
- this.flushRequester = flushRequester;
+ this.flushRequester = new RegionReplicationFlushRequester(conf, flushRequester);
this.conn = conn;
this.retries = conf.getInt(RETRIES_NUMBER, RETRIES_NUMBER_DEFAULT);
this.rpcTimeoutNs =
@@ -178,19 +178,19 @@ public class RegionReplicationSink {
Integer replicaId = entry.getKey();
Throwable error = entry.getValue().getValue();
if (error != null) {
- if (maxSequenceId > lastFlushSequenceNumber) {
+ if (maxSequenceId > lastFlushedSequenceId) {
LOG.warn(
"Failed to replicate to secondary replica {} for {}, since the max sequence" +
" id of sunk entris is {}, which is greater than the last flush SN {}," +
" we will stop replicating for a while and trigger a flush",
- replicaId, primary, maxSequenceId, lastFlushSequenceNumber, error);
+ replicaId, primary, maxSequenceId, lastFlushedSequenceId, error);
failed.add(replicaId);
} else {
LOG.warn(
"Failed to replicate to secondary replica {} for {}, since the max sequence" +
" id of sunk entris is {}, which is less than or equal to the last flush SN {}," +
" we will not stop replicating",
- replicaId, primary, maxSequenceId, lastFlushSequenceNumber, error);
+ replicaId, primary, maxSequenceId, lastFlushedSequenceId, error);
}
}
}
@@ -198,7 +198,7 @@ public class RegionReplicationSink {
pendingSize -= toReleaseSize;
if (!failed.isEmpty()) {
failedReplicas.addAll(failed);
- flushRequester.run();
+ flushRequester.requestFlush(maxSequenceId);
}
sending = false;
if (stopping) {
@@ -296,6 +296,7 @@ public class RegionReplicationSink {
continue;
}
if (flushDesc != null && flushAllStores(flushDesc)) {
+ long flushedSequenceId = flushDesc.getFlushSequenceNumber();
int toClearCount = 0;
long toClearSize = 0;
for (;;) {
@@ -303,7 +304,7 @@ public class RegionReplicationSink {
if (e == null) {
break;
}
- if (e.key.getSequenceId() < flushDesc.getFlushSequenceNumber()) {
+ if (e.key.getSequenceId() < flushedSequenceId) {
entries.poll();
toClearCount++;
toClearSize += e.size;
@@ -311,13 +312,14 @@ public class RegionReplicationSink {
break;
}
}
- lastFlushSequenceNumber = flushDesc.getFlushSequenceNumber();
+ lastFlushedSequenceId = flushedSequenceId;
failedReplicas.clear();
LOG.debug(
"Got a flush all request with sequence id {}, clear failed replicas {}" +
" and {} pending entries with size {}",
- flushDesc.getFlushSequenceNumber(), failedReplicas, toClearCount,
+ flushedSequenceId, failedReplicas, toClearCount,
StringUtils.TraditionalBinaryPrefix.long2String(toClearSize, "", 1));
+ flushRequester.recordFlush(flushedSequenceId);
}
}
}
@@ -340,7 +342,7 @@ public class RegionReplicationSink {
for (int replicaId = 1; replicaId < regionReplication; replicaId++) {
failedReplicas.add(replicaId);
}
- flushRequester.run();
+ flushRequester.requestFlush(entry.key.getSequenceId());
}
}
}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/regionreplication/TestRegionReplicationFlushRequester.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/regionreplication/TestRegionReplicationFlushRequester.java
new file mode 100644
index 0000000..abe5aa1
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/regionreplication/TestRegionReplicationFlushRequester.java
@@ -0,0 +1,84 @@
+/**
+ * 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.regionreplication;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category({ RegionServerTests.class, MediumTests.class })
+public class TestRegionReplicationFlushRequester {
+
+ @ClassRule
+ public static final HBaseClassTestRule CLASS_RULE =
+ HBaseClassTestRule.forClass(TestRegionReplicationFlushRequester.class);
+
+ private Configuration conf;
+
+ private Runnable requester;
+
+ private RegionReplicationFlushRequester flushRequester;
+
+ @Before
+ public void setUp() {
+ conf = HBaseConfiguration.create();
+ conf.setInt(RegionReplicationFlushRequester.MIN_INTERVAL_SECS, 1);
+ requester = mock(Runnable.class);
+ flushRequester = new RegionReplicationFlushRequester(conf, requester);
+ }
+
+ @Test
+ public void testRequest() throws InterruptedException {
+ // should call request directly
+ flushRequester.requestFlush(100L);
+ verify(requester, times(1)).run();
+
+ // should not call request directly, since the min interval is 1 second
+ flushRequester.requestFlush(200L);
+ verify(requester, times(1)).run();
+ Thread.sleep(2000);
+ verify(requester, times(2)).run();
+
+ // should call request directly because we have already elapsed more than 1 second
+ Thread.sleep(2000);
+ flushRequester.requestFlush(300L);
+ verify(requester, times(3)).run();
+ }
+
+ @Test
+ public void testCancelFlushRequest() throws InterruptedException {
+ flushRequester.requestFlush(100L);
+ flushRequester.requestFlush(200L);
+ verify(requester, times(1)).run();
+
+ // the pending flush request should be canceled
+ flushRequester.recordFlush(300L);
+ Thread.sleep(2000);
+ verify(requester, times(1)).run();
+ }
+}